anchor.diff:      PbProps = riakc_pb:pbify_rpbbucketprops(BucketProps),
       Req = #rpbsetbucketreq{bucket = Bucket, props = PbProps},
       gen_server:call(Pid, {req, Req}).
  +
  +%% @spec mapred(Inputs :: list(),
  +%%              Query :: [riak_kv_mapred_query:mapred_queryterm()]) ->
  +%%       {ok, riak_kv_mapred_query:mapred_result()} |
  +%%       {error, {bad_qterm, riak_kv_mapred_query:mapred_queryterm()}} |
  +%%       {error, timeout} |
  +%%       {error, Err :: term()}
  +%% @doc Perform a map/reduce job across the cluster.
  +%%      See the map/reduce documentation for explanation of behavior.
  +%% @equiv mapred(Inputs, Query, default_timeout())
  +mapred(Pid, Inputs, Query) -> 
  +    mapred(Pid, Inputs, Query, ?DEFAULT_TIMEOUT).
  +
  +%% @spec mapred(Inputs :: list(),
  +%%              Query :: [riak_kv_mapred_query:mapred_queryterm()],
  +%%              TimeoutMillisecs :: integer()  | 'infinity') ->
  +%%       {ok, riak_kv_mapred_query:mapred_result()} |
  +%%       {error, timeout} |
  +%%       {error, Err :: term()}
  +%% @doc Perform a map/reduce job across the cluster.
  +%%      See the map/reduce documentation for explanation of behavior.
  +mapred(Pid, Inputs, Query, Timeout) ->
  +    {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) ->
  +    MapRed = [{'inputs', Inputs},
  +              {'query', Query},
  +              {'timeout', Timeout}],
  +    send_mapred_req(Pid, MapRed, 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) ->
  +    MapRed = [{'inputs', Bucket},
  +              {'query', Query},
  +              {'timeout', Timeout}],
  +    send_mapred_req(Pid, MapRed, ClientPid).
   
   %% ====================================================================
   %% gen_server callbacks
  
positive.code:     {reply, {ok, ReqId}, State};
  after_send(rpblistbucketsreq, {ReqId, _Client}, State) ->
      {reply, {ok, ReqId}, State};
  after_send(_Req, _Ctx, State) ->
      {noreply, State}.
  
  
