anchor.diff:              inet:setopts(Socket, [{active, once}])
       end,
       {noreply, NewState};
  +handle_info({tcp, _Sock, _Data}, State) ->
  +    %% req =/= undefined: received a new request while another was in
  +    %% progress -> Error
  +    lager:error("Received a new PB socket request"
  +                " while another was in progress"),
  +    {stop, normal, State};
   
   %% Handle responses from stream_list_keys 
   handle_info({ReqId, done},
  
positive.code:                 ok ->
                      Ref = (Pipe#pipe.sink)#fitting.ref,
                      %% TODO: timeout
                      {pause, State#state{req=Req, req_ctx=Ref}};
                  Error ->
                      %% even if the structure looks right, it might name
                      %% a non-existent key filter or something
  
negative.code:     {stop, normal, State};
  handle_info({tcp_error, Socket, _Reason}, State=#state{sock=Socket}) ->
      {stop, normal, State};
  handle_info({tcp, _Sock, Data}, State=#state{sock=Socket, req=undefined}) ->
      [MsgCode|MsgData] = Data,
      Msg = riakc_pb:decode(MsgCode, MsgData),
      case process_message(Msg, State) of
  