   defp enqueue(s, what) do
     update_in s.queue, &:queue.in(what, &1)
   end
+
+  defp dequeue_and_parse_resp(s, {{:value, :fetch_schema}, new_queue}, data) do
+    sid = s.session_id
+
+    case Protocol.parse_resp(:record_load, data, s.schema) do
+      :incomplete ->
+        %{s | tail: data}
+      {:error, %Error{}, _} ->
+        raise "couldn't fetch schema"
+      {:ok, ^sid, [resp], rest} ->
+        %{s | schema: parse_schema(resp), tail: rest, queue: new_queue}
+    end
+  end
+
+  defp dequeue_and_parse_resp(s, {{:value, {from, op_name}}, new_queue}, data) do
+    sid = s.session_id
+
+    case Protocol.parse_resp(op_name, data, s.schema) do
+      :incomplete ->
+        %{s | tail: data}
+      {:unknown_property_id, rest} ->
+        Connection.reply(from, {:error, :unknown_property_id})
+        %{s | tail: rest}
+      {:error, error, rest} ->
+        Connection.reply(from, {:error, error})
+        %{s | tail: rest, queue: new_queue}
+      {:ok, ^sid, resp, rest} ->
+        Connection.reply(from, {:ok, resp})
+        %{s | tail: rest, queue: new_queue}
+    end
+  end
 end


  def handle_info({:tcp, socket, msg}, %{session_id: sid, socket: socket} = s) do
    :inet.setopts(socket, active: :once)
    data = s.tail <> msg

    s =
      case :queue.out(s.queue) do
        {{:value, :fetch_schema}, new_queue} ->
          case Protocol.parse_resp(:record_load, data, s.schema) do
            :incomplete ->
              %{s | tail: data}
            {:error, %Error{}, _} ->
              raise "couldn't fetch schema"
            {:ok, ^sid, [resp], rest} ->
              %{s | schema: parse_schema(resp), tail: rest, queue: new_queue}
          end
        {{:value, {from, op_name}}, new_queue} ->
          case Protocol.parse_resp(op_name, data, s.schema) do
            :incomplete ->
              %{s | tail: data}
            {:unknown_property_id, rest} ->
              Connection.reply(from, {:error, :unknown_property_id})
              %{s | tail: rest}
            {:error, error, rest} ->
              Connection.reply(from, {:error, error})
              %{s | tail: rest, queue: new_queue}
            {:ok, ^sid, resp, rest} ->
              Connection.reply(from, {:ok, resp})
              %{s | tail: rest, queue: new_queue}
          end
      end

    {:noreply, s}
  end



  @doc false
  def disconnect(error, s) do
    # We only care about {from, _} tuples, ignoring queued stuff like
    # :fetch_schema.
    for {from, _operation} <- :queue.to_list(s.queue) do
      Connection.reply(from, error)
    end

