anchor.diff:      {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}) ->
  +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
  
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: 
  %% PIPE Handle response from mapred_stream
  handle_info(#pipe_eoi{ref=ReqId},
              State=#state{req=#rpbmapredreq{}, req_ctx=ReqId}) ->
      NewState = send_msg(#rpbmapredresp{done = 1}, State),
      {noreply, NewState#state{req = undefined, req_ctx = undefined}};
  
  handle_info(#pipe_result{ref=ReqId, from=PhaseId, result=Res},
              State=#state{req=#rpbmapredreq{content_type = ContentType}, 
                           req_ctx=ReqId}) ->
      case encode_mapred_phase([Res], ContentType) of
          {error, Reason} ->
              NewState = send_error("~p", [Reason], State),
              {noreply, NewState#state{req = undefined, req_ctx = undefined}};
          Response ->
              {noreply, send_msg(#rpbmapredresp{phase=PhaseId, 
  