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:                          is_binary(Inputs) ->
                                Client:mapred_bucket(Inputs, Query)
                        end,
              case Results of
                  {ok, Result} ->
                      RD1 = wrq:set_resp_header("Content-Type", "application/json", RD),
                      {true, wrq:set_resp_body(mochijson2:encode(Result), RD1), State};
                  Error ->
                      error_logger:error_report(Error),
                      {{halt, 500}, send_error(Error, RD), State}
              end
      end.
  
  %% Internal functions
  send_error(Error, RD)  ->
      RD1 = wrq:set_resp_header("Content-Type", "application/json", RD),
      wrq:set_resp_body(format_error(Error), RD1).
  
  format_error({error, Message}=Error) when is_atom(Message);
                                      is_binary(Message),
  
negative.code: -define(INPUTS_TOKEN, <<"inputs">>).
  -define(DEFAULT_TIMEOUT, 30000).
  
  -record(state, {client, inputs, mrquery, boundary}).
  
  init(_) ->
      {ok, undefined}.
  