Current section
Files
Jump to
Current section
Files
src/h1.erl
%% Copyright (c) 2026 Benoit Chesneau.
%% SPDX-License-Identifier: Apache-2.0
%%
%% @doc HTTP/1.1 public API.
%%
%% Mirrors the surface of `h2' (HTTP/2) and `quic_h3' (HTTP/3) so
%% applications can swap protocols without rewriting call sites.
%%
%% == Client ==
%% ```
%% {ok, Conn} = h1:connect("example.com", 80, #{}).
%% {ok, StreamId} = h1:request(Conn, <<"GET">>, <<"/">>,
%% [{<<"host">>, <<"example.com">>}]).
%% receive
%% {h1, Conn, {response, StreamId, Status, _Headers}} -> ok
%% end.
%% ok = h1:close(Conn).
%% '''
%%
%% == Server ==
%% ```
%% {ok, S} = h1:start_server(8080, #{
%% transport => tcp,
%% handler => fun(Conn, Id, _Method, _Path, _Hs) ->
%% h1:send_response(Conn, Id, 200, [{<<"content-length">>, <<"2">>}]),
%% h1:send_data(Conn, Id, <<"ok">>, true)
%% end}).
%% '''
-module(h1).
%% 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, stop_accepting/1,
server_port/1]).
-export([serve_socket/2]).
-export([send_response/4, respond/5, respond/6]).
-export([send_informational/4]).
%% Common API
-export([send_data/3, send_data/4]).
-export([send_trailers/3]).
-export([cancel/2, cancel/3]).
-export([cancel_stream/2, cancel_stream/3]).
-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([peername/1]).
-export([controlling_process/2]).
%% HTTP/1.1-specific
-export([upgrade/3, upgrade/4]).
-export([accept_upgrade/3]).
-export([accept_connect/3, accept_connect/4]).
-export([continue/2]).
-export([pipeline/2]).
-type connection() :: pid().
-type stream_id() :: non_neg_integer().
-type headers() :: [{binary(), binary()}].
-type status() :: 100..599.
-type server_ref() :: {pid(), reference(), inet:port_number()}.
-type connect_opts() :: #{
transport => tcp | ssl,
ssl_opts => [ssl:tls_client_option()],
connect_timeout => timeout(),
timeout => timeout(),
pipeline => boolean(),
max_keepalive_requests => pos_integer(),
max_header_block_size => pos_integer(),
idle_timeout => timeout(),
request_timeout => timeout()
}.
-type server_opts() :: #{
transport => tcp | ssl,
cert => binary() | string(),
key => binary() | string(),
cacerts => [binary()],
verify => verify_none | verify_peer,
ssl_opts => [ssl:tls_option()],
ip => inet:ip_address(),
inet6 => boolean(),
handler := fun((connection(), stream_id(), binary(), binary(), headers()) -> any())
| module(),
acceptors => pos_integer(),
handshake_timeout => timeout(),
idle_timeout => timeout(),
request_timeout => timeout(),
early_response_drain => early_response_drain(),
lingering_timeout => timeout(),
pipeline => boolean(),
max_keepalive_requests => pos_integer(),
max_line_length => pos_integer(),
max_request_line_size => pos_integer() | infinity,
max_empty_lines => non_neg_integer(),
max_header_name_size => pos_integer(),
max_header_value_size => pos_integer(),
max_headers => pos_integer(),
max_header_block_size => pos_integer(),
max_body_size => pos_integer() | infinity
}.
%% Early-response inbound drain budget (lingering close). `{MaxBytes, MaxMs}'
%% reads and discards an unfinished request body after an early response, up
%% to MaxBytes / MaxMs (either component `infinity'), before closing. `0'
%% disables the drain and closes immediately. Default `{infinity, 30000}'.
-type early_response_drain() ::
0 | {non_neg_integer() | infinity, non_neg_integer() | infinity}.
-type respond_opts() :: #{early_response_drain => early_response_drain()}.
-export_type([connection/0, stream_id/0, headers/0, status/0, server_ref/0,
connect_opts/0, server_opts/0, early_response_drain/0,
respond_opts/0]).
%% ============================================================================
%% Client
%% ============================================================================
-spec connect(string() | binary(), inet:port_number()) ->
{ok, connection()} | {error, term()}.
connect(Host, Port) ->
connect(Host, Port, #{}).
-spec connect(string() | binary(), inet:port_number(), connect_opts()) ->
{ok, connection()} | {error, term()}.
connect(Host, Port, Opts) ->
h1_client:connect(Host, Port, Opts).
-spec wait_connected(connection()) -> ok | {error, term()}.
wait_connected(Conn) -> h1_connection:wait_connected(Conn).
-spec wait_connected(connection(), timeout()) -> ok | {error, term()}.
wait_connected(Conn, Timeout) -> h1_connection:wait_connected(Conn, Timeout).
%% @doc Send a request using h2-compatible pseudo-headers. The list
%% may contain `:method', `:path', `:authority'; they're translated
%% into the HTTP/1.1 request line + `Host' header.
-spec request(connection(), headers()) ->
{ok, stream_id()} | {error, term()}.
request(Conn, Headers) ->
request(Conn, Headers, #{}).
-spec request(connection(), headers(), map()) ->
{ok, stream_id()} | {error, term()}.
request(Conn, Headers, Opts) ->
{Method, Path, Rest0} = extract_pseudo(Headers),
Rest1 = case proplists:is_defined(<<"host">>, Rest0) of
true -> Rest0;
false ->
case proplists:get_value(<<":authority">>, Headers) of
undefined -> Rest0;
Authority -> [{<<"host">>, Authority} | Rest0]
end
end,
SendOpts = maps:without([protocol, end_stream], Opts),
case maps:get(body, Opts, undefined) of
undefined ->
h1_connection:send_request(Conn, Method, Path, Rest1, SendOpts);
Body ->
h1_connection:send_request(Conn, Method, Path, Rest1,
SendOpts#{body => Body,
end_stream => true})
end.
-spec request(connection(), binary(), binary(), headers()) ->
{ok, stream_id()} | {error, term()}.
request(Conn, Method, Path, Headers) ->
h1_connection:send_request(Conn, Method, Path, Headers, #{}).
-spec request(connection(), binary(), binary(), headers(), binary()) ->
{ok, stream_id()} | {error, term()}.
request(Conn, Method, Path, Headers, Body) ->
h1_connection:send_request(Conn, Method, Path, Headers,
#{body => Body, end_stream => true}).
extract_pseudo(Headers) ->
Method = proplists:get_value(<<":method">>, Headers, <<"GET">>),
Path = proplists:get_value(<<":path">>, Headers, <<"/">>),
Rest = [{N, V} || {N, V} <- Headers,
N =/= <<":method">>,
N =/= <<":path">>,
N =/= <<":scheme">>,
N =/= <<":authority">>,
N =/= <<":protocol">>],
{Method, Path, Rest}.
%% ============================================================================
%% Server
%% ============================================================================
-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.
-spec start_server(inet:port_number(), server_opts()) ->
{ok, server_ref()} | {error, term()}.
start_server(Port, Opts) ->
Transport = maps:get(transport, Opts, tcp),
case maps:find(handler, Opts) of
{ok, _} ->
case Transport of
tcp -> start_tcp(Port, Opts);
ssl -> start_ssl(Port, Opts)
end;
error ->
{error, {missing_required_option, [handler]}}
end.
start_tcp(Port, Opts) ->
TcpOpts = [binary, {active, false}, {packet, raw},
{reuseaddr, true}, {backlog, 1024}, {nodelay, true}]
++ socket_addr_opts(Opts),
case gen_tcp:listen(Port, TcpOpts) of
{ok, ListenSocket} ->
{ok, {_, Bound}} = inet:sockname(ListenSocket),
spawn_listener(gen_tcp, ListenSocket, Bound, Opts);
{error, Reason} ->
{error, {listen_failed, Reason}}
end.
start_ssl(Port, Opts) ->
case {maps:find(cert, Opts), maps:find(key, Opts)} of
{{ok, Cert}, {ok, Key}} ->
case server_ssl_opts(Cert, Key, Opts) of
{ok, Listen} ->
case ssl:listen(Port, Listen) of
{ok, ListenSocket} ->
{ok, {_, Bound}} = ssl:sockname(ListenSocket),
spawn_listener(ssl, ListenSocket, Bound, Opts);
{error, Reason} ->
{error, {listen_failed, Reason}}
end;
{error, _} = Error ->
Error
end;
_ ->
{error, {missing_required_option, [cert, key]}}
end.
%% Build the ssl:listen/2 option list. Honours `verify' (default
%% `verify_none'), `cacerts' (needed for mutual TLS), and `ssl_opts' as a raw
%% override. `verify_peer' without CA certificates is rejected so a
%% misconfigured server fails closed instead of accepting unauthenticated
%% peers — same contract as h2.
server_ssl_opts(Cert, Key, Opts) ->
Verify = maps:get(verify, Opts, verify_none),
CACerts = maps:get(cacerts, Opts, []),
case Verify of
verify_peer when CACerts =:= [] ->
{error, verify_peer_requires_cacerts};
_ ->
Defaults = [binary, {active, false}, {packet, raw},
{reuseaddr, true}, {backlog, 1024}, {nodelay, true},
{certfile, to_list(Cert)}, {keyfile, to_list(Key)},
{alpn_preferred_protocols, [<<"http/1.1">>]},
{verify, Verify}],
%% `verify_peer' alone lets a TLS 1.2 client skip its certificate
%% (`fail_if_no_peer_cert' defaults to false), which is not what
%% asking for client authentication means. Require the
%% certificate; `ssl_opts' can still override.
Auth = case {Verify, CACerts} of
{_, []} -> [];
{verify_peer, _} -> [{cacerts, CACerts},
{fail_if_no_peer_cert, true}];
{_, _} -> [{cacerts, CACerts}]
end,
SslOpts = maps:get(ssl_opts, Opts, []),
Base = Defaults ++ Auth ++ socket_addr_opts(Opts),
{ok, merge(Base, SslOpts)}
end.
spawn_listener(Transport, ListenSocket, Bound, Opts) ->
Handler = maps:get(handler, Opts),
Acceptors = maps:get(acceptors, Opts, erlang:system_info(schedulers)),
ConnOpts = conn_opts(Opts),
ServerOpts = maps:with([handshake_timeout], Opts),
Ref = make_ref(),
Args = #{transport => Transport,
listen_socket => ListenSocket,
acceptor_count => Acceptors,
ref => Ref,
handler => Handler,
conn_opts => ConnOpts,
server_opts => ServerOpts},
case h1_sup:start_listener(Args) of
{ok, Pid} ->
{ok, {Pid, Ref, Bound}};
{error, Reason} ->
close_socket(Transport, ListenSocket),
{error, Reason}
end.
close_socket(gen_tcp, S) -> _ = gen_tcp:close(S), ok;
close_socket(ssl, S) -> _ = ssl:close(S), ok.
%% The per-connection slice of the server options. Every path that starts a
%% server connection (`spawn_listener/4', `serve_socket/2') goes through
%% this one list, so an option cannot be wired into one path and forgotten
%% in the other.
conn_opts(Opts) ->
maps:with([idle_timeout, request_timeout,
early_response_drain, lingering_timeout,
max_keepalive_requests, pipeline,
max_line_length, max_request_line_size, max_empty_lines,
max_header_name_size, max_header_value_size,
max_headers, max_header_block_size,
max_body_size], Opts).
%% @doc Serve an already-accepted connection: run the server loop on a
%% socket someone else accepted. The caller has completed the TCP accept
%% and, for TLS, the handshake and ALPN negotiation — h1 does not
%% handshake again, so `cert', `key', `verify' and `ssl_opts' are ignored
%% here. Use it to serve HTTP/1.1 and HTTP/2 on one TLS port: negotiate
%% ALPN yourself, then hand `http/1.1' sockets to this function.
%%
%% Requires `handler'; every other option is the per-connection subset
%% `start_server/2' accepts (timeouts, parser limits, drain budget).
%%
%% The socket must be passive (`{active, false}'): h1 arms it itself, and
%% bytes an active socket already delivered to the caller's mailbox cannot
%% be recovered.
%%
%% On `{ok, Pid}' the returned process is linked to the caller and owns
%% the socket: it is the socket's controlling process from that point on,
%% and closes it when it exits (unless the socket was handed off by
%% Upgrade or CONNECT). Killing the caller therefore closes the
%% connection. These connections belong to no listener, so
%% `stop_server/1' does not close them.
%%
%% ```
%% {ok, Sock} = ssl:handshake(Raw, 5000),
%% {ok, <<"http/1.1">>} = ssl:negotiated_protocol(Sock),
%% {ok, _Pid} = h1:serve_socket(Sock, #{handler => fun my_app:handle/5}).
%% '''
-spec serve_socket(ssl:sslsocket() | gen_tcp:socket(), server_opts()) ->
{ok, pid()} | {error, term()}.
serve_socket(Socket, Opts) ->
case maps:find(handler, Opts) of
{ok, Handler} ->
Transport = h1_connection:transport_of(Socket),
Args = [Socket, Transport, Handler, conn_opts(Opts), #{}],
Pid = proc_lib:spawn_link(h1_server, init_serve, Args),
case transfer_socket(Transport, Socket, Pid) of
ok ->
Pid ! {h1_server, socket_ready},
{ok, Pid};
{error, Reason} ->
Pid ! {h1_server, transfer_failed},
_ = close_socket(Transport, Socket),
{error, {controlling_process, Reason}}
end;
error ->
{error, {missing_required_option, [handler]}}
end.
transfer_socket(gen_tcp, Socket, Pid) -> gen_tcp:controlling_process(Socket, Pid);
transfer_socket(ssl, Socket, Pid) -> ssl:controlling_process(Socket, Pid).
%% @doc Stop a server. Synchronous: closes the listen socket, the
%% acceptor pool, and every accepted connection (kept-alive and
%% in-flight included) before returning.
-spec stop_server(server_ref()) -> ok.
stop_server({Pid, Ref, _Port} = ServerRef) ->
erase_server_term(ServerRef),
h1_listener:stop(Pid, Ref).
%% @doc Stop accepting new connections while continuing to serve the
%% established ones (graceful drain). Synchronous. Call
%% `stop_server/1' afterwards to close the remaining connections.
-spec stop_accepting(server_ref()) -> ok.
stop_accepting({Pid, Ref, _Port}) ->
h1_listener:stop_accepting(Pid, Ref).
%% Drop any name->ref registration made by start_server/3 so repeated
%% start/stop cycles don't leak persistent_term entries.
erase_server_term(ServerRef) ->
_ = [persistent_term:erase(K)
|| {{?MODULE, server, _} = K, V} <- persistent_term:get(),
V =:= ServerRef],
ok.
-spec server_port(server_ref()) -> inet:port_number().
server_port({_, _, Port}) -> Port.
-spec send_response(connection(), stream_id(), status(), headers()) ->
ok | {error, term()}.
send_response(Conn, StreamId, Status, Headers) ->
h1_connection:send_response(Conn, StreamId, Status, Headers).
-spec respond(connection(), stream_id(), status(), headers(), iodata()) ->
ok | {error, term()}.
%% @doc Server: send a complete response (status, headers, and body) in a
%% single socket write and end the stream. A Content-Length is added when
%% the headers carry neither Content-Length nor Transfer-Encoding, so the
%% body is sent fixed-length rather than chunked. Use this for fully-known
%% bodies; use send_response/4 + send_data/4 for streaming.
%%
%% If the request body has not been fully received when this is called (an
%% early response, e.g. rejecting an oversized upload with 413), h1 adds
%% `Connection: close', sends the response, then drains and discards the rest
%% of the inbound body before closing the socket. The response is delivered
%% cleanly and the connection is not reused. The drain is bounded by the
%% listener's `early_response_drain' budget (default `{infinity, 30000}').
respond(Conn, StreamId, Status, Headers, Body) ->
h1_connection:respond(Conn, StreamId, Status, Headers, Body).
-spec respond(connection(), stream_id(), status(), headers(), iodata(),
respond_opts()) -> ok | {error, term()}.
%% @doc As respond/5, with per-response options. `early_response_drain'
%% overrides the listener's drain budget for this response only: pass a
%% larger `{MaxBytes, MaxMs}' to a known-large upload endpoint, or `0' to
%% close immediately without draining.
respond(Conn, StreamId, Status, Headers, Body, Opts) ->
h1_connection:respond(Conn, StreamId, Status, Headers, Body, Opts).
%% ============================================================================
%% Common
%% ============================================================================
-spec send_data(connection(), stream_id(), binary()) -> ok | {error, term()}.
send_data(Conn, StreamId, Data) ->
h1_connection:send_data(Conn, StreamId, Data).
-spec send_data(connection(), stream_id(), binary(), boolean()) ->
ok | {error, term()}.
%% @doc Send a body chunk. With `EndStream = true' the stream is ended; if the
%% request body was not fully received first (server side), h1 advertises
%% `Connection: close' and drains the remaining inbound body before closing,
%% as for respond/5.
send_data(Conn, StreamId, Data, EndStream) ->
h1_connection:send_data(Conn, StreamId, Data, EndStream).
-spec send_trailers(connection(), stream_id(), headers()) ->
ok | {error, term()}.
send_trailers(Conn, StreamId, Trailers) ->
h1_connection:send_trailers(Conn, StreamId, Trailers).
-spec cancel(connection(), stream_id()) -> ok | {error, term()}.
cancel(Conn, StreamId) ->
h1_connection:cancel_stream(Conn, StreamId).
-spec cancel(connection(), stream_id(), term()) -> ok | {error, term()}.
cancel(Conn, StreamId, Reason) ->
h1_connection:cancel_stream(Conn, StreamId, Reason).
-spec cancel_stream(connection(), stream_id()) -> ok | {error, term()}.
cancel_stream(Conn, StreamId) -> cancel(Conn, StreamId).
-spec cancel_stream(connection(), stream_id(), term()) -> ok | {error, term()}.
cancel_stream(Conn, StreamId, Reason) -> cancel(Conn, StreamId, Reason).
-spec set_stream_handler(connection(), stream_id(), pid()) ->
ok | {error, term()}.
set_stream_handler(Conn, StreamId, Pid) ->
h1_connection:set_stream_handler(Conn, StreamId, Pid).
-spec set_stream_handler(connection(), stream_id(), pid(), map()) ->
ok | {error, term()}.
set_stream_handler(Conn, StreamId, Pid, Opts) ->
h1_connection:set_stream_handler(Conn, StreamId, Pid, Opts).
-spec unset_stream_handler(connection(), stream_id()) -> ok.
unset_stream_handler(Conn, StreamId) ->
h1_connection:unset_stream_handler(Conn, StreamId).
-spec goaway(connection()) -> ok | {error, term()}.
goaway(Conn) -> h1_connection:send_goaway(Conn).
-spec goaway(connection(), term()) -> ok | {error, term()}.
goaway(Conn, Reason) -> h1_connection:send_goaway(Conn, Reason).
-spec close(connection()) -> ok.
close(Conn) -> h1_connection:close(Conn).
-spec get_settings(connection()) -> map().
get_settings(Conn) -> h1_connection:get_settings(Conn).
-spec get_peer_settings(connection()) -> map().
get_peer_settings(Conn) -> h1_connection:get_peer_settings(Conn).
%% @doc Address and port of the connected peer. Works in both modes:
%% the remote client on a server connection, the remote server on a
%% client connection.
-spec peername(connection()) ->
{ok, {inet:ip_address(), inet:port_number()}} | {error, term()}.
peername(Conn) -> h1_connection:peername(Conn).
-spec controlling_process(connection(), pid()) -> ok | {error, term()}.
controlling_process(Conn, Pid) ->
h1_connection:controlling_process(Conn, Pid).
%% ============================================================================
%% HTTP/1.1-specific
%% ============================================================================
%% @doc Client: send an Upgrade request and wait for 101 Switching Protocols.
-spec upgrade(connection(), binary(), headers()) ->
{ok, stream_id(), term(), binary(), headers()} | {error, term()}.
upgrade(Conn, Protocol, Headers) ->
h1_connection:upgrade(Conn, Protocol, Headers).
-spec upgrade(connection(), binary(), headers(), timeout()) ->
{ok, stream_id(), term(), binary(), headers()} | {error, term()}.
upgrade(Conn, Protocol, Headers, Timeout) ->
h1_connection:upgrade(Conn, Protocol, Headers, Timeout).
%% @doc Server: reply 101 Switching Protocols to an upgrade request
%% and take ownership of the raw socket. Injects the `Connection:
%% Upgrade' and `Upgrade: <proto>' framing headers itself and strips
%% any caller-supplied copies (case-insensitive), so the 101 carries
%% exactly one of each. Pass only protocol-specific extras in
%% ExtraHeaders (e.g. `sec-websocket-accept').
-spec accept_upgrade(connection(), stream_id(), headers()) ->
{ok, term(), binary()} | {error, term()}.
accept_upgrade(Conn, StreamId, ExtraHeaders) ->
h1_connection:accept_upgrade(Conn, StreamId, ExtraHeaders).
%% @doc Server: reply 200 Connection Established to a classic HTTP/1.1
%% CONNECT (RFC 9110 §9.3.6, RFC 9112 §3.2.3 authority-form request-target)
%% and take ownership of the raw socket. Mirror of `accept_upgrade/3' for
%% the 101 Switching Protocols case, but writes status 200 and injects no
%% Connection/Upgrade headers so bytes after CRLF belong to the tunnel.
-spec accept_connect(connection(), stream_id(), headers()) ->
{ok, gen_tcp | ssl, term(), binary()} | {error, term()}.
accept_connect(Conn, StreamId, ExtraHeaders) ->
h1_connection:accept_connect(Conn, StreamId, ExtraHeaders).
-spec accept_connect(connection(), stream_id(), headers(), timeout()) ->
{ok, gen_tcp | ssl, term(), binary()} | {error, term()}.
accept_connect(Conn, StreamId, ExtraHeaders, Timeout) ->
h1_connection:accept_connect(Conn, StreamId, ExtraHeaders, Timeout).
%% @doc Server: send 100 Continue to a client waiting on Expect.
-spec continue(connection(), stream_id()) -> ok | {error, term()}.
continue(Conn, StreamId) ->
h1_connection:continue(Conn, StreamId).
%% @doc Server: send an interim (1xx) response, e.g. 103 Early Hints,
%% ahead of the final response. May be called several times per stream,
%% until the final response headers are sent. 101 is rejected (use the
%% Upgrade machinery), and so are HTTP/1.0 clients (RFC 9110 §15.2).
-spec send_informational(connection(), stream_id(), 100..199, headers()) ->
ok | {error, term()}.
send_informational(Conn, StreamId, Status, Headers) ->
h1_connection:send_informational(Conn, StreamId, Status, Headers).
%% @doc Toggle request pipelining on a client connection.
-spec pipeline(connection(), boolean()) -> ok | {error, term()}.
pipeline(Conn, Enabled) when is_boolean(Enabled) ->
h1_connection:set_pipeline(Conn, Enabled).
%% ============================================================================
%% Internal
%% ============================================================================
%% 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 suitable for 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.
merge(Default, Override) ->
Key = fun({K, _}) -> K; (K) -> K end,
Kept = [D || D <- Default,
not lists:any(fun(O) -> Key(O) =:= Key(D) end, Override)],
Kept ++ Override.
to_list(P) when is_list(P) -> P;
to_list(P) when is_binary(P) -> binary_to_list(P).