anchor.diff:           stream_list_buckets/1,
            list_keys/2,
            stream_list_keys/2,
  -         mapred/3,
  -         mapred/4,
  -         mapred_stream/5]).
  +         mapred/3, mapred/4,
  +         mapred_stream/5,
  +         mapred_bucket/3, mapred_bucket/4,
  +         mapred_bucket_stream/5]).
  +
   -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
            terminate/2, code_change/3]).
   
  
positive.code: %%                                 [{map, {jsfun, "Riak.mapValuesJson"}, undefined, false},
  %%                                  {reduce, {jsfun, "Riak.reduceSum"}, undefined, true}])).
  
  %% %% Add testJavascriptBucketMapReduce
  
  %% javascript_arg_map_reduce_test() ->
  %%     reset_riak(),
  
negative.code:     {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_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
  