From 789490eca8382edca34bae593e88cdc5fd4fe444 Mon Sep 17 00:00:00 2001 From: Masatake Daimon Date: Fri, 17 May 2013 13:01:20 +0900 Subject: Fix {stream, {self, once}} in httpc Previously the only difference between {stream, self} and {stream, {self, once}} was an extra Pid in the stream_start message due to a bug in httpc_handler. It was immediately sending a bunch of messages till the end instead of waiting for httpc:stream_next/1 being called. --- lib/inets/src/http_client/httpc_handler.erl | 54 +++++++++++++++++++++++++---- lib/inets/test/httpc_SUITE.erl | 9 +++++ 2 files changed, 57 insertions(+), 6 deletions(-) diff --git a/lib/inets/src/http_client/httpc_handler.erl b/lib/inets/src/http_client/httpc_handler.erl index f6b13c2998..ef16cddc81 100644 --- a/lib/inets/src/http_client/httpc_handler.erl +++ b/lib/inets/src/http_client/httpc_handler.erl @@ -65,7 +65,7 @@ options, % #options{} timers = #timers{}, % #timers{} profile_name, % atom() - id of httpc_manager process. - once % send | undefined + once = inactive % inactive | once }). @@ -231,6 +231,8 @@ init([Parent, Request, Options, ProfileName]) -> ProxyOptions = handle_proxy_options(Request#request.scheme, Options), Address = handle_proxy(Request#request.address, ProxyOptions), {ok, State} = + %% #state.once should initially be 'inactive' because we + %% activate the socket at first regardless of the state. case {Address /= Request#request.address, Request#request.scheme} of {true, https} -> connect_and_send_upgrade_request(Address, Request, @@ -425,7 +427,9 @@ handle_cast({cancel, RequestId, From}, handle_cast(stream_next, #state{session = Session} = State) -> activate_once(Session), - {noreply, State#state{once = once}}. + %% Inactivate the #state.once here because we don't want + %% next_body_chunk/1 to activate the socket twice. + {noreply, State#state{once = inactive}}. %%-------------------------------------------------------------------- @@ -478,6 +482,41 @@ handle_info({Proto, _Socket, Data}, NewMFA = {Module, whole_body, [NewBody, NewLength]}, {noreply, NewState#state{mfa = NewMFA, request = NewRequest}}; + {Module, decode_size, + [TotalChunk, HexList, + {MaxBodySize, BodySoFar, AccLength, MaxHeaderSize}]} + when BodySoFar =/= <<>> -> + ?hcrd("data processed - decode_size", []), + %% The response body is chunk-encoded. Steal decoded + %% chunks as much as possible to stream. + {_, Code, _} = StatusLine, + {NewBody, NewRequest} = stream(BodySoFar, Request, Code), + NewState = next_body_chunk(State), + NewMFA = {Module, decode_size, + [TotalChunk, HexList, + {MaxBodySize, NewBody, AccLength, MaxHeaderSize}]}, + {noreply, NewState#state{mfa = NewMFA, + request = NewRequest}}; + {Module, decode_data, + [ChunkSize, TotalChunk, + {MaxBodySize, BodySoFar, AccLength, MaxHeaderSize, false}]} + when TotalChunk =/= <<>> orelse BodySoFar =/= <<>> -> + ?hcrd("data processed - decode_data", []), + %% The response body is chunk-encoded. Steal decoded + %% chunks as much as possible to stream. + ChunkSizeToSteal = min(ChunkSize, byte_size(TotalChunk)), + <> = TotalChunk, + StolenBody = <>, + NewChunkSize = ChunkSize - ChunkSizeToSteal, + {_, Code, _} = StatusLine, + + {NewBody, NewRequest} = stream(StolenBody, Request, Code), + NewState = next_body_chunk(State), + NewMFA = {Module, decode_data, + [NewChunkSize, NewTotalChunk, + {MaxBodySize, NewBody, AccLength, MaxHeaderSize, false}]}, + {noreply, NewState#state{mfa = NewMFA, + request = NewRequest}}; NewMFA -> ?hcrd("data processed - new mfa", []), activate_once(Session), @@ -1027,11 +1066,15 @@ handle_http_msg({Version, StatusCode, ReasonPharse, Headers, Body}, status_line = StatusLine, headers = Headers}) end; -handle_http_msg({ChunkedHeaders, Body}, #state{headers = Headers} = State) -> +handle_http_msg({ChunkedHeaders, Body}, + #state{status_line = {_, Code, _}, headers = Headers} = State) -> ?hcrt("handle_http_msg", [{chunked_headers, ChunkedHeaders}, {headers, Headers}]), NewHeaders = http_chunk:handle_headers(Headers, ChunkedHeaders), - handle_response(State#state{headers = NewHeaders, body = Body}); + {NewBody, NewRequest} = stream(Body, State#state.request, Code), + handle_response(State#state{headers = NewHeaders, + body = NewBody, + request = NewRequest}); handle_http_msg(Body, #state{status_line = {_,Code, _}} = State) -> ?hcrt("handle_http_msg", [{code, Code}]), {NewBody, NewRequest} = stream(Body, State#state.request, Code), @@ -1070,8 +1113,7 @@ handle_http_body(Body, #state{headers = Headers, "chunked" -> ?hcrt("handle_http_body - chunked", []), case http_chunk:decode(Body, State#state.max_body_size, - State#state.max_header_size, - {Code, Request}) of + State#state.max_header_size) of {Module, Function, Args} -> ?hcrt("handle_http_body - new mfa", [{module, Module}, diff --git a/lib/inets/test/httpc_SUITE.erl b/lib/inets/test/httpc_SUITE.erl index 350192464e..0c35f284f7 100644 --- a/lib/inets/test/httpc_SUITE.erl +++ b/lib/inets/test/httpc_SUITE.erl @@ -1693,6 +1693,15 @@ receive_streamed_body(RequestId, Body, Pid) -> ct:print("~p:receive_streamed_body -> requested next stream ~n", [?MODULE]), receive {http, {RequestId, stream, BinBodyPart}} -> + %% Make sure the httpc hasn't sent us the next 'stream' + %% without our request. + receive + {http, {RequestId, stream, _}} = Msg -> + ct:fail({unexpected_flood_of_stream, Msg}) + after + 1000 -> + ok + end, receive_streamed_body(RequestId, <>, Pid); -- cgit v1.2.3 From e177b41eb2e4ce0996e323d8fd08d18f94b19964 Mon Sep 17 00:00:00 2001 From: Masatake Daimon Date: Fri, 17 May 2013 13:28:47 +0900 Subject: Remove http_chunk:decode/4 as it is no longer used http_chunk:decode/4, the side-effecting chunk decoder, was only used by httpc_handler:handle_http_body/2 but now it's completely unused. --- lib/inets/src/http_client/httpc_handler.erl | 5 ++- lib/inets/src/http_lib/http_chunk.erl | 53 ++++++++--------------------- 2 files changed, 17 insertions(+), 41 deletions(-) diff --git a/lib/inets/src/http_client/httpc_handler.erl b/lib/inets/src/http_client/httpc_handler.erl index ef16cddc81..55794f57dc 100644 --- a/lib/inets/src/http_client/httpc_handler.erl +++ b/lib/inets/src/http_client/httpc_handler.erl @@ -33,7 +33,6 @@ %% connect_and_send/2, send/2, cancel/3, - stream/3, stream_next/1, info/1 ]). @@ -499,7 +498,7 @@ handle_info({Proto, _Socket, Data}, request = NewRequest}}; {Module, decode_data, [ChunkSize, TotalChunk, - {MaxBodySize, BodySoFar, AccLength, MaxHeaderSize, false}]} + {MaxBodySize, BodySoFar, AccLength, MaxHeaderSize}]} when TotalChunk =/= <<>> orelse BodySoFar =/= <<>> -> ?hcrd("data processed - decode_data", []), %% The response body is chunk-encoded. Steal decoded @@ -514,7 +513,7 @@ handle_info({Proto, _Socket, Data}, NewState = next_body_chunk(State), NewMFA = {Module, decode_data, [NewChunkSize, NewTotalChunk, - {MaxBodySize, NewBody, AccLength, MaxHeaderSize, false}]}, + {MaxBodySize, NewBody, AccLength, MaxHeaderSize}]}, {noreply, NewState#state{mfa = NewMFA, request = NewRequest}}; NewMFA -> diff --git a/lib/inets/src/http_lib/http_chunk.erl b/lib/inets/src/http_lib/http_chunk.erl index 57647438e9..24c939e80c 100644 --- a/lib/inets/src/http_lib/http_chunk.erl +++ b/lib/inets/src/http_lib/http_chunk.erl @@ -24,7 +24,7 @@ -include("http_internal.hrl"). %% API --export([decode/3, decode/4, encode/1, encode_last/0, handle_headers/2]). +-export([decode/3, encode/1, encode_last/0, handle_headers/2]). %% Callback API - used for example if the chunkedbody is received a %% little at a time on a socket. -export([decode_size/1, ignore_extensions/1, decode_data/1, decode_trailer/1]). @@ -34,20 +34,14 @@ %%% API %%%========================================================================= %%------------------------------------------------------------------------- -%% decode(ChunkedBody, MaxBodySize, MaxHeaderSize, ) -> +%% decode(ChunkedBody, MaxBodySize, MaxHeaderSize) -> %% {ok, {Headers, Body}} | {Module, Function, Args} %% %% Headers = ["Header:Value"] %% ChunkedBody = binary() %% MaxBodySize = integer() %% MaxHeaderSize = integer() -%% Stream = {Code, Request} - if Request#request.stream =/= none -%% and Code == 200 the side effect of sending each decode chunk to the -%% client/file before the whole body is received will take place. %% -%% Note: decode/4 should only be used from httpc_handler module. -%% Otherwhise use the side effect free decode/3. -%% %% Description: Decodes a body encoded by the chunked transfer %% encoding. If the ChunkedBody is not compleate it returns {Module, %% Function, Args} so that decoding can be continued when more of the @@ -61,12 +55,9 @@ %% the next pass in the loop. %%------------------------------------------------------------------------- decode(ChunkedBody, MaxBodySize, MaxHeaderSize) -> - decode(ChunkedBody, MaxBodySize, MaxHeaderSize, false). - -decode(ChunkedBody, MaxBodySize, MaxHeaderSize, Stream) -> %% Note decode_size will call decode_data. - decode_size([ChunkedBody, <<>>, [], - {MaxBodySize, <<>>, 0, MaxHeaderSize, Stream}]). + decode_size([ChunkedBody, <<>>, [], + {MaxBodySize, <<>>, 0, MaxHeaderSize}]). %%------------------------------------------------------------------------- %% encode(Chunk) -> EncodedChunk @@ -150,7 +141,7 @@ decode_size(<<>>, HexList, Info) -> decode_size(Data = <>, HexList, {MaxBodySize, Body, AccLength, - MaxHeaderSize, Stream}) -> + MaxHeaderSize}) -> ChunkSize = http_util:hexlist_to_integer(lists:reverse(HexList)), case ChunkSize of 0 -> % Last chunk, there was no data @@ -164,7 +155,7 @@ decode_size(Data = <>, HexList, %% to this function comes in. decode_data(ChunkSize, ChunkRest, {MaxBodySize, Body, ChunkSize + AccLength , - MaxHeaderSize, Stream}) + MaxHeaderSize}) end; decode_size(<<";", Rest/binary>>, HexList, Info) -> %% Note ignore_extensions will call decode_size/1 again when @@ -189,50 +180,42 @@ ignore_extensions(<<_Octet, Rest/binary>>, NextFunction) -> ignore_extensions(Rest, NextFunction). decode_data(ChunkSize, TotalChunk, - Info = {MaxBodySize, BodySoFar, AccLength, MaxHeaderSize, Stream}) + Info = {MaxBodySize, BodySoFar, AccLength, MaxHeaderSize}) when ChunkSize =< size(TotalChunk) -> case TotalChunk of %% Last chunk <> -> %% Note ignore_extensions will call decode_trailer/1 %% once it ignored all extensions. - {NewBody, _} = - stream(<>, Stream), {?MODULE, ignore_extensions, [<<>>, {?MODULE, decode_trailer, [<<>>, [],[], MaxHeaderSize, - NewBody, + <>, integer_to_list(AccLength)]}]}; <> -> %% Note ignore_extensions will call decode_trailer/1 %% once it ignored all extensions. - {NewBody, _} = stream(<>, Stream), ignore_extensions(Rest, {?MODULE, decode_trailer, [<<>>, [],[], MaxHeaderSize, - NewBody, + <>, integer_to_list(AccLength)]}); <> -> - {NewBody, _} = stream(<>, Stream), {?MODULE, decode_trailer, [<>, [],[], MaxHeaderSize, - NewBody, + <>, integer_to_list(AccLength)]}; <> -> - {NewBody,_}= stream(<>, Stream), decode_trailer(<>, [],[], MaxHeaderSize, - NewBody, + <>, integer_to_list(AccLength)); %% There are more chunks, so here we go agin... <> -> - {NewBody, NewStream} = - stream(<>, Stream), - {?MODULE, decode_size, [<<>>, [], {MaxBodySize, NewBody, AccLength, MaxHeaderSize, NewStream}]}; + NewBody = <>, + {?MODULE, decode_size, [<<>>, [], {MaxBodySize, NewBody, AccLength, MaxHeaderSize}]}; <> when (AccLength < MaxBodySize) or (MaxBodySize == nolimit) -> - {NewBody, NewStream} = - stream(<>, Stream), decode_size(Rest, [], - {MaxBodySize, NewBody, - AccLength, MaxHeaderSize, NewStream}); + {MaxBodySize, <>, + AccLength, MaxHeaderSize}); <<_:ChunkSize/binary, ?CR, ?LF, _/binary>> -> throw({error, body_too_big}); _ -> @@ -286,9 +269,3 @@ decode_trailer(<>, Header, Headers, MaxHeaderSize, Body, BodyLength) -> decode_trailer(Rest, [Octet | Header], Headers, MaxHeaderSize, Body, BodyLength). - -stream(BodyPart, false) -> - {BodyPart, false}; -stream(BodyPart, {Code, Request}) -> - {NewBody, NewRequest} = httpc_handler:stream(BodyPart, Request, Code), - {NewBody, {Code, NewRequest}}. -- cgit v1.2.3