aboutsummaryrefslogtreecommitdiffstats
path: root/src/gun_http2.erl
diff options
context:
space:
mode:
Diffstat (limited to 'src/gun_http2.erl')
-rw-r--r--src/gun_http2.erl143
1 files changed, 82 insertions, 61 deletions
diff --git a/src/gun_http2.erl b/src/gun_http2.erl
index 79d7841..eb6f06c 100644
--- a/src/gun_http2.erl
+++ b/src/gun_http2.erl
@@ -65,7 +65,12 @@
%% Currently active HTTP/2 streams. Streams may be initiated either
%% by the client or by the server through PUSH_PROMISE frames.
- streams = [] :: [#stream{}]
+ %%
+ %% Streams can be found by ID or by Ref. The most common should be
+ %% the idea, that's why the main map has the ID as key. Then we also
+ %% have a Ref->ID index for faster lookup when we only have the Ref.
+ streams = #{} :: #{cow_http2:streamid() => #stream{}},
+ stream_refs = #{} :: #{reference() => cow_http2:streamid()}
}).
check_options(Opts) ->
@@ -162,7 +167,7 @@ parse(Data, State0=#http2_state{status=Status, http2_machine=HTTP2Machine, strea
{connection_error(State0, Error), EvHandlerState0};
%% If we both received and sent a GOAWAY frame and there are no streams
%% currently running, we can close the connection immediately.
- more when Status =/= connected, Streams =:= [] ->
+ more when Status =/= connected, Streams =:= #{} ->
{[{state, State0#http2_state{buffer=Data, status=closing}}, close], EvHandlerState0};
%% Otherwise we enter the closing state.
more when Status =:= goaway ->
@@ -329,10 +334,9 @@ trailers_frame(State, StreamID, Trailers, EvHandler, EvHandlerState0) ->
EvHandlerState = EvHandler:response_end(ResponseEvent, EvHandlerState1),
{maybe_delete_stream(State, StreamID, remote, fin), EvHandlerState}.
-rst_stream_frame(State=#http2_state{streams=Streams0},
- StreamID, Reason, EvHandler, EvHandlerState0) ->
- case lists:keytake(StreamID, #stream.id, Streams0) of
- {value, #stream{ref=StreamRef, reply_to=ReplyTo}, Streams} ->
+rst_stream_frame(State0, StreamID, Reason, EvHandler, EvHandlerState0) ->
+ case take_stream(State0, StreamID) of
+ {#stream{ref=StreamRef, reply_to=ReplyTo}, State} ->
ReplyTo ! {gun_error, self(), StreamRef,
{stream_error, Reason, 'Stream reset by server.'}},
EvHandlerState = EvHandler:cancel(#{
@@ -341,14 +345,14 @@ rst_stream_frame(State=#http2_state{streams=Streams0},
endpoint => remote,
reason => Reason
}, EvHandlerState0),
- {State#http2_state{streams=Streams}, EvHandlerState};
- false ->
- {State, EvHandlerState0}
+ {State, EvHandlerState};
+ error ->
+ {State0, EvHandlerState0}
end.
%% Pushed streams receive the same initial flow value as the parent stream.
push_promise_frame(State=#http2_state{socket=Socket, transport=Transport,
- status=Status, http2_machine=HTTP2Machine0, streams=Streams},
+ status=Status, http2_machine=HTTP2Machine0},
StreamID, PromisedStreamID, Headers, #{
method := Method, scheme := Scheme,
authority := Authority, path := Path},
@@ -375,7 +379,7 @@ push_promise_frame(State=#http2_state{socket=Socket, transport=Transport,
connected ->
NewStream = #stream{id=PromisedStreamID, ref=PromisedStreamRef,
reply_to=ReplyTo, flow=InitialFlow},
- {State#http2_state{streams=[NewStream|Streams]}, EvHandlerState};
+ {create_stream(State, NewStream), EvHandlerState};
%% We cancel the push_promise immediately when we are shutting down.
_ ->
{ok, HTTP2Machine} = cow_http2_machine:reset_stream(PromisedStreamID, HTTP2Machine0),
@@ -406,7 +410,7 @@ update_flow(State, _ReplyTo, StreamRef, Inc) ->
true ->
{state, store_stream(State, Stream#stream{flow=Flow})}
end;
- false ->
+ error ->
[]
end.
@@ -443,9 +447,13 @@ update_window(State=#http2_state{socket=Socket, transport=Transport,
%% GOAWAY frames as the LastStreamID value may be lower than
%% the one previously received.
goaway(State0=#http2_state{socket=Socket, transport=Transport, http2_machine=HTTP2Machine,
- status=Status, streams=Streams0}, {goaway, LastStreamID, Reason, _}) ->
- Streams = goaway_streams(Streams0, LastStreamID, {goaway, Reason, 'The connection is going away.'}, []),
- State = State0#http2_state{streams=Streams},
+ status=Status, streams=Streams0, stream_refs=Refs}, {goaway, LastStreamID, Reason, _}) ->
+ {Streams, RemovedRefs} = goaway_streams(maps:to_list(Streams0), LastStreamID,
+ {goaway, Reason, 'The connection is going away.'}, [], []),
+ State = State0#http2_state{
+ streams=maps:from_list(Streams),
+ stream_refs=maps:without(RemovedRefs, Refs)
+ },
case Status of
connected ->
Transport:send(Socket, cow_http2:goaway(
@@ -457,14 +465,14 @@ goaway(State0=#http2_state{socket=Socket, transport=Transport, http2_machine=HTT
end.
%% Cancel server-initiated streams that are above LastStreamID.
-goaway_streams([], _, _, Acc) ->
- Acc;
-goaway_streams([Stream=#stream{id=StreamID}|Tail], LastStreamID, Reason, Acc)
+goaway_streams([], _, _, Acc, RefsAcc) ->
+ {Acc, RefsAcc};
+goaway_streams([{StreamID, Stream=#stream{ref=StreamRef}}|Tail], LastStreamID, Reason, Acc, RefsAcc)
when StreamID > LastStreamID, (StreamID rem 2) =:= 1 ->
close_stream(Stream, Reason),
- goaway_streams(Tail, LastStreamID, Reason, Acc);
-goaway_streams([Stream|Tail], LastStreamID, Reason, Acc) ->
- goaway_streams(Tail, LastStreamID, Reason, [Stream|Acc]).
+ goaway_streams(Tail, LastStreamID, Reason, Acc, [StreamRef|RefsAcc]);
+goaway_streams([StreamWithID|Tail], LastStreamID, Reason, Acc, RefsAcc) ->
+ goaway_streams(Tail, LastStreamID, Reason, [StreamWithID|Acc], RefsAcc).
%% We are already closing, do nothing.
closing(_, #http2_state{status=closing}, _, EvHandlerState) ->
@@ -488,19 +496,16 @@ closing(#http2_state{opts=Opts}) ->
Timeout = maps:get(closing_timeout, Opts, 15000),
{closing, Timeout}.
-close(Reason, #http2_state{streams=Streams}, _, EvHandlerState) ->
- close_streams(Streams, close_reason(Reason)),
+close(Reason0, #http2_state{streams=Streams}, _, EvHandlerState) ->
+ Reason = close_reason(Reason0),
+ _ = maps:fold(fun(_, Stream, _) ->
+ close_stream(Stream, Reason)
+ end, [], Streams),
EvHandlerState.
close_reason(closed) -> closed;
close_reason(Reason) -> {closed, Reason}.
-close_streams([], _) ->
- ok;
-close_streams([Stream|Tail], Reason) ->
- close_stream(Stream, Reason),
- close_streams(Tail, Reason).
-
%% @todo Do we want an event for this?
close_stream(#stream{ref=StreamRef, reply_to=ReplyTo}, Reason) ->
ReplyTo ! {gun_error, self(), StreamRef, Reason},
@@ -511,9 +516,8 @@ keepalive(State=#http2_state{socket=Socket, transport=Transport}, _, EvHandlerSt
{State, EvHandlerState}.
headers(State=#http2_state{socket=Socket, transport=Transport, opts=Opts,
- http2_machine=HTTP2Machine0, streams=Streams},
- StreamRef, ReplyTo, Method, Host, Port, Path, Headers0,
- InitialFlow0, EvHandler, EvHandlerState0) ->
+ http2_machine=HTTP2Machine0}, StreamRef, ReplyTo, Method, Host, Port,
+ Path, Headers0, InitialFlow0, EvHandler, EvHandlerState0) ->
{ok, StreamID, HTTP2Machine1} = cow_http2_machine:init_stream(
iolist_to_binary(Method), HTTP2Machine0),
{ok, PseudoHeaders, Headers} = prepare_headers(State, Method, Host, Port, Path, Headers0),
@@ -533,13 +537,11 @@ headers(State=#http2_state{socket=Socket, transport=Transport, opts=Opts,
EvHandlerState = EvHandler:request_headers(RequestEvent, EvHandlerState1),
InitialFlow = initial_flow(InitialFlow0, Opts),
Stream = #stream{id=StreamID, ref=StreamRef, reply_to=ReplyTo, flow=InitialFlow},
- {State#http2_state{http2_machine=HTTP2Machine, streams=[Stream|Streams]},
- EvHandlerState}.
+ {create_stream(State#http2_state{http2_machine=HTTP2Machine}, Stream), EvHandlerState}.
request(State0=#http2_state{socket=Socket, transport=Transport, opts=Opts,
- http2_machine=HTTP2Machine0, streams=Streams},
- StreamRef, ReplyTo, Method, Host, Port, Path, Headers0, Body,
- InitialFlow0, EvHandler, EvHandlerState0) ->
+ http2_machine=HTTP2Machine0}, StreamRef, ReplyTo, Method, Host, Port,
+ Path, Headers0, Body, InitialFlow0, EvHandler, EvHandlerState0) ->
Headers1 = lists:keystore(<<"content-length">>, 1, Headers0,
{<<"content-length">>, integer_to_binary(iolist_size(Body))}),
{ok, StreamID, HTTP2Machine1} = cow_http2_machine:init_stream(
@@ -565,7 +567,7 @@ request(State0=#http2_state{socket=Socket, transport=Transport, opts=Opts,
EvHandlerState = EvHandler:request_headers(RequestEvent, EvHandlerState1),
InitialFlow = initial_flow(InitialFlow0, Opts),
Stream = #stream{id=StreamID, ref=StreamRef, reply_to=ReplyTo, flow=InitialFlow},
- State = State0#http2_state{http2_machine=HTTP2Machine, streams=[Stream|Streams]},
+ State = create_stream(State0#http2_state{http2_machine=HTTP2Machine}, Stream),
case IsFin of
fin ->
RequestEndEvent = #{
@@ -617,7 +619,7 @@ data(State=#http2_state{http2_machine=HTTP2Machine}, StreamRef, ReplyTo, IsFin,
{ok, _, _} ->
maybe_send_data(State, StreamID, IsFin, Data, EvHandler, EvHandlerState)
end;
- false ->
+ error ->
{error_stream_not_found(State, StreamRef, ReplyTo), EvHandlerState}
end.
@@ -677,15 +679,15 @@ send_data_frame(State=#http2_state{socket=Socket, transport=Transport,
Transport:send(Socket, cow_http2:headers(StreamID, fin, HeaderBlock)),
State#http2_state{http2_machine=HTTP2Machine}.
-reset_stream(State=#http2_state{socket=Socket, transport=Transport,
- streams=Streams0}, StreamID, StreamError={stream_error, Reason, _}) ->
+reset_stream(State0=#http2_state{socket=Socket, transport=Transport},
+ StreamID, StreamError={stream_error, Reason, _}) ->
Transport:send(Socket, cow_http2:rst_stream(StreamID, Reason)),
- case lists:keytake(StreamID, #stream.id, Streams0) of
- {value, #stream{ref=StreamRef, reply_to=ReplyTo}, Streams} ->
+ case take_stream(State0, StreamID) of
+ {#stream{ref=StreamRef, reply_to=ReplyTo}, State} ->
ReplyTo ! {gun_error, self(), StreamRef, StreamError},
- State#http2_state{streams=Streams};
- false ->
- State
+ State;
+ error ->
+ State0
end.
cancel(State=#http2_state{socket=Socket, transport=Transport, http2_machine=HTTP2Machine0},
@@ -702,7 +704,7 @@ cancel(State=#http2_state{socket=Socket, transport=Transport, http2_machine=HTTP
}, EvHandlerState0),
{delete_stream(State#http2_state{http2_machine=HTTP2Machine}, StreamID),
EvHandlerState};
- false ->
+ error ->
{error_stream_not_found(State, StreamRef, ReplyTo),
EvHandlerState0}
end.
@@ -715,17 +717,19 @@ stream_info(State, StreamRef) ->
reply_to => ReplyTo,
state => running
}};
- false ->
+ error ->
{ok, undefined}
end.
-down(#http2_state{streams=Streams}) ->
- [Ref || #stream{ref=Ref} <- Streams].
+down(#http2_state{stream_refs=Refs}) ->
+ maps:keys(Refs).
connection_error(#http2_state{socket=Socket, transport=Transport,
http2_machine=HTTP2Machine, streams=Streams},
{connection_error, Reason, _}) ->
- Pids = lists:usort([ReplyTo || #stream{reply_to=ReplyTo} <- Streams]),
+ Pids = lists:usort(maps:fold(
+ fun(_, #stream{reply_to=ReplyTo}, Acc) -> [ReplyTo|Acc] end,
+ [], Streams)),
_ = [Pid ! {gun_error, self(), Reason} || Pid <- Pids],
Transport:send(Socket, cow_http2:goaway(
cow_http2_machine:get_last_streamid(HTTP2Machine),
@@ -745,17 +749,31 @@ error_stream_not_found(State, StreamRef, ReplyTo) ->
State.
%% Streams.
-%% @todo probably change order of args and have state first? Yes.
get_stream_by_id(#http2_state{streams=Streams}, StreamID) ->
- lists:keyfind(StreamID, #stream.id, Streams).
+ maps:get(StreamID, Streams).
+
+get_stream_by_ref(#http2_state{streams=Streams, stream_refs=Refs}, StreamRef) ->
+ case maps:get(StreamRef, Refs, error) of
+ error -> error;
+ StreamID -> maps:get(StreamID, Streams)
+ end.
-get_stream_by_ref(#http2_state{streams=Streams}, StreamRef) ->
- lists:keyfind(StreamRef, #stream.ref, Streams).
+create_stream(State=#http2_state{streams=Streams, stream_refs=Refs},
+ Stream=#stream{id=StreamID, ref=StreamRef}) ->
+ State#http2_state{
+ streams=Streams#{StreamID => Stream},
+ stream_refs=Refs#{StreamRef => StreamID}
+ }.
-store_stream(State=#http2_state{streams=Streams0}, Stream=#stream{id=StreamID}) ->
- Streams = lists:keyreplace(StreamID, #stream.id, Streams0, Stream),
- State#http2_state{streams=Streams}.
+store_stream(State=#http2_state{streams=Streams}, Stream=#stream{id=StreamID}) ->
+ State#http2_state{streams=Streams#{StreamID => Stream}}.
+
+take_stream(State=#http2_state{streams=Streams0}, StreamID) ->
+ case maps:take(StreamID, Streams0) of
+ {Stream, Streams} -> {Stream, State#http2_state{streams=Streams}};
+ error -> error
+ end.
maybe_delete_stream(State=#http2_state{http2_machine=HTTP2Machine}, StreamID, local, fin) ->
case cow_http2_machine:get_stream_remote_state(StreamID, HTTP2Machine) of
@@ -772,6 +790,9 @@ maybe_delete_stream(State=#http2_state{http2_machine=HTTP2Machine}, StreamID, re
maybe_delete_stream(State, _, _, _) ->
State.
-delete_stream(State=#http2_state{streams=Streams}, StreamID) ->
- Streams2 = lists:keydelete(StreamID, #stream.id, Streams),
- State#http2_state{streams=Streams2}.
+delete_stream(State=#http2_state{streams=Streams, stream_refs=Refs}, StreamID) ->
+ #{StreamID := #stream{ref=StreamRef}} = Streams,
+ State#http2_state{
+ streams=maps:remove(StreamID, Streams),
+ stream_refs=maps:remove(StreamRef, Refs)
+ }.