Current section
Files
Jump to
Current section
Files
src/h2.erl
%% @doc HTTP/2 Public API
%%
%% This module provides the public API for HTTP/2 client and server operations.
%% It wraps the h2_connection state machine with a clean interface.
%%
%% == Client Usage ==
%%
%% ```
%% %% Connect to a server
%% {ok, Conn} = h2:connect("example.com", 443, #{}).
%%
%% %% Send a request
%% {ok, StreamId} = h2:request(Conn, <<"GET">>, <<"/">>, [
%% {<<"host">>, <<"example.com">>}
%% ]).
%%
%% %% Receive response (messages sent to caller)
%% receive
%% {h2, Conn, {response, StreamId, Status, Headers}} ->
%% io:format("Status: ~p~n", [Status]);
%% {h2, Conn, {data, StreamId, Data, IsFin}} ->
%% io:format("Data: ~p~n", [Data])
%% end.
%%
%% %% Close connection
%% ok = h2:close(Conn).
%% '''
%%
%% == Server Usage ==
%%
%% ```
%% %% Start server
%% {ok, Server} = h2:start_server(8443, #{
%% cert => "server.pem",
%% key => "server-key.pem",
%% handler => fun(Conn, StreamId, Method, Path, Headers) ->
%% h2:send_response(Conn, StreamId, 200, [{<<"content-type">>, <<"text/plain">>}]),
%% h2:send_data(Conn, StreamId, <<"Hello World!">>, true)
%% end
%% }).
%%
%% %% Stop server
%% ok = h2:stop_server(Server).
%% '''
%%
%% == Event Messages ==
%%
%% The owner process receives the following messages:
%%
%% Client:
%% - `{h2, Conn, {response, StreamId, Status, Headers}}'
%% - `{h2, Conn, {data, StreamId, Data, IsFin}}'
%% - `{h2, Conn, {trailers, StreamId, Trailers}}'
%% - `{h2, Conn, {stream_reset, StreamId, ErrorCode}}'
%% - `{h2, Conn, {goaway, LastStreamId, ErrorCode}}'
%% - `{h2, Conn, closed}'
%%
%% Server handler receives direct calls.
%%
-module(h2).
-include("h2.hrl").
%% Client API
-export([connect/2, connect/3]).
-export([wait_connected/1, wait_connected/2]).
-export([request/2, request/3, request/4, request/5]).
%% Server API
-export([start_server/2, start_server/3, stop_server/1, server_port/1]).
-export([send_response/4]).
-export([respond/5]).
%% Common API
-export([send_data/3, send_data/4, send_data/5]).
-export([consume/3]).
-export([send_trailers/3]).
-export([cancel/2, cancel/3]).
-export([cancel_stream/2, cancel_stream/3]).
-deprecated([{cancel_stream, 2, "use h2:cancel/2 instead"},
{cancel_stream, 3, "use h2:cancel/3 instead"}]).
-export([set_stream_handler/3, set_stream_handler/4, unset_stream_handler/2]).
-export([goaway/1, goaway/2]).
-export([close/1]).
-export([get_settings/1, get_peer_settings/1]).
-export([controlling_process/2]).
%% Types
-type connection() :: pid().
-type stream_id() :: non_neg_integer().
-type headers() :: [{binary(), binary()}].
-type status() :: 100..599.
-type error_code() :: h2_error:error_code().
-type server_ref() :: {pid(), reference(), inet:port_number()}.
-type connect_opts() :: #{
transport => tcp | ssl,
ssl_opts => [ssl:tls_client_option()],
cert => binary() | string(),
key => binary() | string(),
cacerts => [binary()],
verify => verify_none | verify_peer,
settings => h2_settings:settings(),
timeout => timeout(),
connect_timeout => timeout(),
sync => boolean()
}.
-type server_opts() :: #{
transport => ssl | tcp,
cert => binary() | string(),
key => binary() | string(),
cacerts => [binary()],
%% TLS peer-cert verification policy. Defaults to verify_none. When
%% verify_peer is requested, `cacerts' must be supplied; otherwise the
%% listener start fails with {error, verify_peer_requires_cacerts}.
verify => verify_none | verify_peer,
%% Raw ssl:tls_option() overrides; merged on top of our defaults. Useful
%% for pinning a cipher list, disabling renegotiation, etc.
ssl_opts => [ssl:tls_option()],
ip => inet:ip_address(),
inet6 => boolean(),
handler := fun((connection(), stream_id(), binary(), binary(), headers()) -> any()),
settings => h2_settings:settings(),
acceptors => pos_integer(),
%% RFC 8441: when true, advertise SETTINGS_ENABLE_CONNECT_PROTOCOL=1 and
%% accept Extended CONNECT requests carrying the `:protocol` pseudo-header.
enable_connect_protocol => boolean()
}.
-export_type([connection/0, stream_id/0, headers/0, status/0, error_code/0]).
-export_type([connect_opts/0, server_opts/0, server_ref/0]).
%% ============================================================================
%% Client API
%% ============================================================================
%% @doc Connect to an HTTP/2 server.
%% Uses TLS by default on port 443.
-spec connect(string() | binary(), inet:port_number()) ->
{ok, connection()} | {error, term()}.
connect(Host, Port) ->
connect(Host, Port, #{}).
%% @doc Connect to an HTTP/2 server with options.
-spec connect(string() | binary(), inet:port_number(), connect_opts()) ->
{ok, connection()} | {error, term()}.
connect(Host, Port, Opts) when is_binary(Host) ->
connect(binary_to_list(Host), Port, Opts);
connect(Host, Port, Opts) ->
Transport = maps:get(transport, Opts, ssl),
Timeout = maps:get(timeout, Opts, ?DEFAULT_TIMEOUT_MS),
case Transport of
ssl ->
connect_ssl(Host, Port, Opts, Timeout);
tcp ->
connect_tcp(Host, Port, Opts, Timeout)
end.
connect_ssl(Host, Port, Opts, Timeout) ->
DefaultSSLOpts = [
{active, false},
{mode, binary},
{alpn_advertised_protocols, [<<"h2">>]},
{versions, ['tlsv1.2', 'tlsv1.3']}
],
%% Honor top-level verify and cacerts from connect_opts(), merged with
%% any explicit ssl_opts (ssl_opts takes precedence).
TopLevel = [{verify, V} || V <- [maps:get(verify, Opts, undefined)], V =/= undefined]
++ [{cacerts, C} || C <- [maps:get(cacerts, Opts, undefined)], C =/= undefined],
SSLOpts0 = maps:get(ssl_opts, Opts, []),
SSLOpts = merge_opts(merge_opts(DefaultSSLOpts, TopLevel), SSLOpts0),
case ssl:connect(Host, Port, SSLOpts, Timeout) of
{ok, Socket} ->
%% Verify ALPN negotiated h2
case ssl:negotiated_protocol(Socket) of
{ok, <<"h2">>} ->
start_connection(client, Socket, Opts);
{ok, Other} ->
_ = ssl:close(Socket),
{error, {alpn_mismatch, Other}};
{error, protocol_not_negotiated} ->
%% RFC 9113 §3.3: TLS endpoints MUST use ALPN to
%% negotiate "h2". Don't fall back to assuming HTTP/2.
_ = ssl:close(Socket),
{error, alpn_not_negotiated};
{error, Reason} ->
_ = ssl:close(Socket),
{error, {alpn_error, Reason}}
end;
{error, Reason} ->
{error, Reason}
end.
connect_tcp(Host, Port, Opts, Timeout) ->
TCPOpts = [
{active, false},
{mode, binary},
{packet, raw}
],
case gen_tcp:connect(Host, Port, TCPOpts, Timeout) of
{ok, Socket} ->
start_connection(client, Socket, Opts);
{error, Reason} ->
{error, Reason}
end.
start_connection(Mode, Socket, Opts) ->
ConnOpts = maps:with([settings], Opts),
Timeout = maps:get(timeout, Opts, ?DEFAULT_TIMEOUT_MS),
case h2_connection:start_link(Mode, Socket, ConnOpts) of
{ok, Pid} ->
%% Transfer socket ownership to connection process
Transport = case is_ssl_socket(Socket) of
true -> ssl;
false -> gen_tcp
end,
TransferResult = case Transport of
ssl -> ssl:controlling_process(Socket, Pid);
gen_tcp -> gen_tcp:controlling_process(Socket, Pid)
end,
case TransferResult of
ok ->
_ = h2_connection:activate(Pid),
case h2_connection:wait_connected(Pid, Timeout) of
ok ->
{ok, Pid};
{error, Reason} ->
ignore_errors(fun() -> h2_connection:close(Pid) end),
{error, Reason}
end;
{error, TransferReason} ->
ignore_errors(fun() -> h2_connection:close(Pid) end),
ignore_errors(fun() -> close_socket(Transport, Socket) end),
{error, {controlling_process_failed, TransferReason}}
end;
{error, Reason} ->
{error, Reason}
end.
close_socket(ssl, Socket) -> ssl:close(Socket);
close_socket(gen_tcp, Socket) -> gen_tcp:close(Socket).
-spec ignore_errors(fun(() -> any())) -> ok.
ignore_errors(Fun) ->
try Fun() of
_ -> ok
catch
_:_ -> ok
end.
is_ssl_socket(Socket) when is_tuple(Socket) ->
element(1, Socket) =:= sslsocket;
is_ssl_socket(_) ->
false.
%% @doc Wait for a client connection to reach the connected state.
-spec wait_connected(connection()) -> ok | {error, term()}.
wait_connected(Conn) ->
h2_connection:wait_connected(Conn).
-spec wait_connected(connection(), timeout()) -> ok | {error, term()}.
wait_connected(Conn, Timeout) ->
h2_connection:wait_connected(Conn, Timeout).
%% @doc Send an HTTP/2 request with pre-built headers (matches quic_h3:request/2).
%% Headers should include pseudo-headers (:method, :path, :scheme, :authority).
-spec request(connection(), headers()) ->
{ok, stream_id()} | {error, term()}.
request(Conn, Headers) ->
request(Conn, Headers, #{}).
%% Opts may carry per-stream routing/backpressure options for gRPC-style use:
%% handler => pid() route this stream's events to pid from
%% creation (race-free; recommended for bidi)
%% defer => true buffer events until a later
%% set_stream_handler/3 replays them
%% flow_control => manual | auto manual = receive-side backpressure (consume/3)
%% end_stream => boolean() half-close the send side immediately
%% protocol => binary() RFC 8441 Extended CONNECT
-spec request(connection(), headers(), map()) ->
{ok, stream_id()} | {error, term()}.
request(Conn, Headers, Opts) ->
StreamOpts = stream_opts(Opts),
%% RFC 8441 Extended CONNECT: when `protocol` is supplied, inject the
%% `:protocol` pseudo-header (unless caller already included it) and
%% default end_stream to false (the stream stays open as a tunnel).
case maps:get(protocol, Opts, undefined) of
undefined ->
EndStream = maps:get(end_stream, Opts, true),
h2_connection:send_request_headers(Conn, Headers, EndStream, StreamOpts);
Protocol when is_binary(Protocol) ->
Headers1 = case proplists:is_defined(<<":protocol">>, Headers) of
true -> Headers;
false -> inject_protocol_pseudo(Headers, Protocol)
end,
EndStream = maps:get(end_stream, Opts, false),
h2_connection:send_request_headers(Conn, Headers1, EndStream, StreamOpts)
end.
%% Project the per-stream creation options out of the user opts map.
stream_opts(Opts) ->
lists:foldl(fun(Key, Acc) ->
case maps:find(Key, Opts) of
{ok, Value} -> Acc#{Key => Value};
error -> Acc
end
end, #{}, [handler, defer, flow_control]).
%% Insert `:protocol` after the trailing pseudo-header so the block stays
%% well-ordered (RFC 9113 §8.3: pseudo-headers must precede regular ones).
inject_protocol_pseudo(Headers, Protocol) ->
{Pseudos, Regular} = lists:splitwith(
fun({<<$:, _/binary>>, _}) -> true; (_) -> false end, Headers),
Pseudos ++ [{<<":protocol">>, Protocol} | Regular].
%% @doc Send an HTTP/2 request.
%% For body-less requests (GET, HEAD, etc.), sends HEADERS with END_STREAM.
%% For CONNECT (RFC 7540 §8.3) leaves the stream open so the caller can
%% send tunnel bytes via send_data.
-spec request(connection(), binary(), binary(), headers()) ->
{ok, stream_id()} | {error, term()}.
request(Conn, <<"CONNECT">> = Method, Path, Headers) ->
h2_connection:send_request(Conn, Method, Path, Headers, false);
request(Conn, Method, Path, Headers) ->
h2_connection:send_request(Conn, Method, Path, Headers, true).
%% @doc Send an HTTP/2 request with body.
%% Sends HEADERS without END_STREAM, then sends DATA with END_STREAM.
-spec request(connection(), binary(), binary(), headers(), binary()) ->
{ok, stream_id()} | {error, term()}.
request(Conn, Method, Path, Headers, Body) ->
case h2_connection:send_request(Conn, Method, Path, Headers, false) of
{ok, StreamId} ->
case h2_connection:send_data(Conn, StreamId, Body, true) of
ok -> {ok, StreamId};
{error, Reason} -> {error, Reason}
end;
{error, Reason} ->
{error, Reason}
end.
%% ============================================================================
%% Server API
%% ============================================================================
%% @doc Start a named HTTP/2 server (matches quic_h3:start_server/3).
-spec start_server(atom(), inet:port_number(), server_opts()) ->
{ok, server_ref()} | {error, term()}.
start_server(Name, Port, Opts) when is_atom(Name) ->
case start_server(Port, Opts) of
{ok, Ref} ->
persistent_term:put({?MODULE, server, Name}, Ref),
{ok, Ref};
Other ->
Other
end.
%% @doc Start an HTTP/2 server.
-spec start_server(inet:port_number(), server_opts()) ->
{ok, server_ref()} | {error, term()}.
start_server(Port, Opts) ->
Transport = maps:get(transport, Opts, ssl),
case Transport of
ssl -> start_server_ssl(Port, Opts);
tcp -> start_server_tcp(Port, Opts)
end.
start_server_ssl(Port, Opts) ->
case {maps:find(cert, Opts), maps:find(key, Opts), maps:find(handler, Opts)} of
{{ok, Cert}, {ok, Key}, {ok, Handler}} ->
case build_server_ssl_opts(Cert, Key, Opts) of
{error, _} = OptsErr -> OptsErr;
{ok, SSLOpts} ->
NumAcceptors = maps:get(acceptors, Opts, erlang:system_info(schedulers)),
Settings = maps:get(settings, Opts, #{}),
EnableConnectProtocol = maps:get(enable_connect_protocol, Opts, false),
case ssl:listen(Port, SSLOpts) of
{ok, ListenSocket} ->
{ok, {_, BoundPort}} = ssl:sockname(ListenSocket),
Ref = make_ref(),
ServerState = #{
listen_socket => ListenSocket,
handler => Handler,
settings => Settings,
enable_connect_protocol => EnableConnectProtocol,
ref => Ref
},
case h2_sup:start_listener(#{
transport => ssl,
listen_socket => ListenSocket,
acceptor_count => NumAcceptors,
ref => Ref,
acceptor_fun => fun(S) -> acceptor_loop(S) end,
server_state => ServerState
}) of
{ok, ListenerPid} ->
{ok, {ListenerPid, Ref, BoundPort}};
{error, StartReason} ->
_ = ssl:close(ListenSocket),
{error, StartReason}
end;
{error, Reason} ->
{error, {listen_failed, Reason}}
end
end;
_ ->
{error, {missing_required_option, [cert, key, handler]}}
end.
%% Build the final ssl:listen/2 option list from the user-supplied map.
%% Honors `verify' (default verify_none), `cacerts', and `ssl_opts' as a
%% raw override. Rejects verify_peer without cacerts so a misconfigured
%% server fails closed instead of silently accepting unauthenticated peers.
build_server_ssl_opts(Cert, Key, Opts) ->
CertFile = load_file(Cert),
KeyFile = load_file(Key),
Verify = maps:get(verify, Opts, verify_none),
CACerts = maps:get(cacerts, Opts, []),
UserOpts = maps:get(ssl_opts, Opts, []),
case Verify of
verify_peer when CACerts =:= [] ->
{error, verify_peer_requires_cacerts};
_ ->
Base = [
{certfile, CertFile},
{keyfile, KeyFile},
{alpn_preferred_protocols, [<<"h2">>]},
{versions, ['tlsv1.2', 'tlsv1.3']},
{honor_cipher_order, true},
{reuseaddr, true},
{active, false},
{mode, binary},
{backlog, maps:get(backlog, Opts, 1024)},
{verify, Verify}
],
Auth = case CACerts of
[] -> [];
_ -> [{cacerts, CACerts}]
end,
Addr = socket_addr_opts(Opts),
{ok, Addr ++ merge_opts(merge_opts(Base, Auth), UserOpts)}
end.
%% Cleartext (h2c over TCP, prior-knowledge) server listener.
start_server_tcp(Port, Opts) ->
case maps:find(handler, Opts) of
{ok, Handler} ->
NumAcceptors = maps:get(acceptors, Opts, erlang:system_info(schedulers)),
Settings = maps:get(settings, Opts, #{}),
EnableConnectProtocol = maps:get(enable_connect_protocol, Opts, false),
TCPOpts = [
{reuseaddr, true},
{active, false},
{mode, binary},
{packet, raw},
{backlog, maps:get(backlog, Opts, 1024)}
] ++ socket_addr_opts(Opts),
case gen_tcp:listen(Port, TCPOpts) of
{ok, ListenSocket} ->
{ok, {_, BoundPort}} = inet:sockname(ListenSocket),
Ref = make_ref(),
ServerState = #{
listen_socket => ListenSocket,
handler => Handler,
settings => Settings,
enable_connect_protocol => EnableConnectProtocol,
ref => Ref
},
case h2_sup:start_listener(#{
transport => tcp,
listen_socket => ListenSocket,
acceptor_count => NumAcceptors,
ref => Ref,
acceptor_fun => fun(S) -> tcp_acceptor_loop(S) end,
server_state => ServerState
}) of
{ok, ListenerPid} ->
{ok, {ListenerPid, Ref, BoundPort}};
{error, StartReason} ->
_ = gen_tcp:close(ListenSocket),
{error, StartReason}
end;
{error, Reason} ->
{error, {listen_failed, Reason}}
end;
_ ->
{error, {missing_required_option, [handler]}}
end.
%% @doc Stop an HTTP/2 server.
-spec stop_server(server_ref()) -> ok.
stop_server({ListenerPid, Ref, _Port}) ->
h2_listener:stop(ListenerPid, Ref).
%% @doc Return the TCP port the server is actually listening on.
-spec server_port(server_ref()) -> inet:port_number().
server_port({_, _, Port}) -> Port.
acceptor_loop(State) ->
process_flag(trap_exit, true),
acceptor_loop_inner(State).
%% Each accepted connection spawns a linked wrapper; when the wrapper
%% exits the acceptor (trap_exit) receives `{'EXIT', _, _}'. The accept
%% loop never reads its mailbox, so without a drain these queue up until
%% the node runs out of memory.
drain_child_exits() ->
receive
{'EXIT', _Pid, _Reason} -> drain_child_exits()
after 0 -> ok
end.
acceptor_loop_inner(#{listen_socket := ListenSocket, handler := Handler, settings := Settings} = State) ->
EnableConnectProtocol = maps:get(enable_connect_protocol, State, false),
drain_child_exits(),
case ssl:transport_accept(ListenSocket, infinity) of
{ok, Socket} ->
case ssl:handshake(Socket, ?DEFAULT_TIMEOUT_MS) of
{ok, SSLSocket} ->
%% Check ALPN
case ssl:negotiated_protocol(SSLSocket) of
{ok, <<"h2">>} ->
%% Ensure socket is in passive mode before transfer
_ = ssl:setopts(SSLSocket, [{active, false}]),
Pid = spawn_link(fun() ->
receive
{socket_ready, Sock} ->
handle_server_connection(Sock, Handler, Settings,
ssl, EnableConnectProtocol);
{socket_transfer_failed, _} ->
ok
end
end),
case ssl:controlling_process(SSLSocket, Pid) of
ok ->
Pid ! {socket_ready, SSLSocket},
ok;
{error, TransferReason} ->
Pid ! {socket_transfer_failed, TransferReason},
_ = ssl:close(SSLSocket),
ok
end;
_ ->
_ = ssl:close(SSLSocket),
ok
end;
{error, _} ->
ok
end,
acceptor_loop_inner(State);
{error, closed} ->
%% Listen socket closed (server shutdown). Exit with `shutdown`
%% so any wrapper/connection processes still spawn_link'd to us
%% (in-flight connections) tear down too.
exit(shutdown);
{error, _Reason} ->
acceptor_loop_inner(State)
end.
handle_server_connection(Socket, Handler, Settings, Transport, EnableConnectProtocol) ->
ConnOpts = #{settings => Settings,
enable_connect_protocol => EnableConnectProtocol},
TransferFn = case Transport of
ssl -> fun ssl:controlling_process/2;
gen_tcp -> fun gen_tcp:controlling_process/2
end,
CloseFn = case Transport of
ssl -> fun ssl:close/1;
gen_tcp -> fun gen_tcp:close/1
end,
case h2_connection:start_link(server, Socket, self(), ConnOpts) of
{ok, Conn} ->
case TransferFn(Socket, Conn) of
ok ->
_ = h2_connection:activate(Conn),
server_connection_loop(Conn, Handler);
{error, _} ->
ignore_errors(fun() -> h2_connection:close(Conn) end),
ignore_errors(fun() -> CloseFn(Socket) end)
end;
{error, _Reason} ->
ignore_errors(fun() -> CloseFn(Socket) end)
end.
tcp_acceptor_loop(State) ->
process_flag(trap_exit, true),
tcp_acceptor_loop_inner(State).
tcp_acceptor_loop_inner(#{listen_socket := ListenSocket, handler := Handler, settings := Settings} = State) ->
EnableConnectProtocol = maps:get(enable_connect_protocol, State, false),
drain_child_exits(),
case gen_tcp:accept(ListenSocket, infinity) of
{ok, Socket} ->
_ = inet:setopts(Socket, [{active, false}]),
Pid = spawn_link(fun() ->
receive
{socket_ready, Sock} ->
handle_server_connection(Sock, Handler, Settings,
gen_tcp, EnableConnectProtocol);
{socket_transfer_failed, _} ->
ok
end
end),
case gen_tcp:controlling_process(Socket, Pid) of
ok ->
Pid ! {socket_ready, Socket},
ok;
{error, TransferReason} ->
Pid ! {socket_transfer_failed, TransferReason},
_ = gen_tcp:close(Socket),
ok
end,
tcp_acceptor_loop_inner(State);
{error, closed} ->
exit(shutdown);
{error, _Reason} ->
tcp_acceptor_loop_inner(State)
end.
server_connection_loop(Conn, Handler) ->
receive
{h2, Conn, {request, StreamId, Method, Path, Headers}} ->
spawn(fun() ->
try
invoke_handler(Handler, Conn, StreamId, Method, Path, Headers)
catch
Class:Reason:Stack ->
logger:error("h2 handler crash: ~ts:~tp~n~tp",
[Class, Reason, Stack]),
ignore_errors(fun() -> h2:send_response(Conn, StreamId, 500, []) end),
ignore_errors(fun() -> h2:send_data(Conn, StreamId,
<<"Internal Server Error">>, true) end)
end
end),
server_connection_loop(Conn, Handler);
{h2, Conn, {data, _StreamId, _Data, _IsFin}} ->
%% Data received (for POST/PUT bodies)
server_connection_loop(Conn, Handler);
{h2, Conn, {closed, _Reason}} ->
ok;
{h2, Conn, {goaway, _LastStreamId, _ErrorCode}} ->
ok;
Other ->
logger:debug("h2 server_connection_loop: unexpected message ~tp",
[Other]),
server_connection_loop(Conn, Handler)
end.
invoke_handler(Handler, Conn, StreamId, Method, Path, Headers) when is_function(Handler, 5) ->
Handler(Conn, StreamId, Method, Path, Headers);
invoke_handler(Module, Conn, StreamId, Method, Path, Headers) when is_atom(Module) ->
Module:handle_request(Conn, StreamId, Method, Path, Headers).
%% @doc Send an HTTP/2 response (server mode).
-spec send_response(connection(), stream_id(), status(), headers()) ->
ok | {error, term()}.
send_response(Conn, StreamId, Status, Headers) ->
h2_connection:send_response(Conn, StreamId, Status, Headers).
%% @doc Send a complete response (headers + full body) in a single call. This is
%% the fast path for the common request/response case: one message to the
%% connection and one socket write (HEADERS coalesced with DATA), versus the two
%% round-trips of send_response/4 followed by send_data/4. Falls back to the
%% granular path transparently when the response cannot be coalesced (oversized
%% headers/body, CONNECT tunnels).
-spec respond(connection(), stream_id(), status(), headers(), binary()) ->
ok | {error, term()}.
respond(Conn, StreamId, Status, Headers, Body) ->
case h2_connection:respond(Conn, StreamId, Status, Headers, Body) of
need_fallback ->
case h2_connection:send_response(Conn, StreamId, Status, Headers) of
ok -> h2_connection:send_data(Conn, StreamId, Body, true);
Err -> Err
end;
Result ->
Result
end.
%% ============================================================================
%% Common API
%% ============================================================================
%% @doc Send data on a stream.
-spec send_data(connection(), stream_id(), binary()) -> ok | {error, term()}.
send_data(Conn, StreamId, Data) ->
h2_connection:send_data(Conn, StreamId, Data).
%% @doc Send data on a stream with end_stream flag.
%%
%% Non-blocking backpressure: when the peer's send window is exhausted the data
%% is buffered, and `{error, send_buffer_full}' is returned once the buffer would
%% exceed the per-stream cap, so the caller backs off instead of growing memory
%% without bound. For a blocking variant see send_data/5.
-spec send_data(connection(), stream_id(), binary(), boolean()) -> ok | {error, term()}.
send_data(Conn, StreamId, Data, EndStream) ->
h2_connection:send_data(Conn, StreamId, Data, EndStream).
%% @doc Send data with per-call options. Pass `#{block => Timeout}' (ms or
%% `infinity') to block until the peer's window accepts the data, returning `ok',
%% or `{error, timeout}' if the window does not open in time (the data may still
%% be queued; the caller should slow down or cancel the stream).
-spec send_data(connection(), stream_id(), binary(), boolean(), map()) ->
ok | {error, term()}.
send_data(Conn, StreamId, Data, EndStream, Opts) ->
h2_connection:send_data(Conn, StreamId, Data, EndStream, Opts).
%% @doc Acknowledge consumption of ByteCount received bytes on a manual
%% flow-control stream, replenishing its receive window. Receive-side
%% backpressure: a handler calls this only after processing data, gating the
%% peer's WINDOW_UPDATE on consumer progress. No-op on auto-mode streams.
-spec consume(connection(), stream_id(), non_neg_integer()) -> ok | {error, term()}.
consume(Conn, StreamId, ByteCount) ->
h2_connection:consume(Conn, StreamId, ByteCount).
%% @doc Send trailers on a stream.
-spec send_trailers(connection(), stream_id(), headers()) -> ok | {error, term()}.
send_trailers(Conn, StreamId, Trailers) ->
h2_connection:send_trailers(Conn, StreamId, Trailers).
%% @doc Cancel a stream.
-spec cancel(connection(), stream_id()) -> ok | {error, term()}.
cancel(Conn, StreamId) ->
h2_connection:cancel_stream(Conn, StreamId).
%% @doc Cancel a stream with a specific error code.
-spec cancel(connection(), stream_id(), error_code()) -> ok | {error, term()}.
cancel(Conn, StreamId, ErrorCode) ->
h2_connection:cancel_stream(Conn, StreamId, ErrorCode).
%% @deprecated Use {@link cancel/2} instead.
-spec cancel_stream(connection(), stream_id()) -> ok | {error, term()}.
cancel_stream(Conn, StreamId) ->
cancel(Conn, StreamId).
%% @deprecated Use {@link cancel/3} instead.
-spec cancel_stream(connection(), stream_id(), error_code()) -> ok | {error, term()}.
cancel_stream(Conn, StreamId, ErrorCode) ->
cancel(Conn, StreamId, ErrorCode).
%% @doc Register a pid to receive a stream's events.
%% Routes every event for the stream — `{response,...}', `{data,...}',
%% `{trailers,...}', `{informational,...}', `{stream_reset,...}' — to the
%% handler pid as `{h2, Conn, Event}'. By default the connection replays any
%% events buffered before the handler was registered, in arrival order, so a
%% response/trailers that raced ahead of registration is never dropped to the
%% owner. The call returns `ok'. Pass `#{drain_buffer => true}' to instead get
%% the buffered DATA back in the reply (`{ok, [{Data, Fin}, ...]}') and forward
%% it yourself (kept for the WebSocket/MASQUE tunnel callers).
%%
%% To avoid the registration race entirely, set the handler at stream creation
%% with `h2:request(Conn, Headers, #{handler => Pid})'.
%%
%% Backpressure: by default incoming DATA replenishes the receive window on
%% dispatch, so a slow handler's mailbox can grow unbounded. Pass
%% `#{flow_control => manual}' (here or at request time) to gate window
%% replenishment on consume/3 — the handler calls `h2:consume(Conn, StreamId, N)'
%% after processing N bytes, bounding in-flight data to one window.
-spec set_stream_handler(connection(), stream_id(), pid()) ->
ok | {ok, [{binary(), boolean()}]} | {error, term()}.
set_stream_handler(Conn, StreamId, Pid) ->
h2_connection:set_stream_handler(Conn, StreamId, Pid).
-spec set_stream_handler(connection(), stream_id(), pid(), map()) ->
ok | {ok, [{binary(), boolean()}]} | {error, term()}.
set_stream_handler(Conn, StreamId, Pid, Opts) ->
h2_connection:set_stream_handler(Conn, StreamId, Pid, Opts).
-spec unset_stream_handler(connection(), stream_id()) -> ok.
unset_stream_handler(Conn, StreamId) ->
h2_connection:unset_stream_handler(Conn, StreamId).
%% @doc Initiate graceful connection shutdown.
-spec goaway(connection()) -> ok | {error, term()}.
goaway(Conn) ->
h2_connection:send_goaway(Conn).
%% @doc Initiate connection shutdown with error code.
-spec goaway(connection(), error_code()) -> ok | {error, term()}.
goaway(Conn, ErrorCode) ->
h2_connection:send_goaway(Conn, ErrorCode).
%% @doc Close the connection immediately.
-spec close(connection()) -> ok.
close(Conn) ->
h2_connection:close(Conn).
%% @doc Get local settings.
-spec get_settings(connection()) -> h2_settings:settings().
get_settings(Conn) ->
h2_connection:get_settings(Conn).
%% @doc Get peer settings.
-spec get_peer_settings(connection()) -> h2_settings:settings().
get_peer_settings(Conn) ->
h2_connection:get_peer_settings(Conn).
%% @doc Transfer connection ownership.
-spec controlling_process(connection(), pid()) -> ok | {error, term()}.
controlling_process(Conn, NewOwner) ->
h2_connection:controlling_process(Conn, NewOwner).
%% ============================================================================
%% Internal Functions
%% ============================================================================
merge_opts(Default, Override) ->
lists:ukeymerge(1,
lists:ukeysort(1, Override),
lists:ukeysort(1, Default)).
%% Build inet listen options from the `ip'/`inet6' server opts. An IPv6
%% `ip' tuple (or `inet6 => true') selects the inet6 family; `ip' sets the
%% bind address. Returned as a list for the gen_tcp/ssl listen opts.
socket_addr_opts(Opts) ->
IP = maps:get(ip, Opts, undefined),
Family = case {IP, maps:get(inet6, Opts, false)} of
{{_, _, _, _, _, _, _, _}, _} -> [inet6];
{_, true} -> [inet6];
_ -> []
end,
Addr = case IP of
undefined -> [];
_ -> [{ip, IP}]
end,
Family ++ Addr.
load_file(Path) when is_list(Path) -> Path;
load_file(Path) when is_binary(Path) -> binary_to_list(Path).
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
merge_opts_test() ->
Default = [{a, 1}, {b, 2}],
Override = [{b, 3}, {c, 4}],
Merged = merge_opts(Default, Override),
?assertEqual(1, proplists:get_value(a, Merged)),
?assertEqual(3, proplists:get_value(b, Merged)),
?assertEqual(4, proplists:get_value(c, Merged)).
-endif.