anchor.diff:      {ok, ReqId} = mapred_stream(Pid, Inputs, Query, self(), Timeout),
       wait_for_mapred(ReqId, Timeout).
   
  +%% @spec mapred_stream(Pid :: pid(),
  +%%                     Query :: [riak_kv_mapred_query:mapred_queryterm()],
  +%%                     ClientPid :: pid()) ->
  +%%       {ok, {ReqId :: term(), MR_FSM_PID :: pid()}} |
  +%%       {error, Err :: term()}
  +%% @doc Perform a streaming map/reduce job across the cluster.
  +%%      See the map/reduce documentation for explanation of behavior.
  +%% mapred_stream(Pid, Query, ClientPid) ->
  +%%     mapred_stream(Pid, Query, ClientPid,?DEFAULT_TIMEOUT).
  +
  +%% @spec mapred_stream(Pid :: pid(),
  +%%                     Query :: [riak_kv_mapred_query:mapred_queryterm()],
  +%%                     ClientPid :: pid(),
  +%%                     TimeoutMillisecs :: integer() | 'infinity') ->
  +%%       {ok, {ReqId :: term(), MR_FSM_PID :: pid()}} |
  +%%       {error, Err :: term()}
  +%% @doc Perform a streaming map/reduce job with a timeout across the cluster.
  +%%      See the map/reduce documentation for explanation of behavior.
   mapred_stream(Pid, Inputs, Query, ClientPid, _Timeout) ->
       ReqMsg = #rpbmapredreq{input_keys = [riakc_pb:pbify_mapred_input(I) || I <- Inputs],
                              phases = riakc_pb:pbify_mapred_query(Query)},
       ReqId = mk_reqid(),
       gen_server:call(Pid, {req, ReqMsg, {ReqId, ClientPid}}).
       
  -
  -%% %% @spec mapred_stream(Query :: [riak_kv_mapred_query:mapred_queryterm()],
  -%% %%                     ClientPid :: pid()) ->
  -%% %%       {ok, {ReqId :: term(), MR_FSM_PID :: pid()}} |
  -%% %%       {error, {bad_qterm, riak_kv_mapred_query:mapred_queryterm()}} |
  -%% %%       {error, Err :: term()}
  -%% %% @doc Perform a streaming map/reduce job across the cluster.
  -%% %%      See the map/reduce documentation for explanation of behavior.
  -%% mapred_stream(Pid, Query, ClientPid) ->
  -%%     mapred_stream(Pid, Query, ClientPid,?DEFAULT_TIMEOUT).
  -
  -%% %% @spec mapred_stream(Query :: [riak_kv_mapred_query:mapred_queryterm()],
  -%% %%                     ClientPid :: pid(),
  -%% %%                     TimeoutMillisecs :: integer() | 'infinity') ->
  -%% %%       {ok, {ReqId :: term(), MR_FSM_PID :: pid()}} |
  -%% %%       {error, {bad_qterm, riak_kv_mapred_query:mapred_queryterm()}} |
  -%% %%       {error, Err :: term()}
  -%% %% @doc Perform a streaming map/reduce job across the cluster.
  -%% %%      See the map/reduce documentation for explanation of behavior.
  -%% mapred_stream(Pid, Query, ClientPid, Timeout) ->
  -%%     ReqMsg = #rpblistkeysreq{bucket = Bucket},
  -%%     ReqId = mk_reqid(),
  -%%     gen_server:call(Pid, {req, ReqMsg, {ReqId, self()}}).
  -
  -%% mapred_bucket_stream(Pid, Bucket, Query, ClientPid) ->
  -%%     mapred_bucket_stream(Bucket, Query, ClientPid, ?DEFAULT_TIMEOUT).
  -
  -%% mapred_bucket_stream(Pid, Bucket, Query, ClientPid, Timeout) ->
  -%%     mapred_bucket_stream(Bucket, Query, ClientPid, Timeout, ?DEFAULT_ERRTOL).
  -
  -%% mapred_bucket_stream(Pid, Bucket, Query, ClientPid, Timeout, ErrorTolerance) ->
  -%%     {ok,{MR_ReqId,MR_FSM}} = mapred_stream(Query,ClientPid,Timeout),
  -%%     {ok,_Stream_ReqID} = stream_list_keys(Bucket, Timeout, ErrorTolerance,
  -%%                                   MR_FSM, mapred),
  -%%     {ok,MR_ReqId}.
  -
  -%% mapred_bucket(Pid, Bucket, Query) ->
  -%%     mapred_bucket(Bucket, Query, ?DEFAULT_TIMEOUT).
  -
  -%% mapred_bucket(Pid, Bucket, Query, Timeout) ->
  -%%     mapred_bucket(Bucket, Query, Timeout, ?DEFAULT_ERRTOL).
  -
  -%% mapred_bucket(Pid, Bucket, Query, Timeout, ErrorTolerance) ->
  -%%     Me = self(),
  -%%     {ok,MR_ReqId} = mapred_bucket_stream(Bucket, Query, Me,
  -%%                                          Timeout, ErrorTolerance),
  -%%     luke_flow:collect_output(MR_ReqId, Timeout).
  -
  +%% @spec mapred_bucket(Pid :: pid(),
  +%%                     Query :: [riak_kv_mapred_query:mapred_queryterm()],
  +%%                     ClientPid :: pid(),
  +%%                     TimeoutMillisecs :: integer() | 'infinity') ->
  +%%       {ok, {ReqId :: term(), MR_FSM_PID :: pid()}} |
  +%%       {error, Err :: term()}
  +%% @doc Perform a map/reduce job against a bucket with a timeout 
  +%%      across the cluster.
  +%%      See the map/reduce documentation for explanation of behavior.
  +mapred_bucket(Pid, Bucket, Query) ->
  +    mapred_bucket(Pid, Bucket, Query, ?DEFAULT_TIMEOUT).
  +
  +%% @spec mapred_bucket_tream(Pid :: pid(),
  +%%                     Query :: [riak_kv_mapred_query:mapred_queryterm()],
  +%%                     ClientPid :: pid(),
  +%%                     TimeoutMillisecs :: integer() | 'infinity') ->
  +%%       {ok, {ReqId :: term(), MR_FSM_PID :: pid()}} |
  +%%       {error, Err :: term()}
  +%% @doc Perform a map/reduce job against a bucket with a timeout 
  +%%      across the cluster.
  +%%      See the map/reduce documentation for explanation of behavior.
  +mapred_bucket(Pid, Bucket, Query, Timeout) ->
  +    {ok, ReqId} = mapred_bucket_stream(Pid, Bucket, Query, self(), Timeout),
  +    wait_for_mapred(ReqId, Timeout).
  +
  +%% @spec mapred_bucket_stream(Pid :: pid(),
  +%%                     Query :: [riak_kv_mapred_query:mapred_queryterm()],
  +%%                     ClientPid :: pid(),
  +%%                     TimeoutMillisecs :: integer() | 'infinity') ->
  +%%       {ok, {ReqId :: term(), MR_FSM_PID :: pid()}} |
  +%%       {error, Err :: term()}
  +%% @doc Perform a streaming map/reduce job against a bucket with a timeout 
  +%%      across the cluster.
  +%%      See the map/reduce documentation for explanation of behavior.
  +mapred_bucket_stream(Pid, Bucket, Query, ClientPid, _Timeout) ->
  +    ReqMsg = #rpbmapredreq{input_bucket = Bucket,
  +                           phases = riakc_pb:pbify_mapred_query(Query)},
  +    ReqId = mk_reqid(),
  +    gen_server:call(Pid, {req, ReqMsg, {ReqId, ClientPid}}).
   
   %% ====================================================================
   %% gen_server callbacks
  
positive.code:          stream_list_buckets/1,
           list_keys/2,
           stream_list_keys/2,
           mapred/3,
           mapred/4,
           mapred_stream/5]).
  -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
           terminate/2, code_change/3]).
  
  
negative.code: %%                                        undefined, true}]),
  %%     ?assertEqual(lists:sort(Results), ["2", "3", "4"]).
  
  -endif.
  
  