<commit_before>-module(mapred_resource).

-export([init/1, service_available/2, allowed_methods/2]).
-export([malformed_request/2, process_post/2, content_types_provided/2]).
-export([nop/2]).

-include_lib("webmachine/include/webmachine.hrl").

-define(QUERY_TOKEN, <<"query">>).
-define(INPUTS_TOKEN, <<"inputs">>).
-define(DEFAULT_TIMEOUT, 30000).

-record(state, {client, inputs, mrquery}).

init(_) ->
    {ok, undefined}.

service_available(RD, State) ->
    case riak:local_client() of
        {ok, Client} ->
            {true, RD, #state{client=Client}};
        Error ->
            error_logger:error_report(Error),
            {false, RD, State}
    end.

allowed_methods(RD, State) ->
    {['POST'], RD, State}.

malformed_request(RD, State) ->
    {IsMalformed, NewState} = case wrq:req_body(RD) of
                                  undefined ->
                                      {true, State};
                                  Body ->
                                      {Verified, State1} = verify_body(Body, State),
                                      {not(Verified), State1}
                              end,
    {IsMalformed, RD, NewState}.

content_types_provided(RD, State) ->
    {[{"application/json", nop}], RD, State}.

%% This should never get called
nop(_RD, _State) ->
    ok.

process_post(RD, #state{inputs=Inputs, mrquery=Query}=State) ->
    Me = self(),
    {ok, Client} = riak:local_client(),
    case wrq:get_qs_value("chunked", RD) of
        "true" ->
            {ok, ReqId} = 
                if is_list(Inputs) ->
                        {ok, {RId, FSM}} = Client:mapred_stream(Query, Me,
                                                                ?DEFAULT_TIMEOUT),
                        gen_fsm:send_event(FSM,{input, Inputs }),
                        gen_fsm:send_event(FSM,input_done),
                        {ok, RId};
                   is_binary(Inputs) ->
                        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};
        Param when Param =:= "false";
                   Param =:= undefined ->
            Results = if is_list(Inputs) ->
                              Client:mapred(Inputs, Query);
                         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),
                                    is_list(Message) ->
    mochijson2:encode({struct, [Error]});
format_error(_Error) ->
    mochijson2:encode({struct, [{error, map_reduce_error}]}).

stream_mapred_results(RD, ReqId) ->
    receive
        {ReqId, done} -> {<<"">>, done};
        {ReqId, {mr_results, Res}} ->
            Body = mochijson2:encode(Res),
            {iolist_to_binary(Body), fun() -> stream_mapred_results(RD, ReqId) end};
        {ReqId, {error, _}=Error} ->
            {format_error(Error), done}
    after ?DEFAULT_TIMEOUT ->
            {format_error({error, timeout}), done}
    end.

verify_body(Body, State) ->
    case mochijson2:decode(Body) of
        {struct, MapReduceDesc} ->
            Inputs = proplists:get_value(?INPUTS_TOKEN, MapReduceDesc),
            Query = proplists:get_value(?QUERY_TOKEN, MapReduceDesc),
            case not(Inputs =:= undefined) andalso not(Query =:= undefined) of
                true ->
                    case riak_mapred_json:parse_inputs(Inputs) of
                        {ok, ParsedInputs} ->
                            case riak_mapred_json:parse_query(Query) of
                                {ok, ParsedQuery} ->
                                    {true, State#state{inputs=ParsedInputs, mrquery=ParsedQuery}};
                                error ->
                                    {false, State}
                            end;
                        error ->
                            {false, State}
                    end;
                false ->
                    {false, State}
            end
    end.<commit_msg>add some helpful error messages to 400 responses from mapred_resource<commit_after>-module(mapred_resource).

-export([init/1, service_available/2, allowed_methods/2]).
-export([malformed_request/2, process_post/2, content_types_provided/2]).
-export([nop/2]).

-include_lib("webmachine/include/webmachine.hrl").

-define(QUERY_TOKEN, <<"query">>).
-define(INPUTS_TOKEN, <<"inputs">>).
-define(DEFAULT_TIMEOUT, 30000).

-record(state, {client, inputs, mrquery}).

init(_) ->
    {ok, undefined}.

service_available(RD, State) ->
    case riak:local_client() of
        {ok, Client} ->
            {true, RD, #state{client=Client}};
        Error ->
            error_logger:error_report(Error),
            {false, RD, State}
    end.

allowed_methods(RD, State) ->
    {['GET','HEAD','POST'], RD, State}.

malformed_request(RD, State) ->
    {Verified, Message, NewState} =
        case {wrq:method(RD), wrq:req_body(RD)} of
            {'POST', Body} when Body /= undefined ->
                verify_body(Body, State);
            _ ->
                {false, usage(), State}
        end,
    {not Verified,
     if Verified -> RD;
        true ->
             wrq:set_resp_header(
               "Content-Type", "text/plain",
               wrq:set_resp_body(Message, RD))
     end,
     NewState}.

content_types_provided(RD, State) ->
    {[{"application/json", nop}], RD, State}.

nop(RD, State) ->
    {usage(), RD, State}.

process_post(RD, #state{inputs=Inputs, mrquery=Query}=State) ->
    Me = self(),
    {ok, Client} = riak:local_client(),
    case wrq:get_qs_value("chunked", RD) of
        "true" ->
            {ok, ReqId} = 
                if is_list(Inputs) ->
                        {ok, {RId, FSM}} = Client:mapred_stream(Query, Me,
                                                                ?DEFAULT_TIMEOUT),
                        gen_fsm:send_event(FSM,{input, Inputs }),
                        gen_fsm:send_event(FSM,input_done),
                        {ok, RId};
                   is_binary(Inputs) ->
                        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};
        Param when Param =:= "false";
                   Param =:= undefined ->
            Results = if is_list(Inputs) ->
                              Client:mapred(Inputs, Query);
                         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),
                                    is_list(Message) ->
    mochijson2:encode({struct, [Error]});
format_error(_Error) ->
    mochijson2:encode({struct, [{error, map_reduce_error}]}).

stream_mapred_results(RD, ReqId) ->
    receive
        {ReqId, done} -> {<<"">>, done};
        {ReqId, {mr_results, Res}} ->
            Body = mochijson2:encode(Res),
            {iolist_to_binary(Body), fun() -> stream_mapred_results(RD, ReqId) end};
        {ReqId, {error, _}=Error} ->
            {format_error(Error), done}
    after ?DEFAULT_TIMEOUT ->
            {format_error({error, timeout}), done}
    end.

verify_body(Body, State) ->
    case catch mochijson2:decode(Body) of
        {struct, MapReduceDesc} ->
            Inputs = proplists:get_value(?INPUTS_TOKEN, MapReduceDesc),
            Query = proplists:get_value(?QUERY_TOKEN, MapReduceDesc),
            case not(Inputs =:= undefined) andalso not(Query =:= undefined) of
                true ->
                    case riak_mapred_json:parse_inputs(Inputs) of
                        {ok, ParsedInputs} ->
                            case riak_mapred_json:parse_query(Query) of
                                {ok, ParsedQuery} ->
                                    {true, [], State#state{inputs=ParsedInputs,
                                                           mrquery=ParsedQuery}};
                                error ->
                                    {false,
                                     "An error occurred parsing "
                                     "the \"query\" field.\n",
                                     State}
                            end;
                        error ->
                            {false,
                             "An error occurred parsing the \"inputs\" field.\n",
                             State}
                    end;
                false ->
                    {false,
                     "The post body was missing the "
                     "\"inputs\" or \"query\" field.\n",
                     State}
            end;
        {'EXIT', Message} ->
            {false,
             io_lib:format("The POST body was not valid JSON.~n"
                           "The error from the parser was:~n~p~n",
                           [Message]),
             State};
        _ ->
            {false, "The POST body was not a JSON object.\n", State}
    end.

usage() ->
    "This resource accepts POSTs with bodies containing JSON of the form:\n"
        "{\n"
        " \"inputs\":[...list of inputs...],\n"
        " \"query\":[...list of map/reduce phases...]\n"
        "}\n".
{'fraction_non_alphanumeric': 0.12792726604830562, 'fraction_numerical': 0.005418312058040224, 'mean_word_length': 1.4817684594348222, 'pattern_counts': {'":': 2, '<': 15, '<?xml version=': 0, '>': 92, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 29, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1987164', 'n_tokens_mistral': 3181, 'n_tokens_neox': 3064, 'n_words': 601}