anchor.diff:  handle_info({ReqId, {keys, Keys}}, State=#state{req=#rpblistkeysreq{}, req_ctx=ReqId}) ->
       {noreply, send_msg(#rpblistkeysresp{keys = Keys}, State)};
   
  -%% Handle response from mapred_stream/mapred_bucket_stream
  -handle_info({flow_results, ReqId, done},
  -            State=#state{sock = Socket, req=#rpbmapredreq{}, req_ctx=ReqId}) ->
  -    NewState = send_msg(#rpbmapredresp{done = 1}, State),
  -    inet:setopts(Socket, [{active, once}]),
  -    {noreply, NewState#state{req = undefined, req_ctx = undefined}};
  -
  -handle_info({flow_results, ReqId, {error, Reason}},
  -            State=#state{sock = Socket, req=#rpbmapredreq{}, req_ctx=ReqId}) ->
  -    NewState = send_error("~p", [Reason], State),
  -    inet:setopts(Socket, [{active, once}]),
  -    {noreply, NewState#state{req = undefined, req_ctx = undefined}};
  -
  -handle_info({flow_results, PhaseId, ReqId, Res},
  -            State=#state{req=#rpbmapredreq{}, req_ctx=ReqId}) ->
  -    {noreply, send_msg(#rpbmapredresp{phase=PhaseId, data = riakc_pb:pbify_rpbterm(Res)}, State)};
  -
  -handle_info({flow_error, ReqId, Error},
  -            State=#state{sock = Socket, req=#rpbmapredreq{}, req_ctx=ReqId}) ->
  -    NewState = send_error("~p", [Error], State),
  -    inet:setopts(Socket, [{active, once}]),
  -    {noreply, NewState#state{req = undefined, req_ctx = undefined}};
  -
   handle_info(_, State) -> % Ignore any late replies from gen_servers/messages from fsms
       {noreply, State}.
   
  
positive.code:     case C:stream_list_keys(B) of
          {ok, ReqId} ->
              {pause, State#state{req = Req, req_ctx = ReqId}}
      end;
  
  %% Start map/reduce job - results will be processed in handle_info
  process_message(#rpbmapredreq{input_bucket=B, input_keys=PbKeys, phases=PbQuery}=Req, 
                  #state{client=C} = State) ->
      case riakc_pb:erlify_mapred_query(PbQuery) of
          {error, Reason} ->
              send_error("~p", [Reason], State);
  
          {ok, Query} ->
              if %% Check we have B or PbKeys
                  (B =/= undefined andalso PbKeys =:= undefined) ->
                      {ok, ReqId} = C:mapred_bucket_stream(B, Query, self(), ?DEFAULT_TIMEOUT),
                      {pause, State#state{req = Req, req_ctx = ReqId}};
                  (B =:= undefined andalso PbKeys =/= undefined) -> %
                      Inputs = [riakc_pb:erlify_mapred_input(PbKey) || PbKey <- PbKeys],
                      {ok, {ReqId, FSM}} = C:mapred_stream(Query, self(), ?DEFAULT_TIMEOUT),
                      luke_flow:add_inputs(FSM, Inputs),
                      luke_flow:finish_inputs(FSM),
                      {pause, State#state{req = Req, req_ctx = ReqId}};
                  true ->
                      send_error("map/reduce takes either an input_bucket or input_keys, not both", 
                                 [], State)
              end
      end.
  
  %% @private
  
negative.code: %% callbacks are waiting for it.
  %%
  -spec process_message(msg(), #state{}) ->  #state{} | {pause, #state{}}.
  process_message(rpbpingreq, State) ->
      send_msg(rpbpingresp, State);
  
  