aboutsummaryrefslogtreecommitdiffstats
path: root/src/cowboy_stream_h.erl
diff options
context:
space:
mode:
Diffstat (limited to 'src/cowboy_stream_h.erl')
-rw-r--r--src/cowboy_stream_h.erl54
1 files changed, 36 insertions, 18 deletions
diff --git a/src/cowboy_stream_h.erl b/src/cowboy_stream_h.erl
index 287fd95..5c674dd 100644
--- a/src/cowboy_stream_h.erl
+++ b/src/cowboy_stream_h.erl
@@ -34,7 +34,8 @@
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()
+ read_body_buffer = <<>> :: binary(),
+ body_length = 0 :: non_neg_integer()
}).
%% @todo For shutting down children we need to have a timeout before we terminate
@@ -54,17 +55,31 @@ init(_StreamID, Req=#{ref := Ref}, Opts) ->
%% 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{
+ read_body_ref=undefined, read_body_buffer=Buffer, body_length=BodyLen}) ->
+ {[], State#state{
+ read_body_is_fin=IsFin,
+ read_body_buffer= << Buffer/binary, Data/binary >>,
+ body_length=BodyLen + byte_size(Data)}};
+data(_StreamID, 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{
+ read_body_buffer= << Buffer/binary, Data/binary >>,
+ body_length=BodyLen + byte_size(Data)}};
data(_StreamID, IsFin, Data, State=#state{pid=Pid, read_body_ref=Ref,
- read_body_timer_ref=TRef, read_body_buffer=Buffer}) ->
+ read_body_timer_ref=TRef, read_body_buffer=Buffer, body_length=BodyLen0}) ->
+ BodyLen = BodyLen0 + byte_size(Data),
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= <<>>}}.
+ send_request_body(Pid, Ref, IsFin, BodyLen, <<Buffer/binary, Data/binary>>),
+ {[], State#state{
+ read_body_ref=undefined,
+ read_body_timer_ref=undefined,
+ read_body_buffer= <<>>,
+ body_length=BodyLen}}.
-spec info(cowboy_stream:streamid(), any(), State)
-> {cowboy_stream:commands(), State} when State::#state{}.
@@ -86,15 +101,11 @@ info(StreamID, Exit = {'EXIT', Pid, {Reason, Stacktrace}}, State=#state{ref=Ref,
{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},
+info(_StreamID, {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= <<>>}};
%% Request body, not enough to send yet.
info(StreamID, {read_body, Ref, Length, Period}, State) ->
@@ -102,8 +113,8 @@ info(StreamID, {read_body, Ref, Length, Period}, State) ->
{[{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},
+ 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};
@@ -132,6 +143,13 @@ terminate(_StreamID, _Reason, _State) ->
early_error(StreamID, Reason, PartialReq, Resp, Opts) ->
cowboy_stream:early_error(StreamID, Reason, PartialReq, Resp, Opts).
+send_request_body(Pid, Ref, nofin, _, Data) ->
+ Pid ! {request_body, Ref, nofin, Data},
+ ok;
+send_request_body(Pid, Ref, fin, BodyLen, Data) ->
+ Pid ! {request_body, Ref, fin, BodyLen, Data},
+ ok.
+
%% We use ~999999p here instead of ~w because the latter doesn't
%% support printable strings.
report_crash(_, _, _, normal, _) ->