anchor.diff:      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
  
positive.code: -type msg() ::  atom() | tuple().
  
  -record(state, {sock,      % protocol buffers socket
                  hello,     % hello message from client
                  client,    % local client
                  req,       % current request (for multi-message requests like list keys)
                  req_ctx}). % context to go along with request (partial results, request ids etc)
  
negative.code: 
  init([Socket]) -> 
      inet:setopts(Socket, [{active, once}, {packet, 4}, {header, 1}]),
      {ok, C} = riak:local_client(),
      {ok, #state{sock = Socket, client = C}}.
  
  handle_call(_Request, _From, State) ->
      {reply, not_implemented, State}.
  