anchor.diff:  %%      sending results to ClientPid.
   %%      See the map/reduce documentation for explanation of behavior.
   mapred_stream(Pid, Inputs, Query, ClientPid, Timeout) ->
  +    mapred_stream(Pid, Inputs, Query, ClientPid, Timeout,
  +                  default_timeout(mapred_stream_call_timeout)).
  +
  +%% @spec mapred_stream(Pid :: pid(),
  +%%                     Inputs :: list(),
  +%%                     Query :: [riak_kv_mapred_query:mapred_queryterm()],
  +%%                     ClientPid :: pid(),
  +%%                     TimeoutMillisecs :: integer() | 'infinity',
  +%%                     CallTimeoutMillisecs :: integer() | 'infinity') ->
  +%%       {ok, {ReqId :: term(), MR_FSM_PID :: pid()}} |
  +%%       {error, Err :: term()}
  +%% @doc Perform a streaming map/reduce job with a map/red timeout across the cluster,
  +%%      a local call timeout and sending results to ClientPid.
  +%%      See the map/reduce documentation for explanation of behavior.
  +mapred_stream(Pid, Inputs, Query, ClientPid, Timeout, CallTimeout) ->
       MapRed = [{'inputs', Inputs},
                 {'query', Query},
                 {'timeout', Timeout}],
  -    send_mapred_req(Pid, MapRed, ClientPid).
  +    send_mapred_req(Pid, MapRed, ClientPid, CallTimeout).
   
   %% @spec mapred_bucket(Pid :: pid(),
   %%                     Bucket :: bucket(),
  
positive.code: 
  %% Common code for sending a single bucket or multiple inputs map/request
  %% @private
  send_mapred_req(Pid, MapRed, ClientPid) ->
      ReqMsg = #rpbmapredreq{request = encode_mapred_req(MapRed),
                             content_type = <<"application/x-erlang-binary">>},
      ReqId = mk_reqid(),
      %% Add an extra 100ms to the mapred timeout and use that for the 
      %% socket timeout.  This should give the map/reduce a chance to fail and let us know.
      Timeout = proplists:get_value(timeout, MapRed, default_timeout(mapred_timeout)) + 100,
      gen_server:call(Pid, {req, ReqMsg, Timeout, {ReqId, ClientPid}}).
  
  %% @private
  %% Make a new request that can be sent or queued
  
negative.code:     %% Run a Map/Reduce operation using reduce_identity to get a list
      %% of BKeys.
      IdentityQuery = [{reduce, {modfun, riak_kv_mapreduce, reduce_identity}, none, true}],
      case search(Pid, Bucket, SearchQuery, IdentityQuery,
                  default_timeout(search_timeout)) of
          {ok, [{_, Results}]} -> 
              %% Unwrap the results.
              {ok, Results};
  