negative.code:                  resume_riak_pb_sockets(),
                   receive {1,Ping1} -> ?assertEqual(Ping1, pong) end,
                   receive {2,Ping2} -> ?assertEqual(Ping2, pong) end
               end)},
  
       {"javascript_source_map_test()",
        ?_test(begin
                   reset_riak(),
                   {ok, Pid} = start_link(?TEST_IP, ?TEST_PORT),
                   B = <<"bucket">>,
                   K = <<"foo">>,
                   O=riakc_obj:new(B, K),
                   ?MODULE:put(Pid, riakc_obj:update_value(O, <<"2">>, "application/json")),
  
                   ?assertEqual({ok, [{0, [2]}]},
                                ?MODULE:mapred(Pid, 
                                               [{B, K}],
                                               [{map, {jsanon, <<"function (v) { return [JSON.parse(v.values[0].data)]; }">>},
                                                 undefined, true}]))
               end)},
  
       {"javascript_named_map_test()",
        ?_test(begin
                   reset_riak(),
                   {ok, Pid} = start_link(?TEST_IP, ?TEST_PORT),
                   B = <<"bucket">>,
                   K = <<"foo">>,
                   O=riakc_obj:new(B, K),
                   ?MODULE:put(Pid, riakc_obj:update_value(O, <<"99">>, "application/json")),
  
                   ?assertEqual({ok, [{0, [99]}]},
                                ?MODULE:mapred(Pid, 
                                               [{B, K}],
                                               [{map, {jsfun, <<"Riak.mapValuesJson">>},
                                                 undefined, true}]))
               end)},
   
       {"javascript_source_map_reduce_test()", 
        ?_test(begin
                   reset_riak(),
                   {ok, Pid} = start_link(?TEST_IP, ?TEST_PORT),
                   Store = fun({K,V}) ->
                                   O=riakc_obj:new(<<"bucket">>, K),
                                   ?MODULE:put(Pid,riakc_obj:update_value(O, V, "application/json"))
                           end,
                   [Store(KV) || KV <- [{<<"foo">>, <<"2">>}, 
                                        {<<"bar">>, <<"3">>}, 
                                        {<<"baz">>, <<"4">>}]],
  
                   ?assertEqual({ok, [{1, 3}]},
                                ?MODULE:mapred(Pid, 
                                               [{<<"bucket">>, <<"foo">>},
                                                {<<"bucket">>, <<"bar">>},
                                                {<<"bucket">>, <<"baz">>}],
                                               [{map, {jsanon, <<"function (v) { return [1]; }">>}, 
                                                 undefined, false},
                                                {reduce, {jsanon, <<"function(v) { return v.length; } ">>}, 
                                                 undefined, true}]))
               end)},
  
       {"javascript_named_map_reduce_test()",
        ?_test(begin
                   reset_riak(),
                   {ok, Pid} = start_link(?TEST_IP, ?TEST_PORT),
                   Store = fun({K,V}) ->
                                   O=riakc_obj:new(<<"bucket">>, K),
                                   ?MODULE:put(Pid,riakc_obj:update_value(O, V, "application/json"))
                           end,
                   [Store(KV) || KV <- [{<<"foo">>, <<"2">>}, 
                                        {<<"bar">>, <<"3">>}, 
                                        {<<"baz">>, <<"4">>}]],
  
                   ?assertEqual({ok, [{1, [9]}]},
                                ?MODULE:mapred(Pid, 
                                               [{<<"bucket">>, <<"foo">>},
                                                {<<"bucket">>, <<"bar">>},
                                                {<<"bucket">>, <<"baz">>}],
                                               [{map, {jsfun, <<"Riak.mapValuesJson">>}, undefined, false},
                                                {reduce, {jsfun, <<"Riak.reduceSum">>}, undefined, true}]))
               end)},
  
       {"javascript_bucket_map_reduce_test()",
        ?_test(begin
                   reset_riak(),
                   {ok, Pid} = start_link(?TEST_IP, ?TEST_PORT),
                   Store = fun({K,V}) ->
                                   O=riakc_obj:new(<<"bucket">>, K),
                                   ?MODULE:put(Pid,riakc_obj:update_value(O, V, "application/json"))
                           end,
                   [Store(KV) || KV <- [{<<"foo">>, <<"2">>}, 
                                        {<<"bar">>, <<"3">>}, 
                                        {<<"baz">>, <<"4">>}]],
  
                   ?assertEqual({ok, [{1, [9]}]},
                                ?MODULE:mapred_bucket(Pid, <<"bucket">>,
                                                      [{map, {jsfun, <<"Riak.mapValuesJson">>}, undefined, false},
                                                       {reduce, {jsfun, <<"Riak.reduceSum">>}, undefined, true}]))
               end)},
  
       {"javascript_arg_map_reduce_test()",
        ?_test(begin
                   reset_riak(),
                   {ok, Pid} = start_link(?TEST_IP, ?TEST_PORT),
                   O=riakc_obj:new(<<"bucket">>, <<"foo">>),
                   ?MODULE:put(Pid, riakc_obj:update_value(O, <<"2">>, "application/json")),
                   ?assertEqual({ok, [{1, [10]}]},
                                ?MODULE:mapred(Pid, 
                                               [{{<<"bucket">>, <<"foo">>}, 5},
                                                {{<<"bucket">>, <<"foo">>}, 10},
                                                {{<<"bucket">>, <<"foo">>}, 15},
                                                {{<<"bucket">>, <<"foo">>}, -15},
                                                {{<<"bucket">>, <<"foo">>}, -5}],
                                               [{map, {jsanon, <<"function(v, arg) { return [arg]; }">>},
                                                 undefined, false},
                                                {reduce, {jsfun, <<"Riak.reduceSum">>}, undefined, true}]))
               end)},
  
       {"erlang_map_reduce_test()",
        ?_test(begin
                   reset_riak(),
                   {ok, Pid} = start_link(?TEST_IP, ?TEST_PORT),
                   Store = fun({K,V}) ->
                                   O=riakc_obj:new(<<"bucket">>, K),
                                   ?MODULE:put(Pid,riakc_obj:update_value(O, V, "application/json"))
                           end,
                   [Store(KV) || KV <- [{<<"foo">>, <<"2">>}, 
                                        {<<"bar">>, <<"3">>}, 
                                        {<<"baz">>, <<"4">>}]],
  
                   {ok, [{1, Results}]} = ?MODULE:mapred(Pid, 
                                                         [{<<"bucket">>, <<"foo">>},
                                                          {<<"bucket">>, <<"bar">>},
                                                          {<<"bucket">>, <<"baz">>}],
                                                         [{map, {modfun, riak_kv_mapreduce,
                                                                 map_object_value}, 
                                                           undefined, false},
                                                          {reduce, {modfun, riak_kv_mapreduce, 
                                                                    reduce_set_union}, 
                                                           undefined, true}]),
                   ?assertEqual([<<"2">>, <<"3">>, <<"4">>], lists:sort(Results))
               end)}
  
  
       ].
  
  -endif.
  