anchor.diff:              send_error("~p", [Reason], State)
       end;
   
  -process_message(#rpbgetbucketpropsreq{bucket=B, names = Names}, 
  -                #state{client=C} = State) ->
  -    Props = C:get_bucket(B),
  -    PbProps = riakc_pb:pbify_bucket_props(filter_props(Names, Props)),
  -    Resp = #rpbgetbucketpropsresp{properties = PbProps},
  -    send_msg(Resp, State);
  -
  -process_message(#rpbsetbucketpropsreq{bucket=B, properties = RpbTerm}, 
  -                #state{client=C} = State) ->
  -    ErlProps = riakc_pb:erlify_bucket_props(RpbTerm),
  -    C:set_bucket(B, ErlProps),
  -    send_msg(rpbsetbucketpropsresp, State);
  -
   process_message(rpblistbucketsreq, 
                   #state{client=C} = State) ->
       case C:list_buckets() of
  
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: -type msg() ::  atom() | tuple().
  
  -record(state, {sock,      % protocol buffers socket
                  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)
  