path: root/src/gun_pool.erl

%% Copyright (c) 2021-2023, 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(gun_pool).
-behaviour(gen_statem).

%% Pools.
-export([start_pool/3]).
-export([stop_pool/2]).
-export([stop_pool/3]).
% @todo shutdown pool?
-export([info/0]).
-export([info/1]).
-export([info/2]).
-export([await_up/1]).
-export([await_up/2]).
-export([checkout/2]). %% Use responsibly!

%% Requests.
-export([delete/2]).
-export([delete/3]).
-export([get/2]).
-export([get/3]).
-export([head/2]).
-export([head/3]).
-export([options/2]).
-export([options/3]).
-export([patch/2]).
-export([patch/3]).
-export([patch/4]).
-export([post/2]).
-export([post/3]).
-export([post/4]).
-export([put/2]).
-export([put/3]).
-export([put/4]).

%% Generic requests interface.
-export([headers/3]).
-export([headers/4]).
-export([request/4]).
-export([request/5]).

%% Streaming data.
-export([data/3]).

%% Tunneling. (HTTP/2+ only.)
%% @todo -export([connect/2]).
%% @todo -export([connect/3]).
%% @todo -export([connect/4]).

%% Cookies.
%% @todo -export([gc_cookies/1]).
%% @todo -export([session_gc_cookies/1]).

%% Awaiting gun messages.
-export([await/1]).
-export([await/2]).
-export([await/3]).
-export([await_body/1]).
-export([await_body/2]).
-export([await_body/3]).

%% Flushing gun messages.
-export([flush/1]).

%% Streams.
-export([update_flow/2]).
-export([cancel/1]).
-export([stream_info/1]).

%% Websocket. (HTTP/2+ only for upgrade.)
%% -export([ws_upgrade/1]).
%% -export([ws_upgrade/2]).
%% -export([ws_upgrade/3]).
-export([ws_send/2]).
%% -export([ws_send/3]). (HTTP/2+ only.)

%% Internals.
-export([callback_mode/0]).
-export([start_link/3]).
-export([init/1]).
-export([degraded/3]).
-export([operational/3]).
-export([terminate/3]).

-type setup_msg() :: {gun_up, pid(), http | http2 | raw | socks}
	| {gun_upgrade, pid(), gun:stream_ref(), [binary()], [{binary(), binary()}]}.

-type opts() :: #{
	conn_opts => gun:opts(),
	scope => any(),
	setup_fun => {fun((pid(), setup_msg(), any()) -> any()), any()},
	size => non_neg_integer()
}.
-export_type([opts/0]).

-type pool_stream_ref() :: {pid(), gun:stream_ref()}.
-export_type([pool_stream_ref/0]).

-type error_result() :: {error, pool_not_found | no_connection_available, atom()}. %% @todo {pool_start_error, SupError}

-type request_result() :: {async, pool_stream_ref()}
	%% @todo {sync, ...} perhaps with Status, Headers, Body, and Extra info such as intermediate responses.
	| error_result().

-type req_opts() :: #{
	%% Options common with normal Gun.
	flow => pos_integer(),
	reply_to => pid(),
%	@todo tunnel => stream_ref(),

	%% Options specific to pools.
	checkout_call_timeout => timeout(),
	checkout_retry => [pos_integer()],
	scope => any(),
	start_pool_if_missing => boolean()
}.
-export_type([req_opts/0]).

-type ws_send_opts() :: #{
	authority => iodata(),

	%% Options specific to pools.
	checkout_call_timeout => timeout(),
	checkout_retry => [pos_integer()],
	scope => any(),
	start_pool_if_missing => boolean()
}.
-export_type([ws_send_opts/0]).

%% @todo tunnel
-type meta() :: #{pid() => #{ws => gun:stream_ref()}}.

-record(state, {
	host :: inet:hostname() | inet:ip_address(),
	port :: inet:port_number(),
	opts :: opts(),
	table :: ets:tid(),
	conns :: #{pid() => down | {setup, any()} | {up, http | http2 | ws | raw, map()}},
	conns_meta = #{} :: meta(),
	await_up = [] :: [{pid(), any()}]
}).

%% Pool management.

-spec start_pool(inet:hostname() | inet:ip_address(), inet:port_number(), opts())
	-> {ok, pid()} | {error, any()}.
start_pool(Host, Port, Opts) ->
	supervisor:start_child(gun_pools_sup, [Host, Port, Opts]).

-spec stop_pool(inet:hostname() | inet:ip_address(), inet:port_number())
	-> ok.
stop_pool(Host, Port) ->
	stop_pool(Host, Port, #{}).

-type stop_opts() :: #{
	scope => any(),
	transport => tcp | tls
}.

-spec stop_pool(inet:hostname() | inet:ip_address(), inet:port_number(), stop_opts())
	-> ok | {error, pool_not_found, atom()}.
stop_pool(Host, Port, StopOpts) ->
	Transport = maps:get(transport, StopOpts, gun:default_transport(Port)),
	Authority = gun_http:host_header(Transport, Host, Port),
	case get_pool(Authority, StopOpts) of
		undefined ->
			{error, pool_not_found,
				'No pool was found for the given scope and authority.'};
		ManagerPid ->
			supervisor:terminate_child(gun_pools_sup, ManagerPid)
	end.

-spec info() -> [map()].
info() ->
	ets:foldl(fun({{Scope, _}, ManagerPid}, Acc) ->
		{StateName, Info} = info(ManagerPid),
		[Info#{scope => Scope, state => StateName}|Acc]
	end, [], gun_pools).

-spec info(pid() | binary()) -> undefined | {degraded | operational, map()}.
info(ManagerPid) when is_pid(ManagerPid) ->
	gen_statem:call(ManagerPid, info);
info(Authority) ->
	info(Authority, default).

-spec info(binary(), any()) -> undefined | {degraded | operational, map()}.
info(Authority, Scope) ->
	case ets:lookup(gun_pools, {Scope, Authority}) of
		[] ->
			undefined;
		[{_, ManagerPid}] ->
			gen_statem:call(ManagerPid, info)
	end.

-spec await_up(pid() | binary()) -> ok | {error, pool_not_found, atom()}.
await_up(ManagerPid) when is_pid(ManagerPid) ->
	gen_statem:call(ManagerPid, await_up, 5000);
await_up(Authority) ->
	await_up(Authority, default).

-spec await_up(binary(), any()) -> ok | {error, pool_not_found, atom()}.
await_up(Authority, Scope) ->
	case ets:lookup(gun_pools, {Scope, Authority}) of
		[] ->
			{error, pool_not_found,
				'No pool was found for the given scope and authority.'};
		[{_, ManagerPid}] ->
			gen_statem:call(ManagerPid, await_up, 5000)
	end.

-spec checkout(pid(), req_opts() | ws_send_opts()) -> undefined | {pid(), map()}.
checkout(ManagerPid, ReqOpts=#{checkout_retry := Retry}) when is_list(Retry) ->
	CallTimeout = maps:get(checkout_call_timeout, ReqOpts, 5000),
	case gen_server:call(ManagerPid, {checkout, ReqOpts}, CallTimeout) of
		undefined ->
			checkout_retry(ManagerPid, ReqOpts, CallTimeout, Retry);
		Result ->
			Result
	end;
checkout(ManagerPid, ReqOpts) ->
	CallTimeout = maps:get(checkout_call_timeout, ReqOpts, 5000),
	gen_server:call(ManagerPid, {checkout, ReqOpts}, CallTimeout).

%% When the checkout_retry option is used, and the first call resulted
%% in no connection being given out, we wait for the configured amount
%% of time then try again. We loop over the wait times until there is
%% none.
checkout_retry(_, _, _, []) ->
	undefined;
checkout_retry(ManagerPid, ReqOpts, CallTimeout, [Wait|Retry]) ->
	timer:sleep(Wait),
	case gen_server:call(ManagerPid, {checkout, ReqOpts}, CallTimeout) of
		undefined ->
			checkout_retry(ManagerPid, ReqOpts, CallTimeout, Retry);
		Result ->
			Result
	end.

%% Requests.

-spec delete(iodata(), gun:req_headers()) -> request_result().
delete(Path, Headers) ->
	request(<<"DELETE">>, Path, Headers, <<>>).

-spec delete(iodata(), gun:req_headers(), req_opts()) -> request_result().
delete(Path, Headers, ReqOpts) ->
	request(<<"DELETE">>, Path, Headers, <<>>, ReqOpts).

-spec get(iodata(), gun:req_headers()) -> request_result().
get(Path, Headers) ->
	request(<<"GET">>, Path, Headers, <<>>).

-spec get(iodata(), gun:req_headers(), req_opts()) -> request_result().
get(Path, Headers, ReqOpts) ->
	request(<<"GET">>, Path, Headers, <<>>, ReqOpts).

-spec head(iodata(), gun:req_headers()) -> request_result().
head(Path, Headers) ->
	request(<<"HEAD">>, Path, Headers, <<>>).

-spec head(iodata(), gun:req_headers(), req_opts()) -> request_result().
head(Path, Headers, ReqOpts) ->
	request(<<"HEAD">>, Path, Headers, <<>>, ReqOpts).

-spec options(iodata(), gun:req_headers()) -> request_result().
options(Path, Headers) ->
	request(<<"OPTIONS">>, Path, Headers, <<>>).

-spec options(iodata(), gun:req_headers(), req_opts()) -> request_result().
options(Path, Headers, ReqOpts) ->
	request(<<"OPTIONS">>, Path, Headers, <<>>, ReqOpts).

-spec patch(iodata(), gun:req_headers()) -> request_result().
patch(Path, Headers) ->
	headers(<<"PATCH">>, Path, Headers).

-spec patch(iodata(), gun:req_headers(), iodata() | req_opts()) -> request_result().
patch(Path, Headers, ReqOpts) when is_map(ReqOpts) ->
	headers(<<"PATCH">>, Path, Headers, ReqOpts);
patch(Path, Headers, Body) ->
	request(<<"PATCH">>, Path, Headers, Body).

-spec patch(iodata(), gun:req_headers(), iodata(), req_opts()) -> request_result().
patch(Path, Headers, Body, ReqOpts) ->
	request(<<"PATCH">>, Path, Headers, Body, ReqOpts).

-spec post(iodata(), gun:req_headers()) -> request_result().
post(Path, Headers) ->
	headers(<<"POST">>, Path, Headers).

-spec post(iodata(), gun:req_headers(), iodata() | req_opts()) -> request_result().
post(Path, Headers, ReqOpts) when is_map(ReqOpts) ->
	headers(<<"POST">>, Path, Headers, ReqOpts);
post(Path, Headers, Body) ->
	request(<<"POST">>, Path, Headers, Body).

-spec post(iodata(), gun:req_headers(), iodata(), req_opts()) -> request_result().
post(Path, Headers, Body, ReqOpts) ->
	request(<<"POST">>, Path, Headers, Body, ReqOpts).

-spec put(iodata(), gun:req_headers()) -> request_result().
put(Path, Headers) ->
	headers(<<"PUT">>, Path, Headers).

-spec put(iodata(), gun:req_headers(), iodata() | req_opts()) -> request_result().
put(Path, Headers, ReqOpts) when is_map(ReqOpts) ->
	headers(<<"PUT">>, Path, Headers, ReqOpts);
put(Path, Headers, Body) ->
	request(<<"PUT">>, Path, Headers, Body).

-spec put(iodata(), gun:req_headers(), iodata(), req_opts()) -> request_result().
put(Path, Headers, Body, ReqOpts) ->
	request(<<"PUT">>, Path, Headers, Body, ReqOpts).

%% Generic requests interface.
%%
%% @todo Accept a TargetURI map as well as a normal Path.

-spec headers(iodata(), iodata(), gun:req_headers()) -> request_result().
headers(Method, Path, Headers) ->
	headers(Method, Path, Headers, #{}).

-spec headers(iodata(), iodata(), gun:req_headers(), req_opts()) -> request_result().
headers(Method, Path, Headers, ReqOpts) ->
	case get_pool(authority(Headers), ReqOpts) of
		undefined ->
			{error, pool_not_found,
				'No pool was found for the given scope and authority.'};
		ManagerPid ->
			case checkout(ManagerPid, ReqOpts) of
				undefined ->
					{error, no_connection_available,
						'No connection in the pool with enough capacity available to open a new stream.'};
				{ConnPid, _Meta} ->
					StreamRef = gun:headers(ConnPid, Method, Path, Headers, ReqOpts),
					%% @todo Synchronous mode.
					{async, {ConnPid, StreamRef}}
			end
	end.

-spec request(iodata(), iodata(), gun:req_headers(), iodata()) -> request_result().
request(Method, Path, Headers, Body) ->
	request(Method, Path, Headers, Body, #{}).

-spec request(iodata(), iodata(), gun:req_headers(), iodata(), req_opts()) -> request_result().
request(Method, Path, Headers, Body, ReqOpts) ->
	case get_pool(authority(Headers), ReqOpts) of
		undefined ->
			{error, pool_not_found,
				'No pool was found for the given scope and authority.'};
		ManagerPid ->
			case checkout(ManagerPid, ReqOpts) of
				undefined ->
					{error, no_connection_available,
						'No connection in the pool with enough capacity available to open a new stream.'};
				{ConnPid, _Meta} ->
					StreamRef = gun:request(ConnPid, Method, Path, Headers, Body, ReqOpts),
					%% @todo Synchronous mode.
					{async, {ConnPid, StreamRef}}
			end
	end.

%% We require the host to be given in the headers for the time being.
%% @todo Allow passing it in options. Websocket send already does that.
authority(#{<<"host">> := Authority}) ->
	Authority;
authority(Headers) ->
	{_, Authority} = lists:keyfind(<<"host">>, 1, Headers),
	Authority.

-spec get_pool(iolist(), req_opts() | ws_send_opts() | stop_opts()) -> pid() | undefined.
get_pool(Authority0, ReqOpts) ->
	Authority = iolist_to_binary(Authority0),
	%% @todo Perhaps rename this to temporary.
	%% There's two concepts: temporary pool is started and stops
	%% when there's no traffic. Dynamic pool simply reduces its number
	%% of connections when there's no traffic.
	StartPoolIfMissing = maps:get(start_pool_if_missing, ReqOpts, false),
	Scope = maps:get(scope, ReqOpts, default),
	case ets:lookup(gun_pools, {Scope, Authority}) of
		[] when StartPoolIfMissing ->
			start_missing_pool(Authority, ReqOpts);
		[] ->
			undefined;
		[{_, ManagerPid}] ->
			%% @todo With temporary pool, getting a pid here doesn't mean the pool can be used.
			%% Indeed the manager could be in process of stopping. I suppose we must
			%% do a check but perhaps it's best to leave that detail to the user
			%% (they can easily retry and recreate the pool if necessary).
			ManagerPid
	end.

start_missing_pool(_Authority, _ReqOpts) ->
	undefined.

%% Streaming data.

-spec data(pool_stream_ref(), fin | nofin, iodata()) -> ok.
data({ConnPid, StreamRef}, IsFin, Data) ->
	gun:data(ConnPid, StreamRef, IsFin, Data).

%% Awaiting gun messages.

-spec await(pool_stream_ref()) -> gun:await_result().
await({ConnPid, StreamRef}) ->
	gun:await(ConnPid, StreamRef).

-spec await(pool_stream_ref(), timeout() | reference()) -> gun:await_result().
await({ConnPid, StreamRef}, MRefOrTimeout) ->
	gun:await(ConnPid, StreamRef, MRefOrTimeout).

-spec await(pool_stream_ref(), timeout(), reference()) -> gun:await_result().
await({ConnPid, StreamRef}, Timeout, MRef) ->
	gun:await(ConnPid, StreamRef, Timeout, MRef).

-spec await_body(pool_stream_ref()) -> gun:await_body_result().
await_body({ConnPid, StreamRef}) ->
	gun:await_body(ConnPid, StreamRef).

-spec await_body(pool_stream_ref(), timeout() | reference()) -> gun:await_body_result().
await_body({ConnPid, StreamRef}, MRefOrTimeout) ->
	gun:await_body(ConnPid, StreamRef, MRefOrTimeout).

-spec await_body(pool_stream_ref(), timeout(), reference()) -> gun:await_body_result().
await_body({ConnPid, StreamRef}, Timeout, MRef) ->
	gun:await_body(ConnPid, StreamRef, Timeout, MRef).

%% Flushing gun messages.

-spec flush(pool_stream_ref()) -> ok.
flush({ConnPid, _}) ->
	gun:flush(ConnPid).

%% Flow control.

-spec update_flow(pool_stream_ref(), pos_integer()) -> ok.
update_flow({ConnPid, StreamRef}, Flow) ->
	gun:update_flow(ConnPid, StreamRef, Flow).

%% Cancelling a stream.

-spec cancel(pool_stream_ref()) -> ok.
cancel({ConnPid, StreamRef}) ->
	gun:cancel(ConnPid, StreamRef).

%% Information about a stream.

-spec stream_info(pool_stream_ref()) -> {ok, map() | undefined} | {error, not_connected}.
stream_info({ConnPid, StreamRef}) ->
	gun:stream_info(ConnPid, StreamRef).

%% Websocket.

-spec ws_send(gun:ws_frame() | [gun:ws_frame()], ws_send_opts()) -> ok | error_result().
ws_send(Frames, WsSendOpts=#{authority := Authority}) ->
	case get_pool(Authority, WsSendOpts) of
		undefined ->
			{error, pool_not_found,
				'No pool was found for the given scope and authority.'};
		ManagerPid ->
			case checkout(ManagerPid, WsSendOpts) of
				undefined ->
					{error, no_connection_available,
						'No connection in the pool with enough capacity available to send Websocket frames.'};
				{ConnPid, #{ws := StreamRef}} ->
					gun:ws_send(ConnPid, StreamRef, Frames)
			end
	end.

%% Pool manager internals.
%%
%% The pool manager is responsible for starting connection processes
%% and restarting them as necessary. It also provides a suitable
%% connection process to any caller that needs it.
%%
%% The pool manager installs an event handler into each connection.
%% The event handler is responsible for counting the number of
%% active streams. It updates the gun_pooled_conns ets table
%% whenever a stream begins or ends.
%%
%% A connection is deemed suitable if it is possible to open new
%% streams. How many streams can be open at any one time depends
%% on the protocol. For HTTP/2 the manager process keeps track of
%% the connection's settings to know the maximum. For non-stream
%% based protocols, there is no limit.
%%
%% The connection to be used is otherwise chosen randomly. The
%% first connection that is suitable is returned. There is no
%% need to "give back" the connection to the manager.

%% @todo
%% What should happen if we always fail to reconnect? I suspect we keep the manager
%% around and propagate errors, the same as if there's no more capacity? Perhaps have alarms?

callback_mode() -> state_functions.

start_link(Host, Port, Opts) ->
	gen_statem:start_link(?MODULE, {Host, Port, Opts}, []).

init({Host, Port, Opts}) ->
	process_flag(trap_exit, true),
	true = ets:insert_new(gun_pools, {gun_pools_key(Host, Port, Opts), self()}),
	Tid = ets:new(gun_pooled_conns, [ordered_set, public]),
	Size = maps:get(size, Opts, 8),
	%% @todo Only start processes in static mode.
	ConnOpts = conn_opts(Tid, Opts),
	Conns = maps:from_list([begin
		{ok, ConnPid} = gun:open(Host, Port, ConnOpts),
		_ = monitor(process, ConnPid),
		{ConnPid, down}
	end || _ <- lists:seq(1, Size)]),
	State = #state{
		host=Host,
		port=Port,
		opts=Opts,
		table=Tid,
		conns=Conns
	},
	%% If Size is 0 then we can never be operational.
	{ok, degraded, State}.

gun_pools_key(Host, Port, Opts) ->
	Transport = maps:get(transport, Opts, gun:default_transport(Port)),
	Authority = gun_http:host_header(Transport, Host, Port),
	Scope = maps:get(scope, Opts, default),
	{Scope, iolist_to_binary(Authority)}.

conn_opts(Tid, Opts) ->
	ConnOpts = maps:get(conn_opts, Opts, #{}),
	EventHandlerState = maps:with([event_handler], ConnOpts),
	H2Opts = maps:get(http2_opts, ConnOpts, #{}),
	ConnOpts#{
		event_handler => {gun_pool_events_h, EventHandlerState#{
			table => Tid
		}},
		http2_opts => H2Opts#{
			notify_settings_changed => true
		}
	}.

%% We use the degraded state as long as at least one connection is degraded.
%% @todo Probably keep count of connections separately to avoid counting every time.
degraded(info, Msg={gun_up, ConnPid, _}, StateData=#state{opts=Opts, conns=Conns}) ->
	#{ConnPid := down} = Conns,
	%% We optionally run the setup function if one is defined. The
	%% setup function tells us whether we are fully up or not. The
	%% setup function may be called repeatedly until the connection
	%% is established.
	%%
	%% @todo It is possible that the connection never does get
	%% fully established. We should deal with this. We probably
	%% need to handle all messages.
	{SetupFun, SetupState0} = setup_fun(Opts),
	degraded_setup(ConnPid, Msg, StateData, SetupFun, SetupState0);
%% @todo
%degraded(info, Msg={gun_tunnel_up, ConnPid, _, _}, StateData0=#state{conns=Conns}) ->
%	;
degraded(info, Msg={gun_upgrade, ConnPid, _, _, _},
		StateData=#state{opts=#{setup_fun := {SetupFun, _}}, conns=Conns}) ->
	%% @todo Probably shouldn't crash if the state is incorrect, that's programmer error though.
	#{ConnPid := {setup, SetupState0}} = Conns,
	%% We run the setup function again using the state previously kept.
	degraded_setup(ConnPid, Msg, StateData, SetupFun, SetupState0);
degraded(Type, Event, StateData) ->
	handle_common(Type, Event, ?FUNCTION_NAME, StateData).

setup_fun(#{setup_fun := SetupFun}) ->
	SetupFun;
setup_fun(_) ->
	{fun (_, {gun_up, _, Protocol}, _) ->
		{up, Protocol, #{}}
	end, undefined}.

degraded_setup(ConnPid, Msg, StateData0=#state{conns=Conns, conns_meta=ConnsMeta,
		await_up=AwaitUp}, SetupFun, SetupState0) ->
	case SetupFun(ConnPid, Msg, SetupState0) of
		Setup={setup, _SetupState} ->
			StateData = StateData0#state{conns=Conns#{ConnPid => Setup}},
			{keep_state, StateData};
		%% The Meta is different from Settings. It allows passing around
		%% Websocket or tunnel stream refs.
		{up, Protocol, Meta} ->
			Settings = #{},
			StateData = StateData0#state{
				conns=Conns#{ConnPid => {up, Protocol, Settings}},
				conns_meta=ConnsMeta#{ConnPid => Meta}
			},
			case is_degraded(StateData) of
				true -> {keep_state, StateData};
				false -> {next_state, operational, StateData#state{await_up=[]},
					[{reply, ReplyTo, ok} || ReplyTo <- AwaitUp]}
			end
	end.

is_degraded(#state{conns=Conns0}) ->
	Conns = maps:to_list(Conns0),
	Len = length(Conns),
	Ups = [up || {_, {up, _, _}} <- Conns],
	Len =/= length(Ups).

operational(Type, Event, StateData) ->
	handle_common(Type, Event, ?FUNCTION_NAME, StateData).

handle_common({call, From}, {checkout, _ReqOpts}, _,
		StateData=#state{conns_meta=ConnsMeta}) ->
	case find_available_connection(StateData) of
		none ->
			{keep_state_and_data, {reply, From, undefined}};
		ConnPid ->
			Meta = maps:get(ConnPid, ConnsMeta, #{}),
			{keep_state_and_data, {reply, From, {ConnPid, Meta}}}
	end;
handle_common(info, {gun_notify, ConnPid, settings_changed, Settings}, _, StateData=#state{conns=Conns}) ->
	%% Assert that the state is correct.
	{up, http2, _} = maps:get(ConnPid, Conns),
	{keep_state, StateData#state{conns=Conns#{ConnPid => {up, http2, Settings}}}};
handle_common(info, {gun_down, ConnPid, Protocol, _Reason, _KilledStreams}, _, StateData=#state{conns=Conns}) ->
	{up, Protocol, _} = maps:get(ConnPid, Conns),
	{next_state, degraded, StateData#state{conns=Conns#{ConnPid => down}}};
%% @todo We do not want to reconnect automatically when the pool is dynamic.
handle_common(info, {'DOWN', _MRef, process, ConnPid0, Reason}, _,
		StateData=#state{host=Host, port=Port, opts=Opts, table=Tid, conns=Conns0, conns_meta=ConnsMeta0}) ->
	Conns = maps:remove(ConnPid0, Conns0),
	ConnsMeta = maps:remove(ConnPid0, ConnsMeta0),
	case Reason of
		%% The process is down because of a configuration error.
		%% Do NOT attempt to reconnect, leave the pool in a degraded state.
		badarg ->
			{next_state, degraded, StateData#state{conns=Conns, conns_meta=ConnsMeta}};
		_ ->
			ConnOpts = conn_opts(Tid, Opts),
			{ok, ConnPid} = gun:open(Host, Port, ConnOpts),
			_ = monitor(process, ConnPid),
			{next_state, degraded, StateData#state{conns=Conns#{ConnPid => down}, conns_meta=ConnsMeta}}
	end;
handle_common({call, From}, info, StateName, #state{host=Host, port=Port,
		opts=Opts, table=Tid, conns=Conns, conns_meta=ConnsMeta}) ->
	{keep_state_and_data, {reply, From, {StateName, #{
		%% @todo Not sure whether all of this should be documented. Maybe not ConnsMeta for now?
		host => Host,
		port => Port,
		opts => Opts,
		table => Tid,
		conns => Conns,
		conns_meta => ConnsMeta
	}}}};
handle_common({call, From}, await_up, operational, _) ->
	{keep_state_and_data, {reply, From, ok}};
handle_common({call, From}, await_up, _, StateData=#state{await_up=AwaitUp}) ->
	{keep_state, StateData#state{await_up=[From|AwaitUp]}};
handle_common(Type, Event, StateName, StateData) ->
	logger:error("Unexpected event in state ~p of type ~p:~n~w~n~p~n",
		[StateName, Type, Event, StateData]),
	keep_state_and_data.

%% We go over every connection and return the first one
%% we find that has capacity. How we determine whether
%% capacity is available depends on the protocol. For
%% HTTP/2 we look into the protocol settings. The
%% current number of streams is maintained by the
%% event handler gun_pool_events_h.
find_available_connection(#state{table=Tid, conns=Conns}) ->
	I = lists:sort([{rand:uniform(), K} || K <- maps:keys(Conns)]),
	find_available_connection(I, Conns, Tid).

find_available_connection([], _, _) ->
	none;
find_available_connection([{_, ConnPid}|I], Conns, Tid) ->
	case maps:get(ConnPid, Conns) of
		{up, Protocol, Settings} ->
			MaxStreams = max_streams(Protocol, Settings),
			CurrentStreams = case ets:lookup(Tid, ConnPid) of
				[] ->
					0;
				[{_, CS}] ->
					CS
			end,
			if
				CurrentStreams + 1 > MaxStreams ->
					find_available_connection(I, Conns, Tid);
				true ->
					ConnPid
			end;
		_ ->
			find_available_connection(I, Conns, Tid)
	end.

max_streams(http, _) ->
	1;
max_streams(http2, #{max_concurrent_streams := MaxStreams}) ->
	MaxStreams;
max_streams(http2, #{}) ->
	infinity;
%% There are no streams or Gun is not aware of streams when
%% the protocol is Websocket or raw.
max_streams(ws, _) ->
	infinity;
max_streams(raw, _) ->
	infinity.

terminate(Reason, StateName, #state{host=Host, port=Port, opts=Opts, await_up=AwaitUp}) ->
	gen_statem:reply([
		{reply, ReplyTo, {error, {terminate, StateName, Reason}}}
	|| ReplyTo <- AwaitUp]),
	true = ets:delete(gun_pools, gun_pools_key(Host, Port, Opts)),
	ok.