%% Copyright (c) 2015-2017, Loïc Hoguin <[email protected]>
%%
%% 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,
%% 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(),
%% Whether we finished receiving data.
remote = nofin :: cowboy_stream:fin(),
%% Remote flow control window (how much we accept to receive).
remote_window :: integer()
}).
-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, 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) ->
State = #state{parent=Parent, ref=Ref, socket=Socket,
transport=Transport, opts=Opts, peer=Peer, sock=Sock, cert=Cert,
parse_state={preface, sequence, preface_timeout(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.
State = info(State1, 1, {switch_protocol, #{
<<"connection">> => <<"Upgrade">>,
<<"upgrade">> => <<"h2c">>
}, ?MODULE, undefined}), %% @todo undefined or #{}?
preface(State),
case Buffer of
<<>> -> before_loop(State, Buffer);
_ -> parse(State, Buffer)
end.
preface(#state{socket=Socket, transport=Transport, next_settings=Settings}) ->
%% We send next_settings and use defaults until we get a ack.
ok = Transport:send(Socket, cow_http2:settings(Settings)).
preface_timeout(Opts) ->
PrefaceTimeout = maps:get(preface_timeout, Opts, 5000),
erlang:start_timer(PrefaceTimeout, self(), preface_timeout).
%% @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} ->
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, Tag}, which_children} ->
From ! {Tag, cowboy_children:which_children(Children, ?MODULE)},
loop(State, Buffer);
{'$gen_call', {From, Tag}, count_children} ->
From ! {Tag, cowboy_children:count_children(Children)},
loop(State, Buffer);
{'$gen_call', {From, Tag}, _} ->
From ! {Tag, {error, ?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 ->
%% @todo OK we should have a timeout when waiting for the 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, 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) ->
_ = erlang:cancel_timer(TRef, [{async, true}, {info, false}]),
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.
frame(State=#state{client_streamid=LastStreamID}, {headers, StreamID, _, _, _})
when StreamID =< LastStreamID ->
stream_reset(State, StreamID, {stream_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(State=#state{socket=Socket, transport=Transport, remote_settings=Settings0},
{settings, Settings}) ->
Transport:send(Socket, cow_http2:settings_ack()),
State#state{remote_settings=maps:merge(Settings0, Settings)};
%% Ack for a previously sent SETTINGS frame.
frame(State=#state{next_settings=_NextSettings}, settings_ack) ->
%% @todo Apply SETTINGS that require synchronization.
State;
%% 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}) ->
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
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 while expecting a CONTINUATION frame. (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{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 ->
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(State=#state{socket=Socket, transport=Transport, encode_state=EncodeState0},
Stream=#stream{id=StreamID, local=idle}, [{inform, StatusCode, Headers0}|Tail]) ->
Headers = Headers0#{<<":status">> => status(StatusCode)},
{HeaderBlock, EncodeState} = headers_encode(Headers, EncodeState0),
Transport:send(Socket, cow_http2:headers(StreamID, fin, HeaderBlock)),
commands(State#state{encode_state=EncodeState}, 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(State=#state{socket=Socket, transport=Transport, encode_state=EncodeState0},
Stream=#stream{id=StreamID, local=idle}, [{response, StatusCode, Headers0, Body}|Tail]) ->
Headers = Headers0#{<<":status">> => status(StatusCode)},
{HeaderBlock, EncodeState} = headers_encode(Headers, EncodeState0),
case Body of
<<>> ->
Transport:send(Socket, cow_http2:headers(StreamID, fin, HeaderBlock)),
commands(State#state{encode_state=EncodeState}, Stream#stream{local=fin}, Tail);
{sendfile, O, B, P} ->
Transport:send(Socket, cow_http2:headers(StreamID, nofin, HeaderBlock)),
commands(State#state{encode_state=EncodeState}, Stream#stream{local=nofin},
[{sendfile, fin, O, B, P}|Tail]);
_ ->
Transport:send(Socket, cow_http2:headers(StreamID, nofin, HeaderBlock)),
{State1, Stream1} = send_data(State, Stream#stream{local=nofin}, fin, Body),
commands(State1#state{encode_state=EncodeState}, Stream1, Tail)
end;
%% @todo response when local!=idle
%% Send response headers.
commands(State=#state{socket=Socket, transport=Transport, encode_state=EncodeState0},
Stream=#stream{id=StreamID, local=idle}, [{headers, StatusCode, Headers0}|Tail]) ->
Headers = Headers0#{<<":status">> => status(StatusCode)},
{HeaderBlock, EncodeState} = headers_encode(Headers, EncodeState0),
Transport:send(Socket, cow_http2:headers(StreamID, nofin, HeaderBlock)),
commands(State#state{encode_state=EncodeState}, Stream#stream{local=nofin}, 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 a file.
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 = 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">> => iolist_to_binary(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, Headers),
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);
%% HTTP/2 has no support for the Upgrade mechanism.
commands(State, Stream, [{switch_protocol, _Headers, _Mod, _ModState}|Tail]) ->
%% @todo This is an error. Not sure what to do here yet.
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}.
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, []).
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.
%% @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).
%% 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.
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}.
-spec terminate(#state{}, _) -> no_return().
terminate(undefined, Reason) ->
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{socket=Socket, transport=Transport,
decode_state=DecodeState0}, StreamID, IsFin, HeaderBlock) ->
%% @todo Add clause for CONNECT requests (no scheme/path).
try headers_decode(HeaderBlock, DecodeState0) of
{Headers=#{<<":method">> := _, <<":scheme">> := _,
<<":authority">> := _, <<":path">> := _}, DecodeState} ->
stream_req_init(State#state{decode_state=DecodeState}, StreamID, IsFin, Headers);
{_, DecodeState} ->
Transport:send(Socket, cow_http2:rst_stream(StreamID, protocol_error)),
State#state{decode_state=DecodeState}
catch _:_ ->
terminate(State, {connection_error, compression_error,
'Error while trying to decode HPACK-encoded header block. (RFC7540 4.3)'})
end.
stream_req_init(State=#state{ref=Ref, peer=Peer, sock=Sock, cert=Cert},
StreamID, IsFin, Headers0=#{
<<":method">> := Method, <<":scheme">> := Scheme,
<<":authority">> := Authority, <<":path">> := PathWithQs}) ->
Headers = maps:without([<<":method">>, <<":scheme">>, <<":authority">>, <<":path">>], Headers0),
BodyLength = case Headers of
_ when IsFin =:= fin ->
0;
#{<<"content-length">> := <<"0">>} ->
0;
#{<<"content-length">> := BinLength} ->
try
cow_http_hd:parse_content_length(BinLength)
catch _:_ ->
terminate(State, {stream_error, StreamID, protocol_error,
'The content-length header is invalid. (RFC7230 3.3.2)'})
end;
_ ->
undefined
end,
%% @todo If this fails to parse we want to gracefully handle the crash.
{Host, Port} = cow_http_hd:parse_host(Authority),
{Path, Qs} = cow_http:parse_fullpath(PathWithQs),
Req = #{
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
},
stream_handler_init(State, StreamID, IsFin, idle, Req).
stream_handler_init(State=#state{opts=Opts,
local_settings=#{initial_window_size := RemoteWindow},
remote_settings=#{initial_window_size := LocalWindow}},
StreamID, RemoteIsFin, LocalIsFin, Req) ->
try cowboy_stream:init(StreamID, Req, Opts) of
{Commands, StreamState} ->
commands(State#state{client_streamid=StreamID},
#stream{id=StreamID, state=StreamState,
remote=RemoteIsFin, local=LocalIsFin,
local_window=LocalWindow, remote_window=RemoteWindow},
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 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.
%% Headers encode/decode.
headers_decode(HeaderBlock, DecodeState0) ->
{Headers, DecodeState} = cow_hpack:decode(HeaderBlock, DecodeState0),
{headers_to_map(Headers, #{}), DecodeState}.
%% 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).
%% 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).
%% 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, _}) ->
terminate(State, Reason).
-spec system_code_change(Misc, _, _, _) -> {ok, Misc} when Misc::{#state{}, binary()}.
system_code_change(Misc, _, _, _) ->
{ok, Misc}.
