From 32db544782f2528ed0916eecb200f75924dcc407 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Hoguin?= Date: Thu, 27 Apr 2017 15:23:57 +0200 Subject: Add content handlers and built-in SSE support Content handlers are a chain of modules implementing callbacks that receive the body of responses and may modify it (for example for decompressing the content) or act upon it (like sending a message to the owner process. The gun_sse content handler module can be used to translate text/event-stream events on the fly and deliver them to the owner process as a {gun_sse...} message. This feature is currently not documented and is only tested against a public server. It requires an up to date Cowlib. --- src/gun_http.erl | 111 +++++++++++++++++++++++++++++++++++-------------------- 1 file changed, 70 insertions(+), 41 deletions(-) (limited to 'src/gun_http.erl') diff --git a/src/gun_http.erl b/src/gun_http.erl index 65135f9..f5eef71 100644 --- a/src/gun_http.erl +++ b/src/gun_http.erl @@ -31,17 +31,25 @@ -type websocket_info() :: {websocket, reference(), binary(), [binary()], [], gun:ws_opts()}. %% key, extensions, protocols, options +-record(stream, { + ref :: reference() | websocket_info(), + method :: binary(), + is_alive :: boolean(), + handler_state :: undefined | gun_content_handler:state() +}). + -record(http_state, { owner :: pid(), socket :: inet:socket() | ssl:sslsocket(), transport :: module(), version = 'HTTP/1.1' :: cow_http:version(), + content_handlers :: gun_content_handler:opt(), connection = keepalive :: keepalive | close, buffer = <<>> :: binary(), %% Stream reference, request method and whether the stream is alive. - streams = [] :: [{reference() | websocket_info(), binary(), boolean()}], + streams = [] :: [#stream{}], in = head :: io(), - in_state :: {non_neg_integer(), non_neg_integer()}, + in_state = {0, 0} :: {non_neg_integer(), non_neg_integer()}, out = head :: io() }). @@ -54,6 +62,11 @@ do_check_options([{keepalive, K}|Opts]) when is_integer(K), K > 0 -> do_check_options(Opts); do_check_options([{version, V}|Opts]) when V =:= 'HTTP/1.1'; V =:= 'HTTP/1.0' -> do_check_options(Opts); +do_check_options([Opt={content_handlers, Handlers}|Opts]) -> + case gun_content_handler:check_option(Handlers) of + ok -> do_check_options(Opts); + error -> {error, {options, {http, Opt}}} + end; do_check_options([Opt|_]) -> {error, {options, {http, Opt}}}. @@ -61,7 +74,9 @@ name() -> http. init(Owner, Socket, Transport, Opts) -> Version = maps:get(version, Opts, 'HTTP/1.1'), - #http_state{owner=Owner, socket=Socket, transport=Transport, version=Version}. + Handlers = maps:get(content_handlers, Opts, [gun_data]), + #http_state{owner=Owner, socket=Socket, transport=Transport, version=Version, + content_handlers=Handlers}. %% Stop looping when we got no more data. handle(<<>>, State) -> @@ -78,8 +93,7 @@ handle(Data, State=#http_state{in=head, buffer=Buffer}) -> end; %% Everything sent to the socket until it closes is part of the response body. handle(Data, State=#http_state{in=body_close}) -> - send_data_if_alive(Data, State, nofin), - State; + send_data_if_alive(Data, State, nofin); handle(Data, State=#http_state{in=body_chunked, in_state=InState, buffer=Buffer, connection=Conn}) -> Buffer2 = << Buffer/binary, Data/binary >>, @@ -87,30 +101,33 @@ handle(Data, State=#http_state{in=body_chunked, in_state=InState, more -> State#http_state{buffer=Buffer2}; {more, Data2, InState2} -> - send_data_if_alive(Data2, State, nofin), - State#http_state{buffer= <<>>, in_state=InState2}; + send_data_if_alive(Data2, + State#http_state{buffer= <<>>, in_state=InState2}, + nofin); {more, Data2, Length, InState2} when is_integer(Length) -> %% @todo See if we can recv faster than one message at a time. - send_data_if_alive(Data2, State, nofin), - State#http_state{buffer= <<>>, in_state=InState2}; + send_data_if_alive(Data2, + State#http_state{buffer= <<>>, in_state=InState2}, + nofin); {more, Data2, Rest, InState2} -> %% @todo See if we can recv faster than one message at a time. - send_data_if_alive(Data2, State, nofin), - State#http_state{buffer=Rest, in_state=InState2}; + send_data_if_alive(Data2, + State#http_state{buffer=Rest, in_state=InState2}, + nofin); {done, _TotalLength, Rest} -> %% I suppose it doesn't hurt to append an empty binary. - send_data_if_alive(<<>>, State, fin), + State1 = send_data_if_alive(<<>>, State, fin), case Conn of keepalive -> - handle(Rest, end_stream(State#http_state{buffer= <<>>})); + handle(Rest, end_stream(State1#http_state{buffer= <<>>})); close -> close end; {done, Data2, _TotalLength, Rest} -> - send_data_if_alive(Data2, State, fin), + State1 = send_data_if_alive(Data2, State, fin), case Conn of keepalive -> - handle(Rest, end_stream(State#http_state{buffer= <<>>})); + handle(Rest, end_stream(State1#http_state{buffer= <<>>})); close -> close end @@ -121,27 +138,29 @@ handle(Data, State=#http_state{in={body, Length}, connection=Conn}) -> if %% More data coming. DataSize < Length -> - send_data_if_alive(Data, State, nofin), - State#http_state{in={body, Length - DataSize}}; + send_data_if_alive(Data, + State#http_state{in={body, Length - DataSize}}, + nofin); %% Stream finished, no rest. DataSize =:= Length -> - send_data_if_alive(Data, State, fin), + State1 = send_data_if_alive(Data, State, fin), case Conn of - keepalive -> end_stream(State); + keepalive -> end_stream(State1); close -> close end; %% Stream finished, rest. true -> << Body:Length/binary, Rest/bits >> = Data, - send_data_if_alive(Body, State, fin), + State1 = send_data_if_alive(Body, State, fin), case Conn of - keepalive -> handle(Rest, end_stream(State)); + keepalive -> handle(Rest, end_stream(State1)); close -> close end end. handle_head(Data, State=#http_state{owner=Owner, version=ClientVersion, - connection=Conn, streams=[{StreamRef, Method, IsAlive}|_]}) -> + content_handlers=Handlers0, connection=Conn, + streams=[Stream=#stream{ref=StreamRef, method=Method, is_alive=IsAlive}|Tail]}) -> {Version, Status, _, Rest} = cow_http:parse_status_line(Data), {Headers, Rest2} = cow_http:parse_headers(Rest), case {Status, StreamRef} of @@ -150,7 +169,7 @@ handle_head(Data, State=#http_state{owner=Owner, version=ClientVersion, _ -> In = response_io_from_headers(Method, Version, Status, Headers), IsFin = case In of head -> fin; _ -> nofin end, - case IsAlive of + Handlers = case IsAlive of false -> ok; true -> @@ -160,7 +179,13 @@ handle_head(Data, State=#http_state{owner=Owner, version=ClientVersion, end, Owner ! {gun_response, self(), StreamRef2, IsFin, Status, Headers}, - ok + %% @todo Change to ReplyTo. + case IsFin of + fin -> undefined; + nofin -> + gun_content_handler:init(Owner, StreamRef, + Status, Headers, Handlers0) + end end, Conn2 = if Conn =:= close -> close; @@ -174,33 +199,36 @@ handle_head(Data, State=#http_state{owner=Owner, version=ClientVersion, close; IsFin =:= fin -> handle(Rest2, end_stream(State#http_state{in=In, - in_state={0, 0}, connection=Conn2})); + in_state={0, 0}, connection=Conn2, + streams=[Stream#stream{handler_state=Handlers}|Tail]})); true -> - handle(Rest2, State#http_state{in=In, in_state={0, 0}, connection=Conn2}) + handle(Rest2, State#http_state{in=In, + in_state={0, 0}, connection=Conn2, + streams=[Stream#stream{handler_state=Handlers}|Tail]}) end end. -send_data_if_alive(<<>>, _, nofin) -> - ok; +send_data_if_alive(<<>>, State, nofin) -> + State; %% @todo What if we receive data when the HEAD method was used? -send_data_if_alive(Data, #http_state{owner=Owner, - streams=[{StreamRef, _, true}|_]}, IsFin) -> - Owner ! {gun_data, self(), StreamRef, IsFin, Data}, - ok; -send_data_if_alive(_, _, _) -> - ok. +send_data_if_alive(Data, State=#http_state{streams=[Stream=#stream{ + is_alive=true, handler_state=Handlers0}|Tail]}, IsFin) -> + Handlers = gun_content_handler:handle(IsFin, Data, Handlers0), + State#http_state{streams=[Stream#stream{handler_state=Handlers}|Tail]}; +send_data_if_alive(_, State, _) -> + State. close(State=#http_state{in=body_close, owner=Owner, streams=[_|Tail]}) -> - send_data_if_alive(<<>>, State, fin), + _ = send_data_if_alive(<<>>, State, fin), close_streams(Owner, Tail); close(#http_state{owner=Owner, streams=Streams}) -> close_streams(Owner, Streams). close_streams(_, []) -> ok; -close_streams(Owner, [{_, _, false}|Tail]) -> +close_streams(Owner, [#stream{is_alive=false}|Tail]) -> close_streams(Owner, Tail); -close_streams(Owner, [{StreamRef, _, _}|Tail]) -> +close_streams(Owner, [#stream{ref=StreamRef}|Tail]) -> Owner ! {gun_error, self(), StreamRef, {closed, "The connection was lost."}}, close_streams(Owner, Tail). @@ -256,7 +284,7 @@ data(State=#http_state{streams=[]}, StreamRef, _, _) -> data(State=#http_state{socket=Socket, transport=Transport, version=Version, out=Out, streams=Streams}, StreamRef, IsFin, Data) -> case lists:last(Streams) of - {StreamRef, _, true} -> + #stream{ref=StreamRef, is_alive=true} -> DataLength = iolist_size(Data), case Out of body_chunked when Version =:= 'HTTP/1.1', IsFin =:= fin -> @@ -304,7 +332,7 @@ down(#http_state{streams=Streams}) -> KilledStreams = [case Ref of {websocket, Ref2, _, _, _, _} -> Ref2; _ -> Ref - end || {Ref, _, _} <- Streams], + end || #stream{ref=Ref} <- Streams], {KilledStreams, []}. error_stream_closed(State=#http_state{owner=Owner}, StreamRef) -> @@ -373,7 +401,8 @@ response_io_from_headers(_, Version, _Status, Headers) -> %% Streams. new_stream(State=#http_state{streams=Streams}, StreamRef, Method) -> - State#http_state{streams=Streams ++ [{StreamRef, iolist_to_binary(Method), true}]}. + State#http_state{streams=Streams + ++ [#stream{ref=StreamRef, method=iolist_to_binary(Method), is_alive=true}]}. is_stream(#http_state{streams=Streams}, StreamRef) -> lists:keymember(StreamRef, 1, Streams). @@ -384,7 +413,7 @@ cancel_stream(State=#http_state{streams=Streams}, StreamRef) -> {Ref, false}; _ -> Tuple - end || Tuple = {Ref, _, _} <- Streams], + end || Tuple = #stream{ref=Ref} <- Streams], State#http_state{streams=Streams2}. end_stream(State=#http_state{streams=[_|Tail]}) -> -- cgit v1.2.3