%% Copyright (c) 2015-2017, Loïc Hoguin %% %% 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_http2). -export([init/5]). -export([init/9]). -export([init/11]). -export([system_continue/3]). -export([system_terminate/4]). -export([system_code_change/4]). -type opts() :: #{ connection_type => worker | supervisor, env => cowboy_middleware:env(), inactivity_timeout => timeout(), middlewares => [module()], preface_timeout => timeout(), shutdown_timeout => timeout(), stream_handlers => [module()] }. -export_type([opts/0]). -record(stream, { id = undefined :: cowboy_stream:streamid(), %% Stream handlers and their state. state = undefined :: {module(), any()} | flush, %% Request method. method = undefined :: binary(), %% Whether we finished sending data. local = idle :: idle | upgrade | cowboy_stream:fin() | flush, %% Local flow control window (how much we can send). local_window :: integer(), %% Buffered data waiting for the flow control window to increase. local_buffer = queue:new() :: queue:queue( {cowboy_stream:fin(), non_neg_integer(), iolist() | {sendfile, non_neg_integer(), pos_integer(), file:name_all()}}), local_buffer_size = 0 :: non_neg_integer(), local_trailers = undefined :: undefined | cowboy:http_headers(), %% Whether we finished receiving data. remote = nofin :: cowboy_stream:fin(), %% Remote flow control window (how much we accept to receive). remote_window :: integer(), %% Unparsed te header. Used to know if we can send trailers. te :: undefined | binary() }). -type stream() :: #stream{}. -record(state, { parent = undefined :: pid(), ref :: ranch:ref(), socket = undefined :: inet:socket(), transport :: module(), opts = #{} :: opts(), %% Remote address and port for the connection. peer = undefined :: {inet:ip_address(), inet:port_number()}, %% Local address and port for the connection. sock = undefined :: {inet:ip_address(), inet:port_number()}, %% Client certificate (TLS only). cert :: undefined | binary(), %% Settings are separate for each endpoint. In addition, settings %% must be acknowledged before they can be expected to be applied. %% %% @todo Since the ack is required, we must timeout if we don't receive it. %% @todo I haven't put as much thought as I should have on this, %% the final settings handling will be very different. local_settings = #{ % header_table_size => 4096, % enable_push => false, %% We are the server. Push is never enabled. % max_concurrent_streams => infinity, initial_window_size => 65535, max_frame_size => 16384 % max_header_list_size => infinity } :: map(), %% @todo We need a TimerRef to do SETTINGS_TIMEOUT errors. %% We need to be careful there. It's well possible that we send %% two SETTINGS frames before we receive a SETTINGS ack. next_settings = #{} :: undefined | map(), %% @todo perhaps set to undefined by default remote_settings = #{ initial_window_size => 65535 } :: map(), %% Connection-wide flow control window. local_window = 65535 :: integer(), %% How much we can send. remote_window = 65535 :: integer(), %% How much we accept to receive. %% Stream identifiers. client_streamid = 0 :: non_neg_integer(), server_streamid = 2 :: pos_integer(), %% Currently active HTTP/2 streams. Streams may be initiated either %% by the client or by the server through PUSH_PROMISE frames. streams = [] :: [stream()], %% HTTP/2 streams that have been reset recently. We are expected %% to keep receiving additional frames after sending an RST_STREAM. lingering_streams = [] :: [cowboy_stream:streamid()], %% Streams can spawn zero or more children which are then managed %% by this module if operating as a supervisor. children = cowboy_children:init() :: cowboy_children:children(), %% The client starts by sending a sequence of bytes as a preface, %% followed by a potentially empty SETTINGS frame. Then the connection %% is established and continues normally. An exception is when a HEADERS %% frame is sent followed by CONTINUATION frames: no other frame can be %% sent in between. parse_state = undefined :: {preface, sequence, undefined | reference()} | {preface, settings, reference()} | normal | {continuation, cowboy_stream:streamid(), cowboy_stream:fin(), binary()}, %% HPACK decoding and encoding state. decode_state = cow_hpack:init() :: cow_hpack:state(), encode_state = cow_hpack:init() :: cow_hpack:state() }). -spec init(pid(), ranch:ref(), inet:socket(), module(), cowboy:opts()) -> ok. init(Parent, Ref, Socket, Transport, Opts) -> Peer0 = Transport:peername(Socket), Sock0 = Transport:sockname(Socket), Cert1 = case Transport:name() of ssl -> case ssl:peercert(Socket) of {error, no_peercert} -> {ok, undefined}; Cert0 -> Cert0 end; _ -> {ok, undefined} end, case {Peer0, Sock0, Cert1} of {{ok, Peer}, {ok, Sock}, {ok, Cert}} -> init(Parent, Ref, Socket, Transport, Opts, Peer, Sock, Cert, <<>>); {{error, Reason}, _, _} -> terminate(undefined, {socket_error, Reason, 'A socket error occurred when retrieving the peer name.'}); {_, {error, Reason}, _} -> terminate(undefined, {socket_error, Reason, 'A socket error occurred when retrieving the sock name.'}); {_, _, {error, Reason}} -> terminate(undefined, {socket_error, Reason, 'A socket error occurred when retrieving the client TLS certificate.'}) end. -spec init(pid(), ranch:ref(), inet:socket(), module(), cowboy:opts(), {inet:ip_address(), inet:port_number()}, {inet:ip_address(), inet:port_number()}, binary() | undefined, binary()) -> ok. init(Parent, Ref, Socket, Transport, Opts, Peer, Sock, Cert, Buffer) -> State0 = #state{parent=Parent, ref=Ref, socket=Socket, transport=Transport, opts=Opts, peer=Peer, sock=Sock, cert=Cert, parse_state={preface, sequence, preface_timeout(Opts)}}, State = settings_init(State0, Opts), preface(State), case Buffer of <<>> -> before_loop(State, Buffer); _ -> parse(State, Buffer) end. %% @todo Add an argument for the request body. -spec init(pid(), ranch:ref(), inet:socket(), module(), cowboy:opts(), {inet:ip_address(), inet:port_number()}, {inet:ip_address(), inet:port_number()}, binary() | undefined, binary(), map() | undefined, cowboy_req:req()) -> ok. init(Parent, Ref, Socket, Transport, Opts, Peer, Sock, Cert, Buffer, _Settings, Req) -> State0 = #state{parent=Parent, ref=Ref, socket=Socket, transport=Transport, opts=Opts, peer=Peer, sock=Sock, cert=Cert, parse_state={preface, sequence, preface_timeout(Opts)}}, %% @todo Apply settings. %% StreamID from HTTP/1.1 Upgrade requests is always 1. %% The stream is always in the half-closed (remote) state. State1 = stream_handler_init(State0, 1, fin, upgrade, Req), %% We assume that the upgrade will be applied. A stream handler %% must not prevent the normal operations of the server. State2 = info(State1, 1, {switch_protocol, #{ <<"connection">> => <<"Upgrade">>, <<"upgrade">> => <<"h2c">> }, ?MODULE, undefined}), %% @todo undefined or #{}? State = settings_init(State2, Opts), preface(State), case Buffer of <<>> -> before_loop(State, Buffer); _ -> parse(State, Buffer) end. settings_init(State=#state{next_settings=Settings}, Opts) -> EnableConnectProtocol = maps:get(enable_connect_protocol, Opts, false), State#state{next_settings=Settings#{enable_connect_protocol => EnableConnectProtocol}}. preface(#state{socket=Socket, transport=Transport, next_settings=Settings}) -> %% We send next_settings and use defaults until we get a ack. Transport:send(Socket, cow_http2:settings(Settings)). preface_timeout(Opts) -> case maps:get(preface_timeout, Opts, 5000) of infinity -> undefined; PrefaceTimeout -> erlang:start_timer(PrefaceTimeout, self(), preface_timeout) end. %% @todo Add the timeout for last time since we heard of connection. before_loop(State, Buffer) -> loop(State, Buffer). loop(State=#state{parent=Parent, socket=Socket, transport=Transport, opts=Opts, children=Children, parse_state=PS}, Buffer) -> Transport:setopts(Socket, [{active, once}]), {OK, Closed, Error} = Transport:messages(), InactivityTimeout = maps:get(inactivity_timeout, Opts, 300000), receive %% Socket messages. {OK, Socket, Data} -> parse(State, << Buffer/binary, Data/binary >>); {Closed, Socket} -> terminate(State, {socket_error, closed, 'The socket has been closed.'}); {Error, Socket, Reason} -> terminate(State, {socket_error, Reason, 'An error has occurred on the socket.'}); %% System messages. {'EXIT', Parent, Reason} -> %% @todo We should exit gracefully. exit(Reason); {system, From, Request} -> sys:handle_system_msg(Request, From, Parent, ?MODULE, [], {State, Buffer}); %% Timeouts. {timeout, Ref, {shutdown, Pid}} -> cowboy_children:shutdown_timeout(Children, Ref, Pid), loop(State, Buffer); {timeout, TRef, preface_timeout} -> case PS of {preface, _, TRef} -> terminate(State, {connection_error, protocol_error, 'The preface was not received in a reasonable amount of time.'}); _ -> loop(State, Buffer) end; %% Messages pertaining to a stream. {{Pid, StreamID}, Msg} when Pid =:= self() -> loop(info(State, StreamID, Msg), Buffer); %% Exit signal from children. Msg = {'EXIT', Pid, _} -> loop(down(State, Pid, Msg), Buffer); %% Calls from supervisor module. {'$gen_call', From, Call} -> cowboy_children:handle_supervisor_call(Call, From, Children, ?MODULE), loop(State, Buffer); Msg -> error_logger:error_msg("Received stray message ~p.", [Msg]), loop(State, Buffer) after InactivityTimeout -> terminate(State, {internal_error, timeout, 'No message or data received before timeout.'}) end. parse(State=#state{socket=Socket, transport=Transport, parse_state={preface, sequence, TRef}}, Data) -> case Data of << "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n", Rest/bits >> -> parse(State#state{parse_state={preface, settings, TRef}}, Rest); _ when byte_size(Data) >= 24 -> Transport:close(Socket), exit({shutdown, {connection_error, protocol_error, 'The connection preface was invalid. (RFC7540 3.5)'}}); _ -> Len = byte_size(Data), << Preface:Len/binary, _/bits >> = <<"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n">>, case Data of Preface -> before_loop(State, Data); _ -> Transport:close(Socket), exit({shutdown, {connection_error, protocol_error, 'The connection preface was invalid. (RFC7540 3.5)'}}) end end; %% @todo Perhaps instead of just more we can have {more, Len} to avoid all the checks. parse(State=#state{local_settings=#{max_frame_size := MaxFrameSize}, parse_state=ParseState}, Data) -> case cow_http2:parse(Data, MaxFrameSize) of {ok, Frame, Rest} -> case ParseState of normal -> parse(frame(State, Frame), Rest); {preface, settings, TRef} -> parse_settings_preface(State, Frame, Rest, TRef); {continuation, _, _, _} -> parse(continuation_frame(State, Frame), Rest) end; {ignore, _} when element(1, ParseState) =:= continuation -> terminate(State, {connection_error, protocol_error, 'An invalid frame was received in the middle of a header block. (RFC7540 6.2)'}); {ignore, Rest} -> parse(State, Rest); {stream_error, StreamID, Reason, Human, Rest} -> parse(stream_reset(State, StreamID, {stream_error, Reason, Human}), Rest); Error = {connection_error, _, _} -> terminate(State, Error); more -> before_loop(State, Data) end. parse_settings_preface(State, Frame={settings, _}, Rest, TRef) -> ok = case TRef of undefined -> ok; _ -> erlang:cancel_timer(TRef, [{async, true}, {info, false}]) end, parse(frame(State#state{parse_state=normal}, Frame), Rest); parse_settings_preface(State, _, _, _) -> terminate(State, {connection_error, protocol_error, 'The preface sequence must be followed by a SETTINGS frame. (RFC7540 3.5)'}). %% @todo When we get a 'fin' we need to check if the stream had a 'fin' sent back %% and terminate the stream if this is the end of it. %% DATA frame. frame(State=#state{client_streamid=LastStreamID}, {data, StreamID, _, _}) when StreamID > LastStreamID -> terminate(State, {connection_error, protocol_error, 'DATA frame received on a stream in idle state. (RFC7540 5.1)'}); frame(State0=#state{remote_window=ConnWindow, streams=Streams, lingering_streams=Lingering}, {data, StreamID, IsFin, Data}) -> DataLen = byte_size(Data), State = State0#state{remote_window=ConnWindow - DataLen}, case lists:keyfind(StreamID, #stream.id, Streams) of Stream = #stream{state=flush, remote=nofin, remote_window=StreamWindow} -> after_commands(State, Stream#stream{remote=IsFin, remote_window=StreamWindow - DataLen}); Stream = #stream{state=StreamState0, remote=nofin, remote_window=StreamWindow} -> try cowboy_stream:data(StreamID, IsFin, Data, StreamState0) of {Commands, StreamState} -> commands(State, Stream#stream{state=StreamState, remote=IsFin, remote_window=StreamWindow - DataLen}, Commands) catch Class:Exception -> cowboy_stream:report_error(data, [StreamID, IsFin, Data, StreamState0], Class, Exception, erlang:get_stacktrace()), stream_reset(State, StreamID, {internal_error, {Class, Exception}, 'Unhandled exception in cowboy_stream:data/4.'}) end; #stream{remote=fin} -> stream_reset(State, StreamID, {stream_error, stream_closed, 'DATA frame received for a half-closed (remote) stream. (RFC7540 5.1)'}); false -> %% After we send an RST_STREAM frame and terminate a stream, %% the client still might be sending us some more frames %% until it can process this RST_STREAM. We therefore ignore %% DATA frames received for such lingering streams. case lists:member(StreamID, Lingering) of true -> State0; false -> terminate(State, {connection_error, stream_closed, 'DATA frame received for a closed stream. (RFC7540 5.1)'}) end end; %% HEADERS frame with invalid even-numbered streamid. frame(State, {headers, StreamID, _, _, _}) when StreamID rem 2 =:= 0 -> terminate(State, {connection_error, protocol_error, 'HEADERS frame received with even-numbered streamid. (RFC7540 5.1.1)'}); %% HEADERS frame received on (half-)closed stream. %% %% We always close the connection here to avoid having to decode %% the headers to not waste resources on non-compliant clients. frame(State=#state{client_streamid=LastStreamID}, {headers, StreamID, _, _, _}) when StreamID =< LastStreamID -> terminate(State, {connection_error, stream_closed, 'HEADERS frame received on a stream in closed or half-closed state. (RFC7540 5.1)'}); %% Single HEADERS frame headers block. frame(State, {headers, StreamID, IsFin, head_fin, HeaderBlock}) -> %% @todo We probably need to validate StreamID here and in 4 next clauses. stream_decode_init(State, StreamID, IsFin, HeaderBlock); %% HEADERS frame starting a headers block. Enter continuation mode. frame(State, {headers, StreamID, IsFin, head_nofin, HeaderBlockFragment}) -> State#state{parse_state={continuation, StreamID, IsFin, HeaderBlockFragment}}; %% Single HEADERS frame headers block with priority. frame(State, {headers, StreamID, IsFin, head_fin, _IsExclusive, _DepStreamID, _Weight, HeaderBlock}) -> %% @todo Handle priority. stream_decode_init(State, StreamID, IsFin, HeaderBlock); %% HEADERS frame starting a headers block. Enter continuation mode. frame(State, {headers, StreamID, IsFin, head_nofin, _IsExclusive, _DepStreamID, _Weight, HeaderBlockFragment}) -> %% @todo Handle priority. State#state{parse_state={continuation, StreamID, IsFin, HeaderBlockFragment}}; %% PRIORITY frame. frame(State, {priority, _StreamID, _IsExclusive, _DepStreamID, _Weight}) -> %% @todo Validate StreamID? %% @todo Handle priority. State; %% RST_STREAM frame. frame(State=#state{client_streamid=LastStreamID}, {rst_stream, StreamID, _}) when StreamID > LastStreamID -> terminate(State, {connection_error, protocol_error, 'RST_STREAM frame received on a stream in idle state. (RFC7540 5.1)'}); frame(State, {rst_stream, StreamID, Reason}) -> stream_terminate(State, StreamID, {stream_error, Reason, 'Stream reset requested by client.'}); %% SETTINGS frame. frame(State0=#state{socket=Socket, transport=Transport, remote_settings=Settings0}, {settings, Settings}) -> Transport:send(Socket, cow_http2:settings_ack()), State = State0#state{remote_settings=maps:merge(Settings0, Settings)}, case Settings of #{initial_window_size := NewWindowSize} -> OldWindowSize = maps:get(initial_window_size, Settings0, 65535), update_stream_windows(State, NewWindowSize - OldWindowSize); _ -> State end; %% Ack for a previously sent SETTINGS frame. frame(State=#state{local_settings=Local0, next_settings=Next}, settings_ack) -> Local = maps:merge(Local0, Next), State#state{local_settings=Local, next_settings=#{}}; %% Unexpected PUSH_PROMISE frame. frame(State, {push_promise, _, _, _, _}) -> terminate(State, {connection_error, protocol_error, 'PUSH_PROMISE frames MUST only be sent on a peer-initiated stream. (RFC7540 6.6)'}); %% PING frame. frame(State=#state{socket=Socket, transport=Transport}, {ping, Opaque}) -> Transport:send(Socket, cow_http2:ping_ack(Opaque)), State; %% Ack for a previously sent PING frame. %% %% @todo Might want to check contents but probably a waste of time. frame(State, {ping_ack, _Opaque}) -> State; %% GOAWAY frame. frame(State, Frame={goaway, _, _, _}) -> terminate(State, {stop, Frame, 'Client is going away.'}); %% Connection-wide WINDOW_UPDATE frame. frame(State=#state{local_window=ConnWindow}, {window_update, Increment}) when ConnWindow + Increment > 16#7fffffff -> terminate(State, {connection_error, flow_control_error, 'The flow control window must not be greater than 2^31-1. (RFC7540 6.9.1)'}); frame(State=#state{local_window=ConnWindow}, {window_update, Increment}) -> send_data(State#state{local_window=ConnWindow + Increment}); %% Stream-specific WINDOW_UPDATE frame. frame(State=#state{client_streamid=LastStreamID}, {window_update, StreamID, _}) when StreamID > LastStreamID -> terminate(State, {connection_error, protocol_error, 'WINDOW_UPDATE frame received on a stream in idle state. (RFC7540 5.1)'}); frame(State0=#state{streams=Streams0}, {window_update, StreamID, Increment}) -> case lists:keyfind(StreamID, #stream.id, Streams0) of #stream{local_window=StreamWindow} when StreamWindow + Increment > 16#7fffffff -> stream_reset(State0, StreamID, {stream_error, flow_control_error, 'The flow control window must not be greater than 2^31-1. (RFC7540 6.9.1)'}); Stream0 = #stream{local_window=StreamWindow} -> {State, Stream} = send_data(State0, Stream0#stream{local_window=StreamWindow + Increment}), Streams = lists:keystore(StreamID, #stream.id, Streams0, Stream), State#state{streams=Streams}; %% @todo We must reject WINDOW_UPDATE frames on RST_STREAM closed streams. false -> %% WINDOW_UPDATE frames may be received for a short period of time %% after a stream is closed. They must be ignored. State0 end; %% Unexpected CONTINUATION frame. frame(State, {continuation, _, _, _}) -> terminate(State, {connection_error, protocol_error, 'CONTINUATION frames MUST be preceded by a HEADERS frame. (RFC7540 6.10)'}). continuation_frame(State=#state{parse_state={continuation, StreamID, IsFin, HeaderBlockFragment0}}, {continuation, StreamID, head_fin, HeaderBlockFragment1}) -> stream_decode_init(State#state{parse_state=normal}, StreamID, IsFin, << HeaderBlockFragment0/binary, HeaderBlockFragment1/binary >>); continuation_frame(State=#state{parse_state={continuation, StreamID, IsFin, HeaderBlockFragment0}}, {continuation, StreamID, head_nofin, HeaderBlockFragment1}) -> State#state{parse_state={continuation, StreamID, IsFin, << HeaderBlockFragment0/binary, HeaderBlockFragment1/binary >>}}; continuation_frame(State, _) -> terminate(State, {connection_error, protocol_error, 'An invalid frame was received in the middle of a header block. (RFC7540 6.2)'}). down(State=#state{children=Children0}, Pid, Msg) -> case cowboy_children:down(Children0, Pid) of %% The stream was terminated already. {ok, undefined, Children} -> State#state{children=Children}; %% The stream is still running. {ok, StreamID, Children} -> info(State#state{children=Children}, StreamID, Msg); %% The process was unknown. error -> error_logger:error_msg("Received EXIT signal ~p for unknown process ~p.~n", [Msg, Pid]), State end. info(State=#state{client_streamid=LastStreamID, streams=Streams}, StreamID, Msg) -> case lists:keyfind(StreamID, #stream.id, Streams) of #stream{state=flush} -> error_logger:error_msg("Received message ~p for terminated stream ~p.", [Msg, StreamID]), State; Stream = #stream{state=StreamState0} -> try cowboy_stream:info(StreamID, Msg, StreamState0) of {Commands, StreamState} -> commands(State, Stream#stream{state=StreamState}, Commands) catch Class:Exception -> cowboy_stream:report_error(info, [StreamID, Msg, StreamState0], Class, Exception, erlang:get_stacktrace()), stream_reset(State, StreamID, {internal_error, {Class, Exception}, 'Unhandled exception in cowboy_stream:info/3.'}) end; false when StreamID =< LastStreamID -> %% Streams that were reset by the client or streams that are %% in the lingering state may still have Erlang messages going %% around. In these cases we do not want to log anything. State; false -> error_logger:error_msg("Received message ~p for unknown stream ~p.", [Msg, StreamID]), State end. commands(State, Stream, []) -> after_commands(State, Stream); %% Error responses are sent only if a response wasn't sent already. commands(State, Stream=#stream{local=idle}, [{error_response, StatusCode, Headers, Body}|Tail]) -> commands(State, Stream, [{response, StatusCode, Headers, Body}|Tail]); commands(State, Stream, [{error_response, _, _, _}|Tail]) -> commands(State, Stream, Tail); %% Send an informational response. commands(State0, Stream=#stream{local=idle}, [{inform, StatusCode, Headers}|Tail]) -> State = send_headers(State0, Stream, StatusCode, Headers, fin), commands(State, Stream, Tail); %% Send response headers. %% %% @todo Kill the stream if it sent a response when one has already been sent. %% @todo Keep IsFin in the state. %% @todo Same two things above apply to DATA, possibly promise too. commands(State0, Stream0=#stream{local=idle}, [{response, StatusCode, Headers, Body}|Tail]) -> {State, Stream} = send_response(State0, Stream0, StatusCode, Headers, Body), commands(State, Stream, Tail); %% @todo response when local!=idle %% Send response headers. commands(State0, Stream=#stream{method=Method, local=idle}, [{headers, StatusCode, Headers}|Tail]) -> IsFin = case Method of <<"HEAD">> -> fin; _ -> nofin end, State = send_headers(State0, Stream, StatusCode, Headers, IsFin), commands(State, Stream#stream{local=IsFin}, Tail); %% @todo headers when local!=idle %% Send a response body chunk. commands(State0, Stream0=#stream{local=nofin}, [{data, IsFin, Data}|Tail]) -> {State, Stream} = send_data(State0, Stream0, IsFin, Data), commands(State, Stream, Tail); %% @todo data when local!=nofin %% Send trailers. commands(State0, Stream0=#stream{local=nofin, te=TE0}, [{trailers, Trailers}|Tail]) -> %% We only accept TE headers containing exactly "trailers" (RFC7540 8.1.2.1). TE = try cow_http_hd:parse_te(TE0) of {trailers, []} -> trailers; _ -> no_trailers catch _:_ -> %% If we can't parse the TE header, assume we can't send trailers. no_trailers end, {State, Stream} = case TE of trailers -> send_data(State0, Stream0, fin, {trailers, Trailers}); no_trailers -> send_data(State0, Stream0, fin, <<>>) end, commands(State, Stream, Tail); %% Send a file. %% @todo Add the sendfile command. %commands(State0, Stream0=#stream{local=nofin}, % [{sendfile, IsFin, Offset, Bytes, Path}|Tail]) -> % {State, Stream} = send_data(State0, Stream0, IsFin, {sendfile, Offset, Bytes, Path}), % commands(State, Stream, Tail); %% @todo sendfile when local!=nofin %% Send a push promise. %% %% @todo We need to keep track of what promises we made so that we don't %% end up with an infinite loop of promises. commands(State0=#state{socket=Socket, transport=Transport, server_streamid=PromisedStreamID, encode_state=EncodeState0}, Stream=#stream{id=StreamID}, [{push, Method, Scheme, Host, Port, Path, Qs, Headers0}|Tail]) -> Authority = case {Scheme, Port} of {<<"http">>, 80} -> Host; {<<"https">>, 443} -> Host; _ -> iolist_to_binary([Host, $:, integer_to_binary(Port)]) end, PathWithQs = iolist_to_binary(case Qs of <<>> -> Path; _ -> [Path, $?, Qs] end), %% We need to make sure the header value is binary before we can %% pass it to stream_req_init, as it expects them to be flat. Headers1 = maps:map(fun(_, V) -> iolist_to_binary(V) end, Headers0), Headers = Headers1#{ <<":method">> => Method, <<":scheme">> => Scheme, <<":authority">> => Authority, <<":path">> => PathWithQs}, {HeaderBlock, EncodeState} = headers_encode(Headers, EncodeState0), Transport:send(Socket, cow_http2:push_promise(StreamID, PromisedStreamID, HeaderBlock)), State = stream_req_init(State0#state{server_streamid=PromisedStreamID + 2, encode_state=EncodeState}, PromisedStreamID, fin, Headers1, #{ method => Method, scheme => Scheme, authority => Authority, path => PathWithQs }), commands(State, Stream, Tail); commands(State=#state{socket=Socket, transport=Transport, remote_window=ConnWindow}, Stream=#stream{id=StreamID, remote_window=StreamWindow}, [{flow, Size}|Tail]) -> Transport:send(Socket, [ cow_http2:window_update(Size), cow_http2:window_update(StreamID, Size) ]), commands(State#state{remote_window=ConnWindow + Size}, Stream#stream{remote_window=StreamWindow + Size}, Tail); %% Supervise a child process. commands(State=#state{children=Children}, Stream=#stream{id=StreamID}, [{spawn, Pid, Shutdown}|Tail]) -> commands(State#state{children=cowboy_children:up(Children, Pid, StreamID, Shutdown)}, Stream, Tail); %% Error handling. commands(State, Stream=#stream{id=StreamID}, [Error = {internal_error, _, _}|_Tail]) -> %% @todo Do we want to run the commands after an internal_error? %% @todo Do we even allow commands after? %% @todo Only reset when the stream still exists. stream_reset(after_commands(State, Stream), StreamID, Error); %% Upgrade to HTTP/2. This is triggered by cowboy_http2 itself. commands(State=#state{socket=Socket, transport=Transport}, Stream=#stream{local=upgrade}, [{switch_protocol, Headers, ?MODULE, _}|Tail]) -> Transport:send(Socket, cow_http:response(101, 'HTTP/1.1', maps:to_list(Headers))), commands(State, Stream#stream{local=idle}, Tail); %% Use a different protocol within the stream (CONNECT :protocol). %% @todo Make sure we error out when the feature is disabled. commands(State0, #stream{id=StreamID}, [{switch_protocol, Headers, _Mod, _ModState}|Tail]) -> State = #state{streams=Streams} = info(State0, StreamID, {headers, 200, Headers}), Stream = lists:keyfind(StreamID, #stream.id, Streams), commands(State, Stream, Tail); commands(State, Stream=#stream{id=StreamID}, [stop|_Tail]) -> %% @todo Do we want to run the commands after a stop? %% @todo Do we even allow commands after? stream_terminate(after_commands(State, Stream), StreamID, normal). after_commands(State=#state{streams=Streams0}, Stream=#stream{id=StreamID}) -> Streams = lists:keystore(StreamID, #stream.id, Streams0, Stream), State#state{streams=Streams}. send_response(State0, Stream=#stream{method=Method}, StatusCode, Headers0, Body) -> if Method =:= <<"HEAD">>; Body =:= <<>> -> State = send_headers(State0, Stream, StatusCode, Headers0, fin), {State, Stream#stream{local=fin}}; true -> State = send_headers(State0, Stream, StatusCode, Headers0, nofin), %% send_data works with both sendfile and iolists. send_data(State, Stream#stream{local=nofin}, fin, Body) end. send_headers(State=#state{socket=Socket, transport=Transport, encode_state=EncodeState0}, #stream{id=StreamID}, StatusCode, Headers0, IsFin) -> Headers = Headers0#{<<":status">> => status(StatusCode)}, {HeaderBlock, EncodeState} = headers_encode(Headers, EncodeState0), Transport:send(Socket, cow_http2:headers(StreamID, IsFin, HeaderBlock)), State#state{encode_state=EncodeState}. status(Status) when is_integer(Status) -> integer_to_binary(Status); status(<< H, T, U, _/bits >>) when H >= $1, H =< $9, T >= $0, T =< $9, U >= $0, U =< $9 -> << H, T, U >>. %% @todo Should we ever want to implement the PRIORITY mechanism, %% this would be the place to do it. Right now, we just go over %% all streams and send what we can until either everything is %% sent or we run out of space in the window. send_data(State=#state{streams=Streams}) -> resume_streams(State, Streams, []). %% When SETTINGS_INITIAL_WINDOW_SIZE changes we need to update %% the stream windows for all active streams and perhaps resume %% sending data. update_stream_windows(State=#state{streams=Streams0}, Increment) -> Streams = [ S#stream{local_window=StreamWindow + Increment} || S=#stream{local_window=StreamWindow} <- Streams0], resume_streams(State, Streams, []). resume_streams(State, [], Acc) -> State#state{streams=lists:reverse(Acc)}; %% While technically we should never get < 0 here, let's be on the safe side. resume_streams(State=#state{local_window=ConnWindow}, Streams, Acc) when ConnWindow =< 0 -> State#state{streams=lists:reverse(Acc, Streams)}; %% We rely on send_data/2 to do all the necessary checks about the stream. resume_streams(State0, [Stream0|Tail], Acc) -> {State1, Stream} = send_data(State0, Stream0), case Stream of %% We are done flushing, remove the stream. %% Maybe skip the request body if it was not fully read. #stream{state=flush, local=fin} -> State = maybe_skip_body(State1, Stream, normal), resume_streams(State, Tail, Acc); %% Keep the stream. Either the stream handler is still running, %% or we are not finished flushing. _ -> resume_streams(State1, Tail, [Stream|Acc]) end. send_data(State, Stream=#stream{local=Local, local_buffer_size=0, local_trailers=Trailers}) when (Trailers =/= undefined) andalso ((Local =:= idle) orelse (Local =:= nofin)) -> send_trailers(State, Stream#stream{local_trailers=undefined}, Trailers); %% @todo We might want to print an error if local=fin. %% %% @todo It's possible that the stream terminates. We must remove it. send_data(State=#state{local_window=ConnWindow}, Stream=#stream{local=IsFin, local_window=StreamWindow, local_buffer_size=BufferSize}) when ConnWindow =< 0; IsFin =:= fin; StreamWindow =< 0; BufferSize =:= 0 -> {State, Stream}; send_data(State0, Stream0=#stream{local_buffer=Q0, local_buffer_size=BufferSize}) -> %% We know there is an item in the queue. {{value, {IsFin, DataSize, Data}}, Q} = queue:out(Q0), {State, Stream} = send_data(State0, Stream0#stream{local_buffer=Q, local_buffer_size=BufferSize - DataSize}, IsFin, Data, in_r), send_data(State, Stream). send_data(State, Stream, IsFin, Data) -> send_data(State, Stream, IsFin, Data, in). %% We can send trailers immediately if the queue is empty, otherwise we queue. %% We always send trailer frames even if the window is empty. send_data(State, Stream=#stream{local_buffer_size=0}, fin, {trailers, Trailers}, _) -> send_trailers(State, Stream, Trailers); send_data(State, Stream, fin, {trailers, Trailers}, _) -> {State, Stream#stream{local_trailers=Trailers}}; %% Send data immediately if we can, buffer otherwise. %% @todo We might want to print an error if local=fin. send_data(State=#state{local_window=ConnWindow}, Stream=#stream{local_window=StreamWindow}, IsFin, Data, In) when ConnWindow =< 0; StreamWindow =< 0 -> {State, queue_data(Stream, IsFin, Data, In)}; send_data(State=#state{socket=Socket, transport=Transport, local_window=ConnWindow}, Stream=#stream{id=StreamID, local_window=StreamWindow}, IsFin, Data, In) -> MaxFrameSize = 16384, %% @todo Use the real SETTINGS_MAX_FRAME_SIZE set by the client. MaxSendSize = min(min(ConnWindow, StreamWindow), MaxFrameSize), case Data of {sendfile, Offset, Bytes, Path} when Bytes =< MaxSendSize -> Transport:send(Socket, cow_http2:data_header(StreamID, IsFin, Bytes)), Transport:sendfile(Socket, Path, Offset, Bytes), {State#state{local_window=ConnWindow - Bytes}, Stream#stream{local=IsFin, local_window=StreamWindow - Bytes}}; {sendfile, Offset, Bytes, Path} -> Transport:send(Socket, cow_http2:data_header(StreamID, nofin, MaxSendSize)), Transport:sendfile(Socket, Path, Offset, MaxSendSize), send_data(State#state{local_window=ConnWindow - MaxSendSize}, Stream#stream{local_window=StreamWindow - MaxSendSize}, IsFin, {sendfile, Offset + MaxSendSize, Bytes - MaxSendSize, Path}, In); Iolist0 -> IolistSize = iolist_size(Iolist0), if IolistSize =< MaxSendSize -> Transport:send(Socket, cow_http2:data(StreamID, IsFin, Iolist0)), {State#state{local_window=ConnWindow - IolistSize}, Stream#stream{local=IsFin, local_window=StreamWindow - IolistSize}}; true -> {Iolist, More} = cowboy_iolists:split(MaxSendSize, Iolist0), Transport:send(Socket, cow_http2:data(StreamID, nofin, Iolist)), send_data(State#state{local_window=ConnWindow - MaxSendSize}, Stream#stream{local_window=StreamWindow - MaxSendSize}, IsFin, More, In) end end. send_trailers(State=#state{socket=Socket, transport=Transport, encode_state=EncodeState0}, Stream=#stream{id=StreamID}, Trailers) -> {HeaderBlock, EncodeState} = headers_encode(Trailers, EncodeState0), Transport:send(Socket, cow_http2:headers(StreamID, fin, HeaderBlock)), {State#state{encode_state=EncodeState}, Stream#stream{local=fin}}. queue_data(Stream=#stream{local_buffer=Q0, local_buffer_size=Size0}, IsFin, Data, In) -> DataSize = case Data of {sendfile, _, Bytes, _} -> Bytes; Iolist -> iolist_size(Iolist) end, Q = queue:In({IsFin, DataSize, Data}, Q0), Stream#stream{local_buffer=Q, local_buffer_size=Size0 + DataSize}. %% The set-cookie header is special; we can only send one cookie per header. headers_encode(Headers0=#{<<"set-cookie">> := SetCookies}, EncodeState) -> Headers1 = maps:to_list(maps:remove(<<"set-cookie">>, Headers0)), Headers = Headers1 ++ [{<<"set-cookie">>, Value} || Value <- SetCookies], cow_hpack:encode(Headers, EncodeState); headers_encode(Headers0, EncodeState) -> Headers = maps:to_list(Headers0), cow_hpack:encode(Headers, EncodeState). -spec terminate(#state{}, _) -> no_return(). terminate(undefined, Reason) -> exit({shutdown, Reason}); terminate(#state{socket=Socket, transport=Transport, parse_state={preface, _, _}}, Reason) -> Transport:close(Socket), exit({shutdown, Reason}); terminate(#state{socket=Socket, transport=Transport, client_streamid=LastStreamID, streams=Streams, children=Children}, Reason) -> %% @todo We might want to optionally send the Reason value %% as debug data in the GOAWAY frame here. Perhaps more. Transport:send(Socket, cow_http2:goaway(LastStreamID, terminate_reason(Reason), <<>>)), terminate_all_streams(Streams, Reason), cowboy_children:terminate(Children), Transport:close(Socket), exit({shutdown, Reason}). terminate_reason({connection_error, Reason, _}) -> Reason; terminate_reason({stop, _, _}) -> no_error; terminate_reason({socket_error, _, _}) -> internal_error; terminate_reason({internal_error, _, _}) -> internal_error. terminate_all_streams([], _) -> ok; %% This stream was already terminated and is now just flushing the data out. Skip it. terminate_all_streams([#stream{state=flush}|Tail], Reason) -> terminate_all_streams(Tail, Reason); terminate_all_streams([#stream{id=StreamID, state=StreamState}|Tail], Reason) -> stream_call_terminate(StreamID, Reason, StreamState), terminate_all_streams(Tail, Reason). %% Stream functions. stream_decode_init(State=#state{decode_state=DecodeState0}, StreamID, IsFin, HeaderBlock) -> try cow_hpack:decode(HeaderBlock, DecodeState0) of {Headers, DecodeState} -> stream_pseudo_headers_init(State#state{decode_state=DecodeState}, StreamID, IsFin, Headers) catch _:_ -> terminate(State, {connection_error, compression_error, 'Error while trying to decode HPACK-encoded header block. (RFC7540 4.3)'}) end. stream_pseudo_headers_init(State=#state{local_settings=LocalSettings}, StreamID, IsFin, Headers0) -> IsExtendedConnectEnabled = maps:get(enable_connect_protocol, LocalSettings, false), case pseudo_headers(Headers0, #{}) of {ok, PseudoHeaders=#{method := <<"CONNECT">>, scheme := _, authority := _, path := _, protocol := _}, Headers} when IsExtendedConnectEnabled -> stream_regular_headers_init(State, StreamID, IsFin, Headers, PseudoHeaders); {ok, #{method := <<"CONNECT">>, scheme := _, authority := _, path := _}, _} when IsExtendedConnectEnabled -> stream_malformed(State, StreamID, 'The :protocol pseudo-header MUST be sent with an extended CONNECT. (draft_h2_websockets 4)'); {ok, #{protocol := _}, _} -> stream_malformed(State, StreamID, 'The :protocol pseudo-header is only defined for the extended CONNECT. (draft_h2_websockets 4)'); %% @todo Add clause for CONNECT requests (no scheme/path). {ok, PseudoHeaders=#{method := <<"CONNECT">>}, _} -> stream_early_error(State, StreamID, IsFin, 501, PseudoHeaders, 'The CONNECT method is currently not implemented. (RFC7231 4.3.6)'); {ok, PseudoHeaders=#{method := <<"TRACE">>}, _} -> stream_early_error(State, StreamID, IsFin, 501, PseudoHeaders, 'The TRACE method is currently not implemented. (RFC7231 4.3.8)'); {ok, PseudoHeaders=#{method := _, scheme := _, authority := _, path := _}, Headers} -> stream_regular_headers_init(State, StreamID, IsFin, Headers, PseudoHeaders); {ok, _, _} -> stream_malformed(State, StreamID, 'A required pseudo-header was not found. (RFC7540 8.1.2.3)'); {error, HumanReadable} -> stream_malformed(State, StreamID, HumanReadable) end. pseudo_headers([{<<":method">>, _}|_], #{method := _}) -> {error, 'Multiple :method pseudo-headers were found. (RFC7540 8.1.2.3)'}; pseudo_headers([{<<":method">>, Method}|Tail], PseudoHeaders) -> pseudo_headers(Tail, PseudoHeaders#{method => Method}); pseudo_headers([{<<":scheme">>, _}|_], #{scheme := _}) -> {error, 'Multiple :scheme pseudo-headers were found. (RFC7540 8.1.2.3)'}; pseudo_headers([{<<":scheme">>, Scheme}|Tail], PseudoHeaders) -> pseudo_headers(Tail, PseudoHeaders#{scheme => Scheme}); pseudo_headers([{<<":authority">>, _}|_], #{authority := _}) -> {error, 'Multiple :authority pseudo-headers were found. (RFC7540 8.1.2.3)'}; pseudo_headers([{<<":authority">>, Authority}|Tail], PseudoHeaders) -> pseudo_headers(Tail, PseudoHeaders#{authority => Authority}); pseudo_headers([{<<":path">>, _}|_], #{path := _}) -> {error, 'Multiple :path pseudo-headers were found. (RFC7540 8.1.2.3)'}; pseudo_headers([{<<":path">>, Path}|Tail], PseudoHeaders) -> pseudo_headers(Tail, PseudoHeaders#{path => Path}); pseudo_headers([{<<":protocol">>, _}|_], #{protocol := _}) -> {error, 'Multiple :protocol pseudo-headers were found. (RFC7540 8.1.2.3)'}; pseudo_headers([{<<":protocol">>, Protocol}|Tail], PseudoHeaders) -> pseudo_headers(Tail, PseudoHeaders#{protocol => Protocol}); pseudo_headers([{<<":", _/bits>>, _}|_], _) -> {error, 'An unknown or invalid pseudo-header was found. (RFC7540 8.1.2.1)'}; pseudo_headers(Headers, PseudoHeaders) -> {ok, PseudoHeaders, Headers}. stream_regular_headers_init(State, StreamID, IsFin, Headers, PseudoHeaders) -> case regular_headers(Headers) of ok -> stream_req_init(State, StreamID, IsFin, headers_to_map(Headers, #{}), PseudoHeaders); {error, HumanReadable} -> stream_malformed(State, StreamID, HumanReadable) end. regular_headers([{<<":", _/bits>>, _}|_]) -> {error, 'Pseudo-headers were found after regular headers. (RFC7540 8.1.2.1)'}; regular_headers([{<<"connection">>, _}|_]) -> {error, 'The connection header is not allowed. (RFC7540 8.1.2.2)'}; regular_headers([{<<"keep-alive">>, _}|_]) -> {error, 'The keep-alive header is not allowed. (RFC7540 8.1.2.2)'}; regular_headers([{<<"proxy-authenticate">>, _}|_]) -> {error, 'The proxy-authenticate header is not allowed. (RFC7540 8.1.2.2)'}; regular_headers([{<<"proxy-authorization">>, _}|_]) -> {error, 'The proxy-authorization header is not allowed. (RFC7540 8.1.2.2)'}; regular_headers([{<<"transfer-encoding">>, _}|_]) -> {error, 'The transfer-encoding header is not allowed. (RFC7540 8.1.2.2)'}; regular_headers([{<<"upgrade">>, _}|_]) -> {error, 'The upgrade header is not allowed. (RFC7540 8.1.2.2)'}; regular_headers([{<<"te">>, Value}|_]) when Value =/= <<"trailers">> -> {error, 'The te header with a value other than "trailers" is not allowed. (RFC7540 8.1.2.2)'}; regular_headers([{Name, _}|Tail]) -> case cowboy_bstr:to_lower(Name) of Name -> regular_headers(Tail); _ -> {error, 'Header names must be lowercase. (RFC7540 8.1.2)'} end; regular_headers([]) -> ok. %% This function is necessary to properly handle duplicate headers %% and the special-case cookie header. headers_to_map([], Acc) -> Acc; headers_to_map([{Name, Value}|Tail], Acc0) -> Acc = case Acc0 of %% The cookie header does not use proper HTTP header lists. #{Name := Value0} when Name =:= <<"cookie">> -> Acc0#{Name => << Value0/binary, "; ", Value/binary >>}; #{Name := Value0} -> Acc0#{Name => << Value0/binary, ", ", Value/binary >>}; _ -> Acc0#{Name => Value} end, headers_to_map(Tail, Acc). stream_req_init(State, StreamID, IsFin, Headers, PseudoHeaders) -> case Headers of _ when IsFin =:= fin -> stream_req_init(State, StreamID, IsFin, Headers, PseudoHeaders, 0); #{<<"content-length">> := <<"0">>} -> stream_req_init(State, StreamID, IsFin, Headers, PseudoHeaders, 0); #{<<"content-length">> := BinLength} -> try stream_req_init(State, StreamID, IsFin, Headers, PseudoHeaders, cow_http_hd:parse_content_length(BinLength)) catch _:_ -> stream_malformed(State, StreamID, 'The content-length header is invalid. (RFC7230 3.3.2)') end; _ -> stream_req_init(State, StreamID, IsFin, Headers, PseudoHeaders, undefined) end. stream_req_init(State=#state{ref=Ref, peer=Peer, sock=Sock, cert=Cert}, StreamID, IsFin, Headers, PseudoHeaders=#{method := Method, scheme := Scheme, authority := Authority, path := PathWithQs}, BodyLength) -> try cow_http_hd:parse_host(Authority) of {Host, Port} -> try cow_http:parse_fullpath(PathWithQs) of {<<>>, _} -> stream_malformed(State, StreamID, 'The path component must not be empty. (RFC7540 8.1.2.3)'); {Path, Qs} -> Req0 = #{ ref => Ref, pid => self(), streamid => StreamID, peer => Peer, sock => Sock, cert => Cert, method => Method, scheme => Scheme, host => Host, port => Port, path => Path, qs => Qs, version => 'HTTP/2', headers => Headers, has_body => IsFin =:= nofin, body_length => BodyLength }, %% We add the protocol information for extended CONNECTs. Req = case PseudoHeaders of #{protocol := Protocol} -> Req0#{protocol => Protocol}; _ -> Req0 end, stream_handler_init(State, StreamID, IsFin, idle, Req) catch _:_ -> stream_malformed(State, StreamID, 'The :path pseudo-header is invalid. (RFC7540 8.1.2.3)') end catch _:_ -> stream_malformed(State, StreamID, 'The :authority pseudo-header is invalid. (RFC7540 8.1.2.3)') end. stream_malformed(State=#state{socket=Socket, transport=Transport}, StreamID, _) -> Transport:send(Socket, cow_http2:rst_stream(StreamID, protocol_error)), State. stream_early_error(State0=#state{ref=Ref, opts=Opts, peer=Peer, local_settings=#{initial_window_size := RemoteWindow}, remote_settings=#{initial_window_size := LocalWindow}, streams=Streams}, StreamID, IsFin, StatusCode0, #{method := Method}, HumanReadable) -> %% We automatically terminate the stream but it is not an error %% per se (at least not in the first implementation). Reason = {stream_error, no_error, HumanReadable}, %% The partial Req is minimal for now. We only have one case %% where it can be called (when a method is completely disabled). PartialReq = #{ ref => Ref, peer => Peer, method => Method }, Resp = {response, StatusCode0, RespHeaders0=#{<<"content-length">> => <<"0">>}, <<>>}, %% We need a stream to talk to the send_* functions. Stream0 = #stream{id=StreamID, state=flush, method=Method, remote=IsFin, local=idle, local_window=LocalWindow, remote_window=RemoteWindow}, try cowboy_stream:early_error(StreamID, Reason, PartialReq, Resp, Opts) of {response, StatusCode, RespHeaders, RespBody} -> case send_response(State0, Stream0, StatusCode, RespHeaders, RespBody) of {State, #stream{local=fin}} -> State; {State, Stream} -> State#state{streams=[Stream|Streams]} end catch Class:Exception -> cowboy_stream:report_error(early_error, [StreamID, Reason, PartialReq, Resp, Opts], Class, Exception, erlang:get_stacktrace()), %% We still need to send an error response, so send what we initially %% wanted to send. It's better than nothing. send_headers(State0, Stream0, StatusCode0, RespHeaders0, fin) end. stream_handler_init(State=#state{opts=Opts, local_settings=#{initial_window_size := RemoteWindow}, remote_settings=#{initial_window_size := LocalWindow}}, StreamID, RemoteIsFin, LocalIsFin, Req=#{method := Method, headers := Headers}) -> try cowboy_stream:init(StreamID, Req, Opts) of {Commands, StreamState} -> commands(State#state{client_streamid=StreamID}, #stream{id=StreamID, state=StreamState, method=Method, remote=RemoteIsFin, local=LocalIsFin, local_window=LocalWindow, remote_window=RemoteWindow, te=maps:get(<<"te">>, Headers, undefined)}, Commands) catch Class:Exception -> cowboy_stream:report_error(init, [StreamID, Req, Opts], Class, Exception, erlang:get_stacktrace()), stream_reset(State, StreamID, {internal_error, {Class, Exception}, 'Unhandled exception in cowboy_stream:init/3.'}) end. %% @todo Don't send an RST_STREAM if one was already sent. stream_reset(State=#state{socket=Socket, transport=Transport}, StreamID, StreamError) -> Reason = case StreamError of {internal_error, _, _} -> internal_error; {stream_error, Reason0, _} -> Reason0 end, Transport:send(Socket, cow_http2:rst_stream(StreamID, Reason)), stream_terminate(stream_linger(State, StreamID), StreamID, StreamError). %% We only keep up to 100 streams in this state. @todo Make it configurable? stream_linger(State=#state{lingering_streams=Lingering0}, StreamID) -> Lingering = [StreamID|lists:sublist(Lingering0, 100 - 1)], State#state{lingering_streams=Lingering}. stream_terminate(State0=#state{streams=Streams0, children=Children0}, StreamID, Reason) -> case lists:keytake(StreamID, #stream.id, Streams0) of %% When the stream terminates normally (without sending RST_STREAM) %% and no response was sent, we need to send a proper response back to the client. {value, Stream=#stream{local=idle}, Streams} when Reason =:= normal -> State1 = #state{streams=Streams1} = info(State0, StreamID, {response, 204, #{}, <<>>}), State = maybe_skip_body(State1, Stream, Reason), #stream{state=StreamState} = lists:keyfind(StreamID, #stream.id, Streams1), stream_call_terminate(StreamID, Reason, StreamState), Children = cowboy_children:shutdown(Children0, StreamID), State#state{streams=Streams, children=Children}; %% When a response was sent but not terminated, we need to close the stream. {value, Stream=#stream{local=nofin, local_buffer_size=0}, Streams} when Reason =:= normal -> State1 = #state{streams=Streams1} = info(State0, StreamID, {data, fin, <<>>}), State = maybe_skip_body(State1, Stream, Reason), #stream{state=StreamState} = lists:keyfind(StreamID, #stream.id, Streams1), stream_call_terminate(StreamID, Reason, StreamState), Children = cowboy_children:shutdown(Children0, StreamID), State#state{streams=Streams, children=Children}; %% Unless there is still data in the buffer. We can however reset %% a few fields and set a special local state to avoid confusion. %% %% We do not reset the stream in this case (to skip the body) %% because we are still sending data via the buffer. We will %% reset the stream if necessary once the buffer is empty. {value, Stream=#stream{state=StreamState, local=nofin}, Streams} -> stream_call_terminate(StreamID, Reason, StreamState), Children = cowboy_children:shutdown(Children0, StreamID), State0#state{streams=[Stream#stream{state=flush, local=flush}|Streams], children=Children}; %% Otherwise we sent or received an RST_STREAM and/or the stream is already closed. {value, Stream=#stream{state=StreamState}, Streams} -> State = maybe_skip_body(State0, Stream, Reason), stream_call_terminate(StreamID, Reason, StreamState), Children = cowboy_children:shutdown(Children0, StreamID), State#state{streams=Streams, children=Children}; %% The stream doesn't exist. This can occur for various reasons. %% It can happen before the stream has been created, or because %% the cowboy_stream:init call failed, in which case doing nothing %% is correct. false -> State0 end. %% When the stream stops normally without reading the request %% body fully we need to tell the client to stop sending it. %% We do this by sending an RST_STREAM with reason NO_ERROR. (RFC7540 8.1.0) maybe_skip_body(State=#state{socket=Socket, transport=Transport}, #stream{id=StreamID, remote=nofin}, normal) -> Transport:send(Socket, cow_http2:rst_stream(StreamID, no_error)), stream_linger(State, StreamID); maybe_skip_body(State, _, _) -> State. stream_call_terminate(StreamID, Reason, StreamState) -> try cowboy_stream:terminate(StreamID, Reason, StreamState) catch Class:Exception -> cowboy_stream:report_error(terminate, [StreamID, Reason, StreamState], Class, Exception, erlang:get_stacktrace()) end. %% System callbacks. -spec system_continue(_, _, {#state{}, binary()}) -> ok. system_continue(_, _, {State, Buffer}) -> loop(State, Buffer). -spec system_terminate(any(), _, _, {#state{}, binary()}) -> no_return(). system_terminate(Reason, _, _, {State, _}) -> %% @todo We should exit gracefully, if possible. terminate(State, Reason). -spec system_code_change(Misc, _, _, _) -> {ok, Misc} when Misc::{#state{}, binary()}. system_code_change(Misc, _, _, _) -> {ok, Misc}.