anchor.diff:                          Client:mapred_bucket_stream(Inputs, Query, Me,
                                                       ?DEFAULT_TIMEOUT)
                   end,
  -            RD1 = wrq:set_resp_header("Content-Type", "application/json", RD),
  -            {true, wrq:set_resp_body({stream, stream_mapred_results(RD1, ReqId)}, RD1), State};
  +            Boundary = riak_util:unique_id_62(),
  +            RD1 = wrq:set_resp_header("Content-Type", "multipart/mixed;boundary=" ++ Boundary, RD),
  +            State1 = State#state{boundary=Boundary},
  +            {true, wrq:set_resp_body({stream, stream_mapred_results(RD1, ReqId, State1)}, RD1), State1};
           Param when Param =:= "false";
                      Param =:= undefined ->
               Results = if is_list(Inputs) ->
  
positive.code: -define(INPUTS_TOKEN, <<"inputs">>).
  -define(DEFAULT_TIMEOUT, 30000).
  
  -record(state, {client, inputs, mrquery}).
  
  init(_) ->
      {ok, undefined}.
  
negative.code: format_error(_Error) ->
      mochijson2:encode({struct, [{error, map_reduce_error}]}).
  
  stream_mapred_results(RD, ReqId, State) ->
      receive
          {ReqId, done} -> {iolist_to_binary(["\n--", State#state.boundary, "--\n"]), done};
          {ReqId, {mr_results, Res}} ->
              Data = mochijson2:encode(Res),
              Body = ["\n--", State#state.boundary, "\n",
                      "Content-Type: application/json\n\n",
                      Data],
              {iolist_to_binary(Body), fun() -> stream_mapred_results(RD, ReqId, State) end};
          {ReqId, {error, _}=Error} ->
              {format_error(Error), done}
      after ?DEFAULT_TIMEOUT ->
  