anchor.diff:                   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.
  
positive.code:         {ReqId, {error, Reason}} -> {error, Reason}
      after Timeout ->
              {error, {timeout, Acc}}
      end.
  
  %% ====================================================================
  
negative.code:     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
  