anchor.diff:  format_error(_Error) ->
       mochijson2:encode({struct, [{error, map_reduce_error}]}).
   
  -stream_mapred_results(RD, ReqId) ->
  +stream_mapred_results(RD, ReqId, State) ->
       receive
  -        {ReqId, done} -> {<<"">>, done};
  +        {ReqId, done} -> {iolist_to_binary(["\n--", State#state.boundary, "--\n"]), done};
           {ReqId, {mr_results, Res}} ->
  -            Body = mochijson2:encode(Res),
  -            {iolist_to_binary(Body), fun() -> stream_mapred_results(RD, ReqId) end};
  +            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 ->
  
positive.code: -define(INPUTS_TOKEN, <<"inputs">>).
  -define(DEFAULT_TIMEOUT, 30000).
  
  -record(state, {client, inputs, mrquery}).
  
  init(_) ->
      {ok, undefined}.
  
negative.code:                          is_binary(Inputs) ->
                                Client:mapred_bucket(Inputs, Query)
                        end,
              RD1 = wrq:set_resp_header("Content-Type", "application/json", RD),
              case Results of
                  {ok, Result} ->
                      {true, wrq:set_resp_body(mochijson2:encode(Result), RD1), State};
                  Error ->
                      error_logger:error_report(Error),
                      {{halt, 500}, send_error(Error, RD1), State}
              end
      end.
  
  %% Internal functions
  send_error(Error, RD)  ->
      wrq:set_resp_body(format_error(Error), RD).
  
  format_error({error, Message}=Error) when is_atom(Message);
                                      is_binary(Message),
  