%% Copyright (c) 2016-2017, Loïc Hoguin <[email protected]>
%%
%% Permission to use, copy, modify, and/or distribute this software for any
%% purpose with or without fee is hereby granted, provided that the above
%% copyright notice and this permission notice appear in all copies.
%%
%% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
%% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
%% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
%% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
%% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
%% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
%% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-module(cowboy_stream_h).
-behavior(cowboy_stream).
%% @todo Maybe have a callback for the type of process this is, worker or supervisor.
-export([init/3]).
-export([data/4]).
-export([info/3]).
-export([terminate/3]).
-export([proc_lib_hack/3]).
-export([execute/3]).
-export([resume/5]).
%% @todo Need to call subsequent handlers.
-record(state, {
ref = undefined :: ranch:ref(),
pid = undefined :: pid(),
read_body_ref = undefined :: reference() | undefined,
read_body_timer_ref = undefined :: reference() | undefined,
read_body_length = 0 :: non_neg_integer() | infinity,
read_body_is_fin = nofin :: nofin | {fin, non_neg_integer()},
read_body_buffer = <<>> :: binary()
}).
%% @todo For shutting down children we need to have a timeout before we terminate
%% the stream like supervisors do. So here just send a message to yourself first,
%% and then decide what to do when receiving this message.
-spec init(cowboy_stream:streamid(), cowboy_req:req(), cowboy:opts())
-> {[{spawn, pid(), timeout()}], #state{}}.
init(_StreamID, Req=#{ref := Ref}, Opts) ->
Env = maps:get(env, Opts, #{}),
Middlewares = maps:get(middlewares, Opts, [cowboy_router, cowboy_handler]),
Shutdown = maps:get(shutdown, Opts, 5000),
Pid = proc_lib:spawn_link(?MODULE, proc_lib_hack, [Req, Env, Middlewares]),
{[{spawn, Pid, Shutdown}], #state{ref=Ref, pid=Pid}}.
%% If we receive data and stream is waiting for data:
%% If we accumulated enough data or IsFin=fin, send it.
%% If not, buffer it.
%% If not, buffer it.
-spec data(cowboy_stream:streamid(), cowboy_stream:fin(), cowboy_req:resp_body(), State)
-> {cowboy_stream:commands(), State} when State::#state{}.
data(_StreamID, IsFin, Data, State=#state{read_body_ref=undefined, read_body_buffer=Buffer}) ->
{[], State#state{read_body_is_fin=IsFin, read_body_buffer= << Buffer/binary, Data/binary >>}};
data(_StreamID, nofin, Data, State=#state{read_body_length=Length, read_body_buffer=Buffer}) when byte_size(Data) + byte_size(Buffer) < Length ->
{[], State#state{read_body_buffer= << Buffer/binary, Data/binary >>}};
data(_StreamID, IsFin, Data, State=#state{pid=Pid, read_body_ref=Ref,
read_body_timer_ref=TRef, read_body_buffer=Buffer}) ->
ok = erlang:cancel_timer(TRef, [{async, true}, {info, false}]),
Pid ! {request_body, Ref, IsFin, << Buffer/binary, Data/binary >>},
{[], State#state{read_body_ref=undefined, read_body_timer_ref=undefined, read_body_buffer= <<>>}}.
-spec info(cowboy_stream:streamid(), any(), State)
-> {cowboy_stream:commands(), State} when State::#state{}.
info(_StreamID, {'EXIT', Pid, normal}, State=#state{pid=Pid}) ->
%% @todo Do we even reach this clause?
{[stop], State};
info(_StreamID, {'EXIT', Pid, {_Reason, [_, {cow_http_hd, _, _, _}|_]}}, State=#state{pid=Pid}) ->
%% @todo Have an option to enable/disable this specific crash report?
%%report_crash(Ref, StreamID, Pid, Reason, Stacktrace),
%% @todo Headers? Details in body? More stuff in debug only?
{[{error_response, 400, #{<<"content-length">> => <<"0">>}, <<>>}, stop], State};
info(StreamID, Exit = {'EXIT', Pid, {Reason, Stacktrace}}, State=#state{ref=Ref, pid=Pid}) ->
report_crash(Ref, StreamID, Pid, Reason, Stacktrace),
{[
{error_response, 500, #{<<"content-length">> => <<"0">>}, <<>>},
{internal_error, Exit, 'Stream process crashed.'}
], State};
%% Request body, no body buffer but IsFin=fin.
%info(_StreamID, {read_body, Ref, _, _}, State=#state{pid=Pid, read_body_is_fin=fin, read_body_buffer= <<>>}) ->
% Pid ! {request_body, Ref, fin, <<>>},
% {[], State};
%% Request body, body buffered large enough or complete.
info(_StreamID, {read_body, Ref, Length, _},
State=#state{pid=Pid, read_body_is_fin=IsFin, read_body_buffer=Data})
when element(1, IsFin) =:= fin; byte_size(Data) >= Length ->
Pid ! {request_body, Ref, IsFin, Data},
{[], State#state{read_body_buffer= <<>>}};
%% Request body, not enough to send yet.
info(StreamID, {read_body, Ref, Length, Period}, State) ->
TRef = erlang:send_after(Period, self(), {{self(), StreamID}, {read_body_timeout, Ref}}),
{[{flow, Length}], State#state{read_body_ref=Ref, read_body_timer_ref=TRef, read_body_length=Length}};
%% Request body reading timeout; send what we got.
info(_StreamID, {read_body_timeout, Ref}, State=#state{pid=Pid, read_body_ref=Ref,
read_body_is_fin=IsFin, read_body_buffer=Buffer}) ->
Pid ! {request_body, Ref, IsFin, Buffer},
{[], State#state{read_body_ref=undefined, read_body_timer_ref=undefined, read_body_buffer= <<>>}};
info(_StreamID, {read_body_timeout, _}, State) ->
{[], State};
%% Response.
info(_StreamID, Response = {response, _, _, _}, State) ->
{[Response], State};
info(_StreamID, Headers = {headers, _, _}, State) ->
{[Headers], State};
info(_StreamID, Data = {data, _, _}, State) ->
{[Data], State};
info(_StreamID, Push = {push, _, _, _, _, _, _, _}, State) ->
{[Push], State};
info(_StreamID, SwitchProtocol = {switch_protocol, _, _, _}, State) ->
{[SwitchProtocol], State};
%% Stray message.
info(_StreamID, _Info, State) ->
%% @todo Error report.
%% @todo Cleanup if no reply was sent when stream ends.
{[], State}.
-spec terminate(cowboy_stream:streamid(), cowboy_stream:reason(), #state{}) -> ok.
terminate(_StreamID, _Reason, _State) ->
ok.
%% We use ~999999p here instead of ~w because the latter doesn't
%% support printable strings.
report_crash(_, _, _, normal, _) ->
ok;
report_crash(_, _, _, shutdown, _) ->
ok;
report_crash(_, _, _, {shutdown, _}, _) ->
ok;
report_crash(Ref, StreamID, Pid, Reason, Stacktrace) ->
error_logger:error_msg(
"Ranch listener ~p, connection process ~p, stream ~p "
"had its request process ~p exit with reason "
"~999999p and stacktrace ~999999p~n",
[Ref, self(), StreamID, Pid, Reason, Stacktrace]).
%% Request process.
%% @todo This should wrap with try/catch to get the full error
%% in the stream handler. Only then can we decide what to do
%% about it. This means that we should remove any other try/catch
%% in the request process.
%% This hack is necessary because proc_lib does not propagate
%% stacktraces by default. This is ugly because we end up
%% having two try/catch instead of one (the one in proc_lib),
%% just to add the stacktrace information.
%%
%% @todo Remove whenever proc_lib propagates stacktraces.
-spec proc_lib_hack(_, _, _) -> _.
proc_lib_hack(Req, Env, Middlewares) ->
try
execute(Req, Env, Middlewares)
catch
_:Reason when element(1, Reason) =:= cowboy_handler ->
exit(Reason);
_:Reason ->
exit({Reason, erlang:get_stacktrace()})
end.
%% @todo
%-spec execute(cowboy_req:req(), #state{}, cowboy_middleware:env(), [module()])
% -> ok.
-spec execute(_, _, _) -> _.
execute(_, _, []) ->
ok; %% @todo Maybe error reason should differ here and there.
execute(Req, Env, [Middleware|Tail]) ->
case Middleware:execute(Req, Env) of
{ok, Req2, Env2} ->
execute(Req2, Env2, Tail);
{suspend, Module, Function, Args} ->
proc_lib:hibernate(?MODULE, resume, [Env, Tail, Module, Function, Args]);
{stop, _Req2} ->
ok %% @todo Maybe error reason should differ here and there.
end.
-spec resume(cowboy_middleware:env(), [module()],
module(), module(), [any()]) -> ok.
resume(Env, Tail, Module, Function, Args) ->
case apply(Module, Function, Args) of
{ok, Req2, Env2} ->
execute(Req2, Env2, Tail);
{suspend, Module2, Function2, Args2} ->
proc_lib:hibernate(?MODULE, resume, [Env, Tail, Module2, Function2, Args2]);
{stop, _Req2} ->
ok %% @todo Maybe error reason should differ here and there.
end.
