mirror of
https://github.com/ninenines/cowboy.git
synced 2025-07-14 20:30:23 +00:00
Make sure cowboy_stream_h calls subsequent stream handlers
This commit is contained in:
Loïc Hoguin
parent
68ede9cfe4
commit
34f57ebbd3
1 changed files with 80 additions and 56 deletions
|
|
@ -29,9 +29,8 @@
|
|||
-export([execute/3]).
|
||||
-export([resume/5]).
|
||||
|
||||
%% @todo Need to call subsequent handlers.
|
||||
|
||||
-record(state, {
|
||||
next :: any(),
|
||||
ref = undefined :: ranch:ref(),
|
||||
pid = undefined :: pid(),
|
||||
expect = undefined :: undefined | continue,
|
||||
|
|
@ -49,13 +48,15 @@
|
|||
|
||||
-spec init(cowboy_stream:streamid(), cowboy_req:req(), cowboy:opts())
|
||||
-> {[{spawn, pid(), timeout()}], #state{}}.
|
||||
init(_StreamID, Req=#{ref := Ref}, Opts) ->
|
||||
init(StreamID, Req=#{ref := Ref}, Opts) ->
|
||||
Env = maps:get(env, Opts, #{}),
|
||||
Middlewares = maps:get(middlewares, Opts, [cowboy_router, cowboy_handler]),
|
||||
Shutdown = maps:get(shutdown_timeout, Opts, 5000),
|
||||
Pid = proc_lib:spawn_link(?MODULE, request_process, [Req, Env, Middlewares]),
|
||||
Expect = expect(Req),
|
||||
{[{spawn, Pid, Shutdown}], #state{ref=Ref, pid=Pid, expect=Expect}}.
|
||||
{Commands, Next} = cowboy_stream:init(StreamID, Req, Opts),
|
||||
{[{spawn, Pid, Shutdown}|Commands],
|
||||
#state{next=Next, ref=Ref, pid=Pid, expect=Expect}}.
|
||||
|
||||
%% Ignore the expect header in HTTP/1.0.
|
||||
expect(#{version := 'HTTP/1.0'}) ->
|
||||
|
|
@ -81,49 +82,58 @@ expect(Req) ->
|
|||
-spec data(cowboy_stream:streamid(), cowboy_stream:fin(), cowboy_req:resp_body(), State)
|
||||
-> {cowboy_stream:commands(), State} when State::#state{}.
|
||||
%% Stream isn't waiting for data.
|
||||
data(_StreamID, IsFin, Data, State=#state{
|
||||
data(StreamID, IsFin, Data, State=#state{
|
||||
read_body_ref=undefined, read_body_buffer=Buffer, body_length=BodyLen}) ->
|
||||
{[], State#state{
|
||||
do_data(StreamID, IsFin, Data, [], State#state{
|
||||
expect=undefined,
|
||||
read_body_is_fin=IsFin,
|
||||
read_body_buffer= << Buffer/binary, Data/binary >>,
|
||||
body_length=BodyLen + byte_size(Data)}};
|
||||
body_length=BodyLen + byte_size(Data)
|
||||
});
|
||||
%% Stream is waiting for data using auto mode.
|
||||
%%
|
||||
%% There is no buffering done in auto mode.
|
||||
data(_StreamID, IsFin, Data, State=#state{pid=Pid, read_body_ref=Ref,
|
||||
data(StreamID, IsFin, Data, State=#state{pid=Pid, read_body_ref=Ref,
|
||||
read_body_length=auto, body_length=BodyLen}) ->
|
||||
send_request_body(Pid, Ref, IsFin, BodyLen, Data),
|
||||
{[{flow, byte_size(Data)}], State#state{
|
||||
do_data(StreamID, IsFin, Data, [{flow, byte_size(Data)}], State#state{
|
||||
read_body_ref=undefined,
|
||||
body_length=BodyLen}};
|
||||
body_length=BodyLen
|
||||
});
|
||||
%% Stream is waiting for data but we didn't receive enough to send yet.
|
||||
data(_StreamID, nofin, Data, State=#state{
|
||||
data(StreamID, IsFin=nofin, Data, State=#state{
|
||||
read_body_length=ReadLen, read_body_buffer=Buffer, body_length=BodyLen})
|
||||
when byte_size(Data) + byte_size(Buffer) < ReadLen ->
|
||||
{[], State#state{
|
||||
do_data(StreamID, IsFin, Data, [], State#state{
|
||||
expect=undefined,
|
||||
read_body_buffer= << Buffer/binary, Data/binary >>,
|
||||
body_length=BodyLen + byte_size(Data)}};
|
||||
body_length=BodyLen + byte_size(Data)
|
||||
});
|
||||
%% Stream is waiting for data and we received enough to send.
|
||||
data(_StreamID, IsFin, Data, State=#state{pid=Pid, read_body_ref=Ref,
|
||||
data(StreamID, IsFin, Data, State=#state{pid=Pid, read_body_ref=Ref,
|
||||
read_body_timer_ref=TRef, read_body_buffer=Buffer, body_length=BodyLen0}) ->
|
||||
BodyLen = BodyLen0 + byte_size(Data),
|
||||
%% @todo Handle the infinity case where no TRef was defined.
|
||||
ok = erlang:cancel_timer(TRef, [{async, true}, {info, false}]),
|
||||
send_request_body(Pid, Ref, IsFin, BodyLen, <<Buffer/binary, Data/binary>>),
|
||||
{[], State#state{
|
||||
do_data(StreamID, IsFin, Data, [], State#state{
|
||||
expect=undefined,
|
||||
read_body_ref=undefined,
|
||||
read_body_timer_ref=undefined,
|
||||
read_body_buffer= <<>>,
|
||||
body_length=BodyLen}}.
|
||||
body_length=BodyLen
|
||||
}).
|
||||
|
||||
do_data(StreamID, IsFin, Data, Commands1, State=#state{next=Next0}) ->
|
||||
{Commands2, Next} = cowboy_stream:data(StreamID, IsFin, Data, Next0),
|
||||
{Commands1 ++ Commands2, State#state{next=Next}}.
|
||||
|
||||
-spec info(cowboy_stream:streamid(), any(), State)
|
||||
-> {cowboy_stream:commands(), State} when State::#state{}.
|
||||
info(_StreamID, {'EXIT', Pid, normal}, State=#state{pid=Pid}) ->
|
||||
{[stop], State};
|
||||
info(_StreamID, {'EXIT', Pid, {{request_error, Reason, _HumanReadable}, _}}, State=#state{pid=Pid}) ->
|
||||
info(StreamID, Info={'EXIT', Pid, normal}, State=#state{pid=Pid}) ->
|
||||
do_info(StreamID, Info, [stop], State);
|
||||
info(StreamID, Info={'EXIT', Pid, {{request_error, Reason, _HumanReadable}, _}},
|
||||
State=#state{pid=Pid}) ->
|
||||
%% @todo Optionally report the crash to help debugging.
|
||||
%%report_crash(Ref, StreamID, Pid, Reason, Stacktrace),
|
||||
Status = case Reason of
|
||||
|
|
@ -132,80 +142,94 @@ info(_StreamID, {'EXIT', Pid, {{request_error, Reason, _HumanReadable}, _}}, Sta
|
|||
_ -> 400
|
||||
end,
|
||||
%% @todo Headers? Details in body? More stuff in debug only?
|
||||
{[{error_response, Status, #{<<"content-length">> => <<"0">>}, <<>>}, stop], State};
|
||||
info(StreamID, Exit = {'EXIT', Pid, {Reason, Stacktrace}}, State=#state{ref=Ref, pid=Pid}) ->
|
||||
do_info(StreamID, Info, [
|
||||
{error_response, Status, #{<<"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),
|
||||
{[
|
||||
do_info(StreamID, Exit, [
|
||||
{error_response, 500, #{<<"content-length">> => <<"0">>}, <<>>},
|
||||
{internal_error, Exit, 'Stream process crashed.'}
|
||||
], State};
|
||||
], State);
|
||||
%% Request body, auto mode, no body buffered.
|
||||
info(_StreamID, {read_body, Ref, auto, infinity}, State=#state{read_body_buffer= <<>>}) ->
|
||||
{[], State#state{
|
||||
info(StreamID, Info={read_body, Ref, auto, infinity}, State=#state{read_body_buffer= <<>>}) ->
|
||||
do_info(StreamID, Info, [], State#state{
|
||||
read_body_ref=Ref,
|
||||
read_body_length=auto}};
|
||||
read_body_length=auto
|
||||
});
|
||||
%% Request body, auto mode, body buffered or complete.
|
||||
info(_StreamID, {read_body, Ref, auto, infinity}, State=#state{pid=Pid,
|
||||
info(StreamID, Info={read_body, Ref, auto, infinity}, State=#state{pid=Pid,
|
||||
read_body_is_fin=IsFin, read_body_buffer=Buffer, body_length=BodyLen}) ->
|
||||
send_request_body(Pid, Ref, IsFin, BodyLen, Buffer),
|
||||
{[{flow, byte_size(Buffer)}], State#state{read_body_buffer= <<>>}};
|
||||
do_info(StreamID, Info, [{flow, byte_size(Buffer)}],
|
||||
State#state{read_body_buffer= <<>>});
|
||||
%% Request body, body buffered large enough or complete.
|
||||
%%
|
||||
%% We do not send a 100 continue response if the client
|
||||
%% already started sending the body.
|
||||
info(_StreamID, {read_body, Ref, Length, _}, State=#state{pid=Pid,
|
||||
info(StreamID, Info={read_body, Ref, Length, _}, State=#state{pid=Pid,
|
||||
read_body_is_fin=IsFin, read_body_buffer=Buffer, body_length=BodyLen})
|
||||
when IsFin =:= fin; byte_size(Buffer) >= Length ->
|
||||
send_request_body(Pid, Ref, IsFin, BodyLen, Buffer),
|
||||
{[], State#state{read_body_buffer= <<>>}};
|
||||
do_info(StreamID, Info, [], State#state{read_body_buffer= <<>>});
|
||||
%% Request body, not enough to send yet.
|
||||
info(StreamID, {read_body, Ref, Length, Period}, State=#state{expect=Expect}) ->
|
||||
info(StreamID, Info={read_body, Ref, Length, Period}, State=#state{expect=Expect}) ->
|
||||
Commands = case Expect of
|
||||
continue -> [{inform, 100, #{}}, {flow, Length}];
|
||||
undefined -> [{flow, Length}]
|
||||
end,
|
||||
%% @todo Handle the case where Period =:= infinity.
|
||||
TRef = erlang:send_after(Period, self(), {{self(), StreamID}, {read_body_timeout, Ref}}),
|
||||
{Commands, State#state{
|
||||
do_info(StreamID, Info, Commands, State#state{
|
||||
read_body_ref=Ref,
|
||||
read_body_timer_ref=TRef,
|
||||
read_body_length=Length}};
|
||||
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,
|
||||
info(StreamID, Info={read_body_timeout, Ref}, State=#state{pid=Pid, read_body_ref=Ref,
|
||||
read_body_is_fin=IsFin, read_body_buffer=Buffer, body_length=BodyLen}) ->
|
||||
send_request_body(Pid, Ref, IsFin, BodyLen, Buffer),
|
||||
{[], State#state{read_body_ref=undefined, read_body_timer_ref=undefined, read_body_buffer= <<>>}};
|
||||
info(_StreamID, {read_body_timeout, _}, State) ->
|
||||
{[], State};
|
||||
do_info(StreamID, Info, [], State#state{
|
||||
read_body_ref=undefined,
|
||||
read_body_timer_ref=undefined,
|
||||
read_body_buffer= <<>>
|
||||
});
|
||||
info(StreamID, Info={read_body_timeout, _}, State) ->
|
||||
do_info(StreamID, Info, [], State);
|
||||
%% Response.
|
||||
%%
|
||||
%% We reset the expect field when a 100 continue response
|
||||
%% is sent or when any final response is sent.
|
||||
info(_StreamID, Inform = {inform, Status, _}, State0) ->
|
||||
info(StreamID, Inform={inform, Status, _}, State0) ->
|
||||
State = case cow_http:status_to_integer(Status) of
|
||||
100 -> State0#state{expect=undefined};
|
||||
_ -> State0
|
||||
end,
|
||||
{[Inform], State};
|
||||
info(_StreamID, Response = {response, _, _, _}, State) ->
|
||||
{[Response], State#state{expect=undefined}};
|
||||
info(_StreamID, Headers = {headers, _, _}, State) ->
|
||||
{[Headers], State#state{expect=undefined}};
|
||||
info(_StreamID, Data = {data, _, _}, State) ->
|
||||
{[Data], State};
|
||||
info(_StreamID, Trailers = {trailers, _}, State) ->
|
||||
{[Trailers], State};
|
||||
info(_StreamID, Push = {push, _, _, _, _, _, _, _}, State) ->
|
||||
{[Push], State};
|
||||
info(_StreamID, SwitchProtocol = {switch_protocol, _, _, _}, State) ->
|
||||
{[SwitchProtocol], State#state{expect=undefined}};
|
||||
%% Stray message.
|
||||
info(_StreamID, _Info, State) ->
|
||||
{[], State}.
|
||||
do_info(StreamID, Inform, [Inform], State);
|
||||
info(StreamID, Response={response, _, _, _}, State) ->
|
||||
do_info(StreamID, Response, [Response], State#state{expect=undefined});
|
||||
info(StreamID, Headers={headers, _, _}, State) ->
|
||||
do_info(StreamID, Headers, [Headers], State#state{expect=undefined});
|
||||
info(StreamID, Data={data, _, _}, State) ->
|
||||
do_info(StreamID, Data, [Data], State);
|
||||
info(StreamID, Trailers={trailers, _}, State) ->
|
||||
do_info(StreamID, Trailers, [Trailers], State);
|
||||
info(StreamID, Push={push, _, _, _, _, _, _, _}, State) ->
|
||||
do_info(StreamID, Push, [Push], State);
|
||||
info(StreamID, SwitchProtocol={switch_protocol, _, _, _}, State) ->
|
||||
do_info(StreamID, SwitchProtocol, [SwitchProtocol], State#state{expect=undefined});
|
||||
%% Unknown message, either stray or meant for a handler down the line.
|
||||
info(StreamID, Info, State) ->
|
||||
do_info(StreamID, Info, [], State).
|
||||
|
||||
do_info(StreamID, Info, Commands1, State0=#state{next=Next0}) ->
|
||||
{Commands2, Next} = cowboy_stream:info(StreamID, Info, Next0),
|
||||
{Commands1 ++ Commands2, State0#state{next=Next}}.
|
||||
|
||||
-spec terminate(cowboy_stream:streamid(), cowboy_stream:reason(), #state{}) -> ok.
|
||||
terminate(_StreamID, _Reason, _State) ->
|
||||
ok.
|
||||
terminate(StreamID, Reason, #state{next=Next}) ->
|
||||
cowboy_stream:terminate(StreamID, Reason, Next).
|
||||
|
||||
-spec early_error(cowboy_stream:streamid(), cowboy_stream:reason(),
|
||||
cowboy_stream:partial_req(), Resp, cowboy:opts()) -> Resp
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue