%%% -*- erlang -*- %%% %%% This file is part of hackney released under the Apache 2 license. %%% See the NOTICE for more information. %%% %%% Copyright (c) 2024 Benoit Chesneau %%% %%% @doc gen_statem process for managing a single HTTP connection. %%% %%% This module implements a state machine for HTTP connections, %%% handling connection establishment, request/response cycles, %%% and connection reuse. %%% %%% States: %%% - idle: Process started, not connected %%% - connecting: TCP/SSL handshake in progress %%% - connected: Ready for requests %%% - sending: Sending request data %%% - receiving: Awaiting/streaming response %%% - closed: Connection terminated -module(hackney_conn). -behaviour(gen_statem). %% API -export([ start_link/1, stop/1, connect/1, connect/2, get_state/1, %% Request/Response (sync) request/5, request/6, request/7, request_streaming/5, body/1, body/2, stream_body/1, %% Streaming body (request body) send_request_headers/4, send_body_chunk/2, finish_send_body/1, start_response/1, %% Async streaming request_async/6, request_async/7, request_async/8, request_async/9, stream_next/1, stop_async/1, pause_stream/1, resume_stream/1, %% Socket operations setopts/2, peername/1, peercert/1, sockname/1, %% Low-level socket operations (for hackney_request/hackney_response) send/2, recv/2, recv/3, close/1, %% Response info response_headers/1, get_location/1, set_location/2, %% Pool management release_to_pool/1, verify_socket/1, is_ready/1, %% SSL upgrade upgrade_to_ssl/2, is_upgraded_ssl/1, %% Reuse control is_no_reuse/1, %% Owner management set_owner/2, set_owner_async/2, %% Protocol info get_protocol/1 ]). %% gen_statem callbacks -export([ init/1, callback_mode/0, terminate/3, code_change/4 ]). %% State functions -export([ idle/3, connecting/3, connected/3, sending/3, streaming_body/3, receiving/3, streaming/3, streaming_once/3, closed/3 ]). -ifdef(TEST). -export([h3_tls_opts/2]). -endif. -include("hackney.hrl"). -include("hackney_lib.hrl"). -define(CONNECT_TIMEOUT, 8000). -define(IDLE_TIMEOUT, infinity). %% Grace window for pooled hackney_conn in `closed` state, during which %% late-arriving calls race the pool DOWN cleanup and still get a proper %% error reply instead of exit:{normal, _}. See issue #836. -define(CLOSED_GRACE_MS, 50). %% State data record -record(conn_data, { %% Connection owner owner :: pid(), owner_mon :: reference() | undefined, %% Connection identity host :: string(), port :: inet:port_number(), netloc :: binary() | undefined, transport :: module(), %% Socket state socket :: inet:socket() | ssl:sslsocket() | undefined, buffer = <<>> :: binary(), %% Options connect_timeout = ?CONNECT_TIMEOUT :: timeout(), recv_timeout = ?RECV_TIMEOUT :: timeout(), idle_timeout = ?IDLE_TIMEOUT :: timeout(), connect_options = [] :: list(), ssl_options = [] :: list(), inform_fun :: fun((integer(), binary(), list()) -> any()) | undefined, auto_decompress = false :: boolean(), %% Pool integration pool_pid :: pid() | undefined, %% If set, connection is from a pool %% Request tracking request_from :: {pid(), reference()} | undefined, method :: binary() | undefined, path :: binary() | undefined, %% Response state version :: {integer(), integer()} | undefined, status :: integer() | undefined, reason :: binary() | undefined, response_headers :: term() | undefined, location :: binary() | undefined, %% Stores the final URL after redirects %% Parser state parser :: #hparser{} | undefined, %% Async mode async = false :: false | true | once | paused, stream_to :: pid() | undefined, async_ref :: pid() | undefined, %% Connection PID used as message correlation ref follow_redirect = false :: boolean(), %% SSL upgrade tracking - set when TCP connection is upgraded to SSL %% Upgraded connections should NOT be returned to pool upgraded_ssl = false :: boolean(), %% No-reuse flag - set for connections that should never be pooled %% (e.g., SOCKS5 proxy connections which establish per-request tunnels) no_reuse = false :: boolean(), %% HTTP/2 support %% Protocol negotiated via ALPN (http1, http2, or http3) protocol = http1 :: http1 | http2 | http3, %% HTTP/2 connection pid (from h2 library) h2_conn :: pid() | undefined, %% Monitor ref for the h2_connection process so hackney_conn fails fast %% if the h2 lib crashes. h2_mon :: reference() | undefined, %% Map of active HTTP/2 streams: StreamId => {From, StreamState} %% StreamState: %% {sync, waiting_headers} %% {sync, body, Status, Headers, AccBody} %% {async, AsyncMode, StreamTo, Ref, waiting_headers} %% {async, AsyncMode, StreamTo, Ref, streaming, Status, Headers} h2_streams = #{} :: #{pos_integer() => {term(), tuple()}}, %% HTTP/3 support (QUIC) %% HTTP/3 connection reference from hackney_h3 h3_conn :: reference() | undefined, %% Map of active HTTP/3 streams: StreamId => {From, StreamState} h3_streams = #{} :: #{non_neg_integer() => {gen_statem:from() | pid(), atom() | tuple()}}, %% Current HTTP/3 stream ID for streaming body mode h3_stream_id :: non_neg_integer() | undefined, %% Whether to try HTTP/3 (requires UDP) try_http3 = false :: boolean(), %% Pool name + handler module, so the connection can cache the H3 %% 0-RTT/resumption session ticket it receives. pool_name = default :: term(), pool_handler :: module() | undefined, %% Last H3 session ticket delivered by hackney_h3 (opaque term). h3_session_ticket :: term() | undefined }). %%==================================================================== %% API %%==================================================================== %% @doc Start a connection process. %% Options: %% - host: Target host (string) %% - port: Target port (integer) %% - transport: hackney_tcp or hackney_ssl %% - connect_timeout: Connection timeout (default 8000ms) %% - recv_timeout: Receive timeout (default 5000ms) %% - idle_timeout: Idle timeout before closing (default infinity) %% - connect_options: Options passed to transport connect %% - ssl_options: Additional SSL options -spec start_link(map()) -> {ok, pid()} | {error, term()}. start_link(Opts) when is_map(Opts) -> gen_statem:start_link(?MODULE, [self(), Opts], []). %% @doc Stop the connection process. %% Returns ok even if the process has already terminated. -spec stop(pid()) -> ok. stop(Pid) -> try gen_statem:stop(Pid) catch exit:noproc -> ok; exit:{noproc, _} -> ok; exit:normal -> ok; exit:{normal, _} -> ok end. %% @doc Connect to the target host. Blocks until connected or timeout. -spec connect(pid()) -> ok | {error, term()}. connect(Pid) -> connect(Pid, ?CONNECT_TIMEOUT). -spec connect(pid(), timeout()) -> ok | {error, term()}. connect(Pid, Timeout) -> gen_statem:call(Pid, connect, Timeout). %% @doc Get current state name for debugging. -spec get_state(pid()) -> {ok, atom()} | {error, term()}. get_state(Pid) -> gen_statem:call(Pid, get_state). %% @doc Send an HTTP request and wait for the response status and headers. %% Returns {ok, Status, Headers} for HTTP/1.1 or {ok, Status, Headers, Body} for HTTP/2. %% For HTTP/1.1, use body/1 or stream_body/1 to get the response body. -spec request(pid(), binary(), binary(), list(), binary() | iolist()) -> {ok, integer(), list()} | {ok, integer(), list(), binary()} | {error, term()}. request(Pid, Method, Path, Headers, Body) -> request(Pid, Method, Path, Headers, Body, infinity, []). -spec request(pid(), binary(), binary(), list(), binary() | iolist(), timeout()) -> {ok, integer(), list()} | {ok, integer(), list(), binary()} | {error, term()}. request(Pid, Method, Path, Headers, Body, Timeout) -> request(Pid, Method, Path, Headers, Body, Timeout, []). %% @doc Make an HTTP request with additional request options. %% Options: %% - inform_fun: fun(Status, Reason, Headers) - callback for 1xx responses -spec request(pid(), binary(), binary(), list(), binary() | iolist(), timeout(), list()) -> {ok, integer(), list()} | {ok, integer(), list(), binary()} | {error, term()}. request(Pid, Method, Path, Headers, Body, Timeout, ReqOpts) -> case valid_request_target(Path) of ok -> safe_call(Pid, {request, Method, Path, Headers, Body, ReqOpts}, Timeout); Err -> Err end. %% @private gen_statem:call that converts a callee which has already stopped, %% or stops while the call is in flight, into `{error, closed}' instead of %% letting `exit:{normal, _}' / `exit:noproc' reach the caller. A pooled %% connection can stop between checkout and the call (issue #861); the brief %% linger in the `closed' state narrows the window but cannot close it. Other %% exits (e.g. timeout) propagate unchanged. safe_call(Pid, Msg) -> safe_call(Pid, Msg, infinity). safe_call(Pid, Msg, Timeout) -> try gen_statem:call(Pid, Msg, Timeout) catch exit:noproc -> {error, closed}; exit:{noproc, _} -> {error, closed}; exit:normal -> {error, closed}; exit:{normal, _} -> {error, closed}; exit:shutdown -> {error, closed}; exit:{shutdown, _} -> {error, closed} end. %% @private GHSA-j9wq: the request target (path + query) is written verbatim %% into the HTTP/1.1 request line and the HTTP/2 / HTTP/3 :path pseudo-header. %% Raw CR, LF or NUL bytes let a caller-controlled URL inject extra header %% lines or split the request. RFC 3986 requires those bytes to be %% percent-encoded; reject them rather than emit a malformed request. valid_request_target(Path) when is_binary(Path) -> case binary:match(Path, [<<"\r">>, <<"\n">>, <<0>>]) of nomatch -> ok; _ -> {error, {invalid_request_target, Path}} end; valid_request_target(Path) when is_list(Path) -> valid_request_target(iolist_to_binary(Path)); valid_request_target(_) -> ok. %% @doc Send an HTTP/3 request and return headers immediately. %% Returns {ok, Status, Headers} and allows subsequent stream_body/1 calls. %% This is for pull-based body streaming over HTTP/3. -spec request_streaming(pid(), binary(), binary(), list(), binary() | iolist()) -> {ok, integer(), list()} | {error, term()}. request_streaming(Pid, Method, Path, Headers, Body) -> case valid_request_target(Path) of ok -> safe_call(Pid, {request_streaming, Method, Path, Headers, Body}, infinity); Err -> Err end. %% @doc Send only the request headers (for streaming body mode). %% After this, use send_body_chunk/2 and finish_send_body/1 to send the body, %% then start_response/1 to receive the response. -spec send_request_headers(pid(), binary(), binary(), list()) -> ok | {error, term()}. send_request_headers(Pid, Method, Path, Headers) -> case valid_request_target(Path) of ok -> safe_call(Pid, {send_headers, Method, Path, Headers}, infinity); Err -> Err end. %% @doc Send a chunk of the request body. -spec send_body_chunk(pid(), iodata()) -> ok | {error, term()}. send_body_chunk(Pid, Data) -> safe_call(Pid, {send_body_chunk, Data}, infinity). %% @doc Finish sending the request body. -spec finish_send_body(pid()) -> ok | {error, term()}. finish_send_body(Pid) -> safe_call(Pid, finish_send_body, infinity). %% @doc Start receiving the response after sending the full body. -spec start_response(pid()) -> {ok, integer(), list(), pid()} | {error, term()}. start_response(Pid) -> safe_call(Pid, start_response, infinity). %% @doc Get the full response body. -spec body(pid()) -> {ok, binary()} | {error, term()}. body(Pid) -> body(Pid, infinity). -spec body(pid(), timeout()) -> {ok, binary()} | {error, term()}. body(Pid, Timeout) -> safe_call(Pid, body, Timeout). %% @doc Stream the response body in chunks. %% Returns {ok, Data} for each chunk, {done, Pid} when complete. -spec stream_body(pid()) -> {ok, binary()} | done | {error, term()}. stream_body(Pid) -> safe_call(Pid, stream_body). %% @doc Send an HTTP request asynchronously. %% Returns {ok, Ref} immediately. Response is sent as messages: %% - {hackney_response, Ref, {status, Status, Reason}} %% - {hackney_response, Ref, {headers, Headers}} %% - {hackney_response, Ref, Data} (body chunks) %% - {hackney_response, Ref, done} %% - {hackney_response, Ref, {error, Reason}} %% When follow_redirect is true and response is a redirect: %% - {hackney_response, Ref, {redirect, Location, Headers}} for 301,302,307,308 %% - {hackney_response, Ref, {see_other, Location, Headers}} for 303 with POST %% AsyncMode: true (continuous) or once (one message at a time, use stream_next/1) -spec request_async(pid(), binary(), binary(), list(), binary() | iolist(), true | once) -> {ok, pid()} | {error, term()}. request_async(Pid, Method, Path, Headers, Body, AsyncMode) -> request_async(Pid, Method, Path, Headers, Body, AsyncMode, self(), false). -spec request_async(pid(), binary(), binary(), list(), binary() | iolist(), true | once, pid()) -> {ok, pid()} | {error, term()}. request_async(Pid, Method, Path, Headers, Body, AsyncMode, StreamTo) -> request_async(Pid, Method, Path, Headers, Body, AsyncMode, StreamTo, false). -spec request_async(pid(), binary(), binary(), list(), binary() | iolist(), true | once, pid(), boolean()) -> {ok, pid()} | {error, term()}. request_async(Pid, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect) -> case valid_request_target(Path) of ok -> safe_call(Pid, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect}); Err -> Err end. -spec request_async(pid(), binary(), binary(), list(), binary() | iolist(), true | once, pid(), boolean(), list()) -> {ok, pid()} | {error, term()}. request_async(Pid, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, ReqOpts) -> case valid_request_target(Path) of ok -> safe_call(Pid, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, ReqOpts}); Err -> Err end. %% @doc Request the next message in {async, once} mode. -spec stream_next(pid()) -> ok | {error, term()}. stream_next(Pid) -> gen_statem:cast(Pid, stream_next). %% @doc Stop async mode and return to sync mode. -spec stop_async(pid()) -> ok | {error, term()}. stop_async(Pid) -> gen_statem:call(Pid, stop_async). %% @doc Pause async streaming (hibernate the stream). -spec pause_stream(pid()) -> ok | {error, term()}. pause_stream(Pid) -> gen_statem:cast(Pid, pause_stream). %% @doc Resume paused async streaming. -spec resume_stream(pid()) -> ok | {error, term()}. resume_stream(Pid) -> gen_statem:cast(Pid, resume_stream). %% @doc Set socket options on the underlying socket. -spec setopts(pid(), list()) -> ok | {error, term()}. setopts(Pid, Opts) -> gen_statem:call(Pid, {setopts, Opts}). %% @doc Get the remote address and port. -spec peername(pid()) -> {ok, {inet:ip_address(), inet:port_number()}} | {error, term()}. peername(Pid) -> gen_statem:call(Pid, peername). %% @doc Get the peer SSL certificate. %% Returns {ok, Cert} where Cert is the DER-encoded certificate binary, %% or {error, Reason} if the connection is not SSL or the certificate is unavailable. -spec peercert(pid()) -> {ok, binary()} | {error, term()}. peercert(Pid) -> gen_statem:call(Pid, peercert). %% @doc Get the local address and port. -spec sockname(pid()) -> {ok, {inet:ip_address(), inet:port_number()}} | {error, term()}. sockname(Pid) -> gen_statem:call(Pid, sockname). %% @doc Get the last response headers. -spec response_headers(pid()) -> list() | undefined. response_headers(Pid) -> gen_statem:call(Pid, response_headers). %% @doc Get the stored location (final URL after redirects). -spec get_location(pid()) -> binary() | undefined. get_location(Pid) -> gen_statem:call(Pid, get_location). %% @doc Set the location (used after following redirects). -spec set_location(pid(), binary()) -> ok. set_location(Pid, Location) -> gen_statem:call(Pid, {set_location, Location}). %% @doc Send data through the connection process. %% This is a low-level function used by hackney_request. -spec send(pid(), iodata()) -> ok | {error, term()}. send(Pid, Data) -> gen_statem:call(Pid, {send, Data}). %% @doc Receive data from the connection process. %% This is a low-level function used by hackney_response. -spec recv(pid(), timeout()) -> {ok, binary()} | {error, term()}. recv(Pid, Timeout) -> recv(Pid, 0, Timeout). -spec recv(pid(), non_neg_integer(), timeout()) -> {ok, binary()} | {error, term()}. recv(Pid, Length, Timeout) -> gen_statem:call(Pid, {recv, Length, Timeout}, Timeout + 5000). %% @doc Close the connection. %% This is a low-level function that closes the socket but keeps the process. -spec close(pid()) -> ok. close(Pid) -> gen_statem:call(Pid, close_socket). %% @doc Release the connection back to the pool. %% This notifies the pool that the connection is available for reuse. %% Uses a synchronous call to ensure the pool has processed the checkin. -spec release_to_pool(pid()) -> ok. release_to_pool(Pid) -> gen_statem:call(Pid, release_to_pool, 5000). %% @doc Set a new owner for this connection (sync). %% This updates the process being monitored - if the new owner crashes, %% the connection will terminate. Used by the pool when checking out %% a connection to a new requester. -spec set_owner(pid(), pid()) -> ok | {error, invalid_state}. set_owner(Pid, NewOwner) -> gen_statem:call(Pid, {set_owner, NewOwner}, 5000). %% @doc Set a new owner for this connection (async). %% Same as set_owner/2 but non-blocking. Used when the caller cannot %% block (e.g., during pool checkin to avoid deadlock). -spec set_owner_async(pid(), pid()) -> ok. set_owner_async(Pid, NewOwner) -> gen_statem:cast(Pid, {set_owner, NewOwner}). %% @doc Check if the connection's socket is still healthy. %% Returns ok if socket is open, {error, closed} otherwise. -spec verify_socket(pid()) -> ok | {error, closed | term()}. verify_socket(Pid) -> gen_statem:call(Pid, verify_socket). %% @doc Check if the connection is ready for a new request. %% Returns {ok, connected} if ready, or error/closed status. %% This combines state check and socket verification in one call. -spec is_ready(pid()) -> {ok, connected} | {ok, closed} | {error, term()}. is_ready(Pid) -> gen_statem:call(Pid, is_ready). %% @doc Upgrade a TCP connection to SSL. %% This performs an SSL handshake on the existing TCP socket. %% After upgrade, the connection is marked as upgraded_ssl and should %% NOT be returned to the pool (SSL connections are not pooled for security). -spec upgrade_to_ssl(pid(), list()) -> ok | {error, term()}. upgrade_to_ssl(Pid, SslOpts) -> gen_statem:call(Pid, {upgrade_to_ssl, SslOpts}, infinity). %% @doc Check if this connection was upgraded from TCP to SSL. %% Upgraded connections should be closed after use, not returned to pool. -spec is_upgraded_ssl(pid()) -> boolean(). is_upgraded_ssl(Pid) -> gen_statem:call(Pid, is_upgraded_ssl). %% @doc Check if this connection should not be reused/pooled. %% SOCKS5 proxy connections set this flag since each establishes a unique tunnel. -spec is_no_reuse(pid()) -> boolean(). is_no_reuse(Pid) -> gen_statem:call(Pid, is_no_reuse). %% @doc Get the negotiated protocol for this connection. %% Returns http1, http2, or http3 based on ALPN negotiation (SSL connections), %% QUIC (HTTP/3), or http1 for plain TCP connections. -spec get_protocol(pid()) -> http1 | http2 | http3. get_protocol(Pid) -> gen_statem:call(Pid, get_protocol). %%==================================================================== %% gen_statem callbacks %%==================================================================== callback_mode() -> [state_functions, state_enter]. init([DefaultOwner, Opts]) -> process_flag(trap_exit, true), %% Use owner from opts if provided (e.g., from pool), otherwise use default Owner = maps:get(owner, Opts, DefaultOwner), OwnerMon = monitor(process, Owner), Host = maps:get(host, Opts), Port = maps:get(port, Opts, 80), Transport = maps:get(transport, Opts, hackney_tcp), %% Compute netloc for Host header Netloc = compute_netloc(Host, Port, Transport), %% Check if a pre-established socket is provided (e.g., from proxy) Socket = maps:get(socket, Opts, undefined), Data = #conn_data{ owner = Owner, owner_mon = OwnerMon, host = Host, port = Port, netloc = Netloc, transport = Transport, socket = Socket, connect_timeout = maps:get(connect_timeout, Opts, ?CONNECT_TIMEOUT), recv_timeout = maps:get(recv_timeout, Opts, ?RECV_TIMEOUT), idle_timeout = maps:get(idle_timeout, Opts, ?IDLE_TIMEOUT), connect_options = maps:get(connect_options, Opts, []), ssl_options = maps:get(ssl_options, Opts, []), pool_pid = maps:get(pool_pid, Opts, undefined), pool_name = maps:get(pool_name, Opts, default), pool_handler = maps:get(pool_handler, Opts, undefined), no_reuse = maps:get(no_reuse, Opts, false), inform_fun = maps:get(inform_fun, Opts, undefined), auto_decompress = maps:get(auto_decompress, Opts, false) }, %% If socket is provided, start in connected state; otherwise start in idle case Socket of undefined -> {ok, idle, Data}; _ -> {ok, connected, Data} end. terminate(_Reason, _State, #conn_data{socket = Socket, transport = Transport, h2_conn = H2Conn, h2_mon = H2Mon, h3_conn = H3Conn}) -> %% Close HTTP/2 connection (owns its socket) case H2Conn of undefined -> ok; _ -> case H2Mon of undefined -> ok; _ -> erlang:demonitor(H2Mon, [flush]) end, close_h2(H2Conn) end, %% Close HTTP/3 connection if present case H3Conn of undefined -> ok; _ -> hackney_h3:close(H3Conn) end, %% Close socket (only still owned by us for HTTP/1.1; h2 lib closes its own) case {Socket, H2Conn} of {undefined, _} -> ok; {_, undefined} -> Transport:close(Socket); _ -> ok end, ok. code_change(_OldVsn, State, Data, _Extra) -> {ok, State, Data}. %%==================================================================== %% State: idle - Not connected yet %%==================================================================== idle(enter, _OldState, _Data) -> keep_state_and_data; idle({call, From}, connect, Data) -> %% Perform connection synchronously #conn_data{ host = Host, port = Port, transport = Transport, connect_timeout = Timeout, connect_options = ConnectOpts, ssl_options = SslOpts, try_http3 = TryHttp3 } = Data, %% Check if we should try HTTP/3 first (SSL transport + http3 in protocols) Protocols = proplists:get_value(protocols, ConnectOpts, hackney_util:default_protocols()), ShouldTryHttp3 = TryHttp3 orelse (Transport =:= hackney_ssl andalso lists:member(http3, Protocols)), case ShouldTryHttp3 of true -> %% Try HTTP/3 first case try_h3_connect(Host, Port, Timeout, ConnectOpts, SslOpts) of {ok, H3Conn} -> NewData = Data#conn_data{ h3_conn = H3Conn, protocol = http3 }, {next_state, connected, NewData, [{reply, From, ok}]}; {error, _H3Reason} -> %% HTTP/3 failed, fall back to TCP/TLS do_tcp_connect(From, Data) end; false -> %% Standard TCP/TLS connection do_tcp_connect(From, Data) end; idle({call, From}, get_state, _Data) -> {keep_state_and_data, [{reply, From, {ok, idle}}]}; idle(info, {'DOWN', Ref, process, _Pid, _Reason}, #conn_data{owner_mon = Ref} = Data) -> %% Owner died {stop, normal, Data}; idle(EventType, Event, Data) -> handle_common(EventType, Event, idle, Data). %%==================================================================== %% State: connecting - TCP/SSL handshake in progress %%==================================================================== connecting(enter, _OldState, _Data) -> keep_state_and_data; connecting(state_timeout, connect_timeout, Data) -> reply_and_stop(Data#conn_data.request_from, {error, connect_timeout}, Data); connecting({call, From}, get_state, _Data) -> {keep_state_and_data, [{reply, From, {ok, connecting}}]}; connecting(info, {'DOWN', Ref, process, _Pid, _Reason}, #conn_data{owner_mon = Ref} = Data) -> {stop, normal, Data}; connecting(EventType, Event, Data) -> handle_common(EventType, Event, connecting, Data). %%==================================================================== %% State: connected - Ready for requests %%==================================================================== connected(enter, OldState, #conn_data{transport = Transport, socket = Socket, idle_timeout = Timeout, pool_pid = PoolPid} = Data) -> %% Set socket to active mode to receive server close notifications (tcp_closed/ssl_closed) %% This fixes issue #544: stale connections not being detected when server closes idle connections %% Only enable active mode when returning from a completed request cycle (receiving, streaming states) %% NOT on initial connection from 'connecting' state - server might send data before we send request _ = case {Socket, should_enable_active_mode(OldState)} of {undefined, _} -> ok; {_, false} -> ok; {_, true} -> Transport:setopts(Socket, [{active, once}]) end, %% Auto-release to pool when body reading is complete %% This happens when transitioning from receiving state (body fully read) Data2 = case {PoolPid, should_auto_release(OldState)} of {undefined, _} -> Data; {_, false} -> Data; {_, true} -> %% Transfer ownership back to pool and notify it auto_release_to_pool(Data) end, case Timeout of infinity -> {keep_state, Data2}; _ -> {keep_state, Data2, [{state_timeout, Timeout, idle_timeout}]} end; connected({call, From}, release_to_pool, #conn_data{pool_pid = PoolPid, owner_mon = OldMon, transport = Transport, socket = Socket} = Data) -> %% Reset owner to pool before notifying, to avoid deadlock %% (pool might call set_owner back, but connection is blocked here) Data2 = case PoolPid of undefined -> Data; _ -> demonitor(OldMon, [flush]), NewMon = monitor(process, PoolPid), Data#conn_data{owner = PoolPid, owner_mon = NewMon} end, %% Enable active mode for close detection now that connection is idle in pool %% This fixes issue #544: detect server-initiated closes while idle _ = case Socket of undefined -> ok; _ -> Transport:setopts(Socket, [{active, once}]) end, %% Notify pool that connection is available for reuse (sync) notify_pool_available_sync(Data2), {keep_state, Data2, [{reply, From, ok}]}; connected({call, From}, {set_owner, NewOwner}, #conn_data{owner_mon = OldMon} = Data) -> %% Update owner - demonitor old, monitor new demonitor(OldMon, [flush]), NewMon = monitor(process, NewOwner), {keep_state, Data#conn_data{owner = NewOwner, owner_mon = NewMon}, [{reply, From, ok}]}; connected(cast, {set_owner, NewOwner}, #conn_data{owner_mon = OldMon, pool_pid = PoolPid, transport = Transport, socket = Socket} = Data) -> %% Async owner update - used by pool during checkin to avoid deadlock demonitor(OldMon, [flush]), NewMon = monitor(process, NewOwner), %% If new owner is the pool, connection is becoming idle - enable active mode for close detection %% This fixes issue #544: detect server-initiated closes while idle in pool _ = case {NewOwner, PoolPid, Socket} of {PoolPid, PoolPid, Socket} when PoolPid =/= undefined, Socket =/= undefined -> Transport:setopts(Socket, [{active, once}]); _ -> ok end, {keep_state, Data#conn_data{owner = NewOwner, owner_mon = NewMon}}; connected(state_timeout, idle_timeout, Data) -> %% Idle timeout - close connection {next_state, closed, Data}; connected({call, From}, get_state, _Data) -> {keep_state_and_data, [{reply, From, {ok, connected}}]}; connected({call, From}, verify_socket, #conn_data{socket = undefined} = Data) -> %% Socket not connected {next_state, closed, Data, [{reply, From, {error, closed}}]}; connected({call, From}, verify_socket, #conn_data{transport = Transport, socket = Socket} = Data) -> %% Check if socket has any pending data or close messages case check_socket_health(Transport, Socket) of ok -> {keep_state_and_data, [{reply, From, ok}]}; {error, Reason} -> {next_state, closed, Data#conn_data{socket = undefined}, [{reply, From, {error, Reason}}]} end; connected({call, From}, is_ready, #conn_data{socket = undefined} = Data) -> %% Socket not connected {next_state, closed, Data, [{reply, From, {ok, closed}}]}; connected({call, From}, is_ready, #conn_data{transport = Transport, socket = Socket} = Data) -> %% Check for pending close message first (from active mode) case has_pending_close(Socket) of true -> %% Server closed the connection - transition to closed state {next_state, closed, Data#conn_data{socket = undefined}, [{reply, From, {ok, closed}}]}; false -> %% No close message pending - check socket health case check_socket_health(Transport, Socket) of ok -> %% Socket is healthy - set to passive mode for checkout %% This ensures the socket is ready for blocking recv operations _ = Transport:setopts(Socket, [{active, false}]), %% Flush any data messages that arrived while in active mode flush_socket_messages(Socket), {keep_state_and_data, [{reply, From, {ok, connected}}]}; {error, _} -> {keep_state_and_data, [{reply, From, {ok, closed}}]} end end; connected({call, From}, {upgrade_to_ssl, _SslOpts}, #conn_data{transport = hackney_ssl} = _Data) -> %% Already SSL - no upgrade needed {keep_state_and_data, [{reply, From, ok}]}; connected({call, From}, {upgrade_to_ssl, SslOpts}, #conn_data{socket = Socket, host = Host, connect_options = ConnectOpts} = Data) -> %% Upgrade TCP socket to SSL (e.g., after CONNECT proxy tunnel) %% Use ssl_opts/2 to properly merge defaults with user options %% (handles cacertfile vs cacerts correctly) MergedSslOpts = hackney_ssl:ssl_opts(Host, [{ssl_options, SslOpts}]), %% Add ALPN options for HTTP/2 negotiation %% Check both SslOpts (from upgrade call) and ConnectOpts (from initial config) AlpnOpts = case hackney_ssl:alpn_opts(SslOpts) of [] -> hackney_ssl:alpn_opts(ConnectOpts); Opts -> Opts end, FinalSslOpts = hackney_util:merge_opts(MergedSslOpts, AlpnOpts), case ssl:connect(Socket, FinalSslOpts) of {ok, SslSocket} -> %% Detect negotiated protocol Protocol = hackney_ssl:get_negotiated_protocol(SslSocket), %% Update connection to use SSL NewData = Data#conn_data{ transport = hackney_ssl, socket = SslSocket, upgraded_ssl = true, % Mark as upgraded - should NOT return to pool protocol = Protocol }, %% Initialize HTTP/2 if negotiated case Protocol of http2 -> init_h2_after_upgrade(SslSocket, NewData, From); http1 -> {keep_state, NewData, [{reply, From, ok}]} end; {error, Reason} -> {keep_state_and_data, [{reply, From, {error, Reason}}]} end; connected({call, From}, is_upgraded_ssl, #conn_data{upgraded_ssl = Upgraded}) -> {keep_state_and_data, [{reply, From, Upgraded}]}; connected({call, From}, is_no_reuse, #conn_data{no_reuse = NoReuse}) -> {keep_state_and_data, [{reply, From, NoReuse}]}; connected({call, From}, {request, Method, Path, Headers, Body, ReqOpts}, #conn_data{protocol = http2} = Data) -> %% HTTP/2 request - use h2_machine (1xx not applicable for HTTP/2) %% Allow recv_timeout to be overridden per-request (fix for issue #832) RecvTimeout = proplists:get_value(recv_timeout, ReqOpts, Data#conn_data.recv_timeout), NewData = Data#conn_data{recv_timeout = RecvTimeout}, do_h2_request(From, Method, Path, Headers, Body, NewData); connected({call, From}, {request, Method, Path, Headers, Body, ReqOpts}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 request - use hackney_h3 (1xx not applicable for HTTP/3) %% Allow recv_timeout to be overridden per-request (fix for issue #832) RecvTimeout = proplists:get_value(recv_timeout, ReqOpts, Data#conn_data.recv_timeout), NewData = Data#conn_data{recv_timeout = RecvTimeout}, do_h3_request(From, Method, Path, Headers, Body, NewData); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo}, #conn_data{protocol = http2} = Data) -> %% HTTP/2 async request do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, false, Data); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect}, #conn_data{protocol = http2} = Data) -> %% HTTP/2 async request with redirect option do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, Data); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 async request do_h3_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, Data); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, _FollowRedirect}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 async request (redirect not yet implemented for H3) do_h3_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, Data); connected({call, From}, {request_streaming, Method, Path, Headers, Body}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 request with streaming body reads (returns headers, then stream_body for chunks) do_h3_request_streaming(From, Method, Path, Headers, Body, Data); connected({call, From}, {request, Method, Path, Headers, Body, ReqOpts}, Data) -> %% HTTP/1.1 request InformFun = proplists:get_value(inform_fun, ReqOpts, undefined), AutoDecompress = proplists:get_value(auto_decompress, ReqOpts, false), %% Allow recv_timeout to be overridden per-request (fix for issue #832) RecvTimeout = proplists:get_value(recv_timeout, ReqOpts, Data#conn_data.recv_timeout), NewData = Data#conn_data{ request_from = From, method = Method, path = Path, parser = undefined, version = undefined, status = undefined, reason = undefined, response_headers = undefined, buffer = <<>>, async = false, async_ref = undefined, stream_to = undefined, inform_fun = InformFun, auto_decompress = AutoDecompress, recv_timeout = RecvTimeout }, {next_state, sending, NewData, [{next_event, internal, {send_request, Method, Path, Headers, Body}}]}; connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo}, Data) -> %% Start a new async request (no redirect following) do_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, false, Data); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect}, Data) -> %% Start a new async request with redirect option (HTTP/1.1) do_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, Data); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, ReqOpts}, #conn_data{protocol = http2} = Data) -> %% HTTP/2 async request with ReqOpts (fix for issue #832) RecvTimeout = proplists:get_value(recv_timeout, ReqOpts, Data#conn_data.recv_timeout), NewData = Data#conn_data{recv_timeout = RecvTimeout}, do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, NewData); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, _FollowRedirect, ReqOpts}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 async request with ReqOpts (fix for issue #832, redirect not yet implemented for H3) RecvTimeout = proplists:get_value(recv_timeout, ReqOpts, Data#conn_data.recv_timeout), NewData = Data#conn_data{recv_timeout = RecvTimeout}, do_h3_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, NewData); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, ReqOpts}, Data) -> %% HTTP/1.1 async request with ReqOpts (fix for issue #832) RecvTimeout = proplists:get_value(recv_timeout, ReqOpts, Data#conn_data.recv_timeout), NewData = Data#conn_data{recv_timeout = RecvTimeout}, do_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, NewData); connected({call, From}, {send_headers, Method, Path, Headers}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 streaming body - send headers only via QUIC do_h3_send_headers(From, Method, Path, Headers, Data); connected({call, From}, {send_headers, Method, Path, Headers}, Data) -> %% Send only headers for streaming body mode (HTTP/1.1) NewData = Data#conn_data{ request_from = From, method = Method, path = Path, parser = undefined, version = undefined, status = undefined, reason = undefined, response_headers = undefined, buffer = <<>>, async = false, async_ref = undefined, stream_to = undefined }, %% Transition to streaming_body state {next_state, streaming_body, NewData, [{next_event, internal, {send_headers_only, Method, Path, Headers}}]}; %% HTTP/2 owner messages from h2 library connected(info, {h2, H2Conn, Event}, #conn_data{h2_conn = H2Conn} = Data) -> handle_h2_event(Event, Data); %% h2_connection is linked via start_link; trap_exit surfaces its termination %% as an 'EXIT' signal. Convert to the same cleanup path as the monitor DOWN. connected(info, {'EXIT', H2Conn, Reason}, #conn_data{h2_conn = H2Conn} = Data) -> h2_on_closed(Reason, Data#conn_data{h2_conn = undefined, h2_mon = undefined}); connected(info, {'DOWN', Mon, process, _Pid, Reason}, #conn_data{h2_mon = Mon} = Data) -> h2_on_closed(Reason, Data#conn_data{h2_conn = undefined, h2_mon = undefined}); connected(info, {tcp_closed, Socket}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; connected(info, {ssl_closed, Socket}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; connected(info, {tcp_error, Socket, _Reason}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; connected(info, {ssl_error, Socket, _Reason}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; %% Unexpected data received while idle - HTTP/1.1 only (H/2 socket is owned %% by h2_connection; H/3 uses QUIC messages). connected(info, {tcp, Socket, _UnexpectedData}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; connected(info, {ssl, Socket, _UnexpectedData}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; %% HTTP/3 message handling connected(info, {h3, ConnRef, {stream_headers, StreamId, Headers, Fin}}, #conn_data{h3_conn = ConnRef, h3_streams = Streams} = Data) -> handle_h3_headers(StreamId, Headers, Fin, Streams, Data); connected(info, {h3, ConnRef, {stream_data, StreamId, RecvData, Fin}}, #conn_data{h3_conn = ConnRef, h3_streams = Streams} = Data) -> handle_h3_data(StreamId, RecvData, Fin, Streams, Data); connected(info, {h3, ConnRef, {stream_reset, StreamId, ErrorCode}}, #conn_data{h3_conn = ConnRef, h3_streams = Streams} = Data) -> handle_h3_stream_reset(StreamId, ErrorCode, Streams, Data); connected(info, {h3, ConnRef, {closed, Reason}}, #conn_data{h3_conn = ConnRef} = Data) -> %% HTTP/3 connection closed handle_h3_conn_closed(Reason, Data); connected(info, {h3, ConnRef, {transport_error, Code, Msg}}, #conn_data{h3_conn = ConnRef} = Data) -> %% QUIC transport error handle_h3_error({transport_error, Code, Msg}, Data); %% QUIC socket ready - drive event loop connected(info, {select, _Resource, _Ref, ready_input}, #conn_data{h3_conn = ConnRef}) when ConnRef =/= undefined -> _ = hackney_h3:process(ConnRef), keep_state_and_data; connected(info, {'DOWN', Ref, process, _Pid, _Reason}, #conn_data{owner_mon = Ref} = Data) -> {stop, normal, Data}; %% HTTP/3 stream_body call - returns buffered chunk or waits for data connected({call, From}, stream_body, #conn_data{protocol = http3, h3_streams = Streams} = Data) -> handle_h3_stream_body(From, Streams, Data); connected(EventType, Event, Data) -> handle_common(EventType, Event, connected, Data). %%==================================================================== %% State: sending - Sending request data %%==================================================================== sending(enter, connected, #conn_data{transport = Transport, socket = Socket}) -> %% Set socket to passive mode for blocking send/recv operations %% (socket was in active mode while idle in connected state) %% Note: socket may be undefined for HTTP/3 (QUIC) connections _ = case Socket of undefined -> ok; _ -> Transport:setopts(Socket, [{active, false}]) end, keep_state_and_data; sending(internal, {send_request, Method, Path, Headers, Body}, Data) -> case do_send_request(Method, Path, Headers, Body, Data) of {ok, NewData} -> {next_state, receiving, NewData, [{next_event, internal, do_recv_response}]}; {error, Reason} -> {next_state, closed, Data, [{reply, Data#conn_data.request_from, {error, Reason}}]} end; sending(internal, {send_request_async, Method, Path, Headers, Body}, Data) -> case do_send_request(Method, Path, Headers, Body, Data) of {ok, NewData} -> %% Reply with ref immediately, then start async receiving From = NewData#conn_data.request_from, Ref = NewData#conn_data.async_ref, {next_state, receiving, NewData#conn_data{request_from = undefined}, [{reply, From, {ok, Ref}}, {next_event, internal, do_recv_response_async}]}; {error, Reason} -> From = Data#conn_data.request_from, {next_state, closed, Data, [{reply, From, {error, Reason}}]} end; sending({call, From}, get_state, _Data) -> {keep_state_and_data, [{reply, From, {ok, sending}}]}; sending(info, {tcp_closed, Socket}, #conn_data{socket = Socket, request_from = From} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}, [{reply, From, {error, closed}}]}; sending(info, {ssl_closed, Socket}, #conn_data{socket = Socket, request_from = From} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}, [{reply, From, {error, closed}}]}; sending(info, {'DOWN', Ref, process, _Pid, _Reason}, #conn_data{owner_mon = Ref} = Data) -> {stop, normal, Data}; sending(EventType, Event, Data) -> handle_common(EventType, Event, sending, Data). %%==================================================================== %% State: streaming_body - Streaming request body %%==================================================================== streaming_body(enter, connected, #conn_data{transport = Transport, socket = Socket}) -> %% Set socket to passive mode for blocking send/recv operations %% (socket was in active mode while idle in connected state) %% Note: socket may be undefined for HTTP/3 (QUIC) connections _ = case Socket of undefined -> ok; _ -> Transport:setopts(Socket, [{active, false}]) end, keep_state_and_data; streaming_body(internal, {send_headers_only, Method, Path, Headers}, Data) -> %% Send only headers, then return ok and wait for body chunks #conn_data{host = _Host, port = _Port, transport = Transport, socket = Socket} = Data, %% Build request line and headers (with Transfer-Encoding: chunked for streaming) HeadersObj = hackney_headers:from_list(Headers), %% Add Transfer-Encoding: chunked if not present HeadersWithTE = case hackney_headers:get_value(<<"transfer-encoding">>, HeadersObj) of undefined -> hackney_headers:store(<<"Transfer-Encoding">>, <<"chunked">>, HeadersObj); _ -> HeadersObj end, HeadersList = hackney_headers:to_list(HeadersWithTE), RequestLine = build_request_line(Method, Path), HeaderLines = [[Name, <<": ">>, Value, <<"\r\n">>] || {Name, Value} <- HeadersList], HeadersData = [RequestLine, HeaderLines, <<"\r\n">>], case Transport:send(Socket, HeadersData) of ok -> From = Data#conn_data.request_from, {keep_state, Data#conn_data{request_from = undefined}, [{reply, From, ok}]}; {error, Reason} -> From = Data#conn_data.request_from, {next_state, closed, Data, [{reply, From, {error, Reason}}]} end; streaming_body({call, From}, {send_body_chunk, BodyData}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 - send body chunk via QUIC #conn_data{h3_conn = ConnRef, h3_stream_id = StreamId} = Data, Result = case BodyData of Fun when is_function(Fun, 0) -> stream_body_fun_h3(ConnRef, StreamId, Fun); {Fun, State} when is_function(Fun, 1) -> stream_body_fun_h3(ConnRef, StreamId, {Fun, State}); _ -> hackney_h3:send_body_chunk(ConnRef, StreamId, iolist_to_binary(BodyData), false) end, case Result of ok -> {keep_state_and_data, [{reply, From, ok}]}; {error, Reason} -> {next_state, closed, Data, [{reply, From, {error, Reason}}]} end; streaming_body({call, From}, {send_body_chunk, BodyData}, Data) -> #conn_data{transport = Transport, socket = Socket} = Data, %% Send as chunked encoding (HTTP/1.1) Result = case BodyData of Fun when is_function(Fun, 0) -> stream_body_fun(Transport, Socket, Fun); {Fun, State} when is_function(Fun, 1) -> stream_body_fun(Transport, Socket, {Fun, State}); _ -> Transport:send(Socket, encode_chunk(BodyData)) end, case Result of ok -> {keep_state_and_data, [{reply, From, ok}]}; {error, Reason} -> {next_state, closed, Data, [{reply, From, {error, Reason}}]} end; streaming_body({call, From}, finish_send_body, #conn_data{protocol = http3} = Data) -> %% HTTP/3 - finish sending body (send empty data with Fin=true) #conn_data{h3_conn = ConnRef, h3_stream_id = StreamId, h3_streams = Streams} = Data, case hackney_h3:finish_send_body(ConnRef, StreamId, Streams) of {ok, _} -> {keep_state, Data#conn_data{request_from = From}, [{reply, From, ok}]}; {error, Reason} -> {next_state, closed, Data, [{reply, From, {error, Reason}}]} end; streaming_body({call, From}, finish_send_body, Data) -> #conn_data{transport = Transport, socket = Socket} = Data, %% Send final chunk marker (HTTP/1.1) case Transport:send(Socket, <<"0\r\n\r\n">>) of ok -> {keep_state, Data#conn_data{request_from = From}, [{reply, From, ok}]}; {error, Reason} -> {next_state, closed, Data, [{reply, From, {error, Reason}}]} end; streaming_body({call, From}, start_response, #conn_data{protocol = http3} = Data) -> %% HTTP/3 - wait for response headers via QUIC messages %% Transition to connected and wait for QUIC stream_headers message NewData = Data#conn_data{request_from = From}, {next_state, connected, NewData}; streaming_body({call, From}, start_response, Data) -> %% Transition to receiving state and get response (HTTP/1.1) %% Use {do_recv_response, include_pid} to include pid in response NewData = Data#conn_data{request_from = From}, {next_state, receiving, NewData, [{next_event, internal, {do_recv_response, include_pid}}]}; streaming_body({call, From}, get_state, _Data) -> {keep_state_and_data, [{reply, From, {ok, streaming_body}}]}; streaming_body(info, {tcp_closed, Socket}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; streaming_body(info, {ssl_closed, Socket}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; %% HTTP/3 message handling in streaming_body state streaming_body(info, {h3, ConnRef, {stream_headers, StreamId, Headers, Fin}}, #conn_data{h3_conn = ConnRef, h3_streams = Streams} = Data) -> %% Early response headers while still sending body handle_h3_headers(StreamId, Headers, Fin, Streams, Data); streaming_body(info, {h3, ConnRef, {stream_data, StreamId, RecvData, Fin}}, #conn_data{h3_conn = ConnRef, h3_streams = Streams} = Data) -> handle_h3_data(StreamId, RecvData, Fin, Streams, Data); streaming_body(info, {h3, ConnRef, {stream_reset, StreamId, ErrorCode}}, #conn_data{h3_conn = ConnRef, h3_streams = Streams} = Data) -> handle_h3_stream_reset(StreamId, ErrorCode, Streams, Data); streaming_body(info, {h3, ConnRef, {closed, Reason}}, #conn_data{h3_conn = ConnRef} = Data) -> handle_h3_conn_closed(Reason, Data); streaming_body(info, {h3, ConnRef, {transport_error, Code, Msg}}, #conn_data{h3_conn = ConnRef} = Data) -> handle_h3_error({transport_error, Code, Msg}, Data); %% QUIC socket ready - drive event loop streaming_body(info, {select, _Resource, _Ref, ready_input}, #conn_data{h3_conn = ConnRef}) when ConnRef =/= undefined -> _ = hackney_h3:process(ConnRef), keep_state_and_data; streaming_body(info, {'DOWN', Ref, process, _Pid, _Reason}, #conn_data{owner_mon = Ref} = Data) -> {stop, normal, Data}; streaming_body(EventType, Event, Data) -> handle_common(EventType, Event, streaming_body, Data). %%==================================================================== %% State: receiving - Awaiting/streaming response %%==================================================================== receiving(enter, sending, _Data) -> %% Just enter state, request handling happens in internal event keep_state_and_data; receiving(enter, streaming_body, _Data) -> %% Coming from streaming body state keep_state_and_data; receiving(enter, receiving, _Data) -> %% Re-entering after body streaming keep_state_and_data; receiving(internal, do_recv_response, Data) -> %% Receive and parse response status and headers (sync mode, no pid in response) do_recv_response_impl(Data, false); receiving(internal, {do_recv_response, include_pid}, Data) -> %% Receive and parse response status and headers (sync mode, include pid in response) do_recv_response_impl(Data, true); receiving(internal, do_recv_response_async, Data) -> %% Receive and parse response status and headers (async mode) Parser = hackney_http:parser([response]), DataWithParser = Data#conn_data{parser = Parser}, #conn_data{async_ref = Ref, stream_to = StreamTo, async = AsyncMode, follow_redirect = FollowRedirect, method = Method} = Data, case recv_status_and_headers(DataWithParser) of {ok, Status, Headers, NewData} -> HeadersList = hackney_headers:to_list(Headers), maybe_record_altsvc(HeadersList, NewData), %% Check if this is a redirect and we should handle it case maybe_handle_async_redirect(Status, Method, Headers, FollowRedirect) of {redirect, Location} -> %% Skip body and send redirect message _ = skip_response_body(NewData), StreamTo ! {hackney_response, Ref, {redirect, Location, HeadersList}}, notify_pool_available(NewData), {next_state, connected, NewData#conn_data{response_headers = Headers}}; {see_other, Location} -> %% Skip body and send see_other message _ = skip_response_body(NewData), StreamTo ! {hackney_response, Ref, {see_other, Location, HeadersList}}, notify_pool_available(NewData), {next_state, connected, NewData#conn_data{response_headers = Headers}}; no_redirect -> %% Normal response - send status and headers StreamTo ! {hackney_response, Ref, {status, Status, NewData#conn_data.reason}}, StreamTo ! {hackney_response, Ref, {headers, HeadersList}}, %% Check if response has no body (204, 304, or HEAD request) HasNoBody = (Status =:= 204) orelse (Status =:= 304) orelse (Method =:= <<"HEAD">>), case HasNoBody of true -> %% No body - send done immediately StreamTo ! {hackney_response, Ref, done}, notify_pool_available(NewData), {next_state, connected, NewData#conn_data{response_headers = Headers, async = false, async_ref = undefined}}; false -> %% Transition to appropriate streaming state based on mode NextState = case AsyncMode of true -> streaming; once -> streaming_once end, {next_state, NextState, NewData#conn_data{response_headers = Headers}} end end; {error, Reason} -> StreamTo ! {hackney_response, Ref, {error, Reason}}, {next_state, closed, Data} end; receiving({call, From}, body, Data) -> %% Read full body case read_full_body(Data, <<>>) of {ok, Body, #conn_data{socket = undefined} = NewData} -> %% Socket was closed during body read (e.g., no Content-Length) %% Transition to closed state instead of connected {next_state, closed, NewData, [{reply, From, {ok, Body}}]}; {ok, Body, NewData} -> %% Socket still valid - return to connected state {next_state, connected, NewData, [{reply, From, {ok, Body}}]}; {error, Reason} -> {next_state, closed, Data, [{reply, From, {error, Reason}}]} end; receiving({call, From}, stream_body, Data) -> %% Stream body chunk case stream_body_chunk(Data) of {ok, Chunk, NewData} -> {keep_state, NewData, [{reply, From, {ok, Chunk}}]}; {done, #conn_data{socket = undefined} = NewData} -> %% Socket was closed during body read - transition to closed state {next_state, closed, NewData, [{reply, From, done}]}; {done, NewData} -> {next_state, connected, NewData, [{reply, From, done}]}; {error, Reason} -> {next_state, closed, Data, [{reply, From, {error, Reason}}]} end; receiving({call, From}, get_state, _Data) -> {keep_state_and_data, [{reply, From, {ok, receiving}}]}; receiving(info, {tcp_closed, Socket}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; receiving(info, {ssl_closed, Socket}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; receiving(info, {tcp_error, Socket, _Reason}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; receiving(info, {ssl_error, Socket, _Reason}, #conn_data{socket = Socket} = Data) -> {next_state, closed, Data#conn_data{socket = undefined}}; receiving(info, {'DOWN', Ref, process, _Pid, _Reason}, #conn_data{owner_mon = Ref} = Data) -> {stop, normal, Data}; receiving(EventType, Event, Data) -> handle_common(EventType, Event, receiving, Data). %%==================================================================== %% State: streaming - Continuous async body streaming (async=true) %%==================================================================== streaming(enter, receiving, Data) -> %% Start streaming immediately by sending self a cast gen_statem:cast(self(), do_stream), {keep_state, Data}; streaming(enter, streaming, _Data) -> keep_state_and_data; streaming(cast, do_stream, Data) -> %% Stream body chunks continuously #conn_data{async_ref = Ref, stream_to = StreamTo} = Data, case stream_body_chunk(Data) of {ok, Chunk, NewData} -> StreamTo ! {hackney_response, Ref, Chunk}, %% Continue streaming gen_statem:cast(self(), do_stream), {keep_state, NewData}; {done, NewData} -> StreamTo ! {hackney_response, Ref, done}, %% Check Connection: close header and pool status finish_async_streaming(NewData); {error, Reason} -> StreamTo ! {hackney_response, Ref, {error, Reason}}, {next_state, closed, Data} end; streaming(cast, pause_stream, Data) -> %% Pause - transition to streaming_once which waits for explicit next {next_state, streaming_once, Data}; streaming({call, From}, stop_async, Data) -> {next_state, receiving, reset_async(Data), [{reply, From, ok}]}; streaming({call, From}, get_state, _Data) -> {keep_state_and_data, [{reply, From, {ok, streaming}}]}; streaming(info, {tcp_closed, Socket}, #conn_data{socket = Socket, async_ref = Ref, stream_to = StreamTo} = Data) -> StreamTo ! {hackney_response, Ref, {error, closed}}, {next_state, closed, Data#conn_data{socket = undefined}}; streaming(info, {ssl_closed, Socket}, #conn_data{socket = Socket, async_ref = Ref, stream_to = StreamTo} = Data) -> StreamTo ! {hackney_response, Ref, {error, closed}}, {next_state, closed, Data#conn_data{socket = undefined}}; streaming(info, {tcp_error, Socket, Reason}, #conn_data{socket = Socket, async_ref = Ref, stream_to = StreamTo} = Data) -> StreamTo ! {hackney_response, Ref, {error, Reason}}, {next_state, closed, Data#conn_data{socket = undefined}}; streaming(info, {ssl_error, Socket, Reason}, #conn_data{socket = Socket, async_ref = Ref, stream_to = StreamTo} = Data) -> StreamTo ! {hackney_response, Ref, {error, Reason}}, {next_state, closed, Data#conn_data{socket = undefined}}; streaming(info, {'DOWN', Ref, process, _Pid, _Reason}, #conn_data{owner_mon = Ref} = Data) -> {stop, normal, Data}; streaming(EventType, Event, Data) -> handle_common(EventType, Event, streaming, Data). %%==================================================================== %% State: streaming_once - On-demand async streaming (async=once) %%==================================================================== streaming_once(enter, _OldState, _Data) -> %% Wait for stream_next keep_state_and_data; streaming_once(cast, stream_next, Data) -> %% Stream one chunk #conn_data{async_ref = Ref, stream_to = StreamTo} = Data, case stream_body_chunk(Data) of {ok, Chunk, NewData} -> StreamTo ! {hackney_response, Ref, Chunk}, {keep_state, NewData}; {done, NewData} -> StreamTo ! {hackney_response, Ref, done}, %% Check Connection: close header and pool status finish_async_streaming(NewData); {error, Reason} -> StreamTo ! {hackney_response, Ref, {error, Reason}}, {next_state, closed, Data} end; streaming_once(cast, resume_stream, Data) -> %% Switch to continuous mode {next_state, streaming, Data#conn_data{async = true}}; streaming_once({call, From}, stop_async, Data) -> {next_state, receiving, reset_async(Data), [{reply, From, ok}]}; streaming_once({call, From}, get_state, _Data) -> {keep_state_and_data, [{reply, From, {ok, streaming_once}}]}; streaming_once(info, {tcp_closed, Socket}, #conn_data{socket = Socket, async_ref = Ref, stream_to = StreamTo} = Data) -> StreamTo ! {hackney_response, Ref, {error, closed}}, {next_state, closed, Data#conn_data{socket = undefined}}; streaming_once(info, {ssl_closed, Socket}, #conn_data{socket = Socket, async_ref = Ref, stream_to = StreamTo} = Data) -> StreamTo ! {hackney_response, Ref, {error, closed}}, {next_state, closed, Data#conn_data{socket = undefined}}; streaming_once(info, {tcp_error, Socket, Reason}, #conn_data{socket = Socket, async_ref = Ref, stream_to = StreamTo} = Data) -> StreamTo ! {hackney_response, Ref, {error, Reason}}, {next_state, closed, Data#conn_data{socket = undefined}}; streaming_once(info, {ssl_error, Socket, Reason}, #conn_data{socket = Socket, async_ref = Ref, stream_to = StreamTo} = Data) -> StreamTo ! {hackney_response, Ref, {error, Reason}}, {next_state, closed, Data#conn_data{socket = undefined}}; streaming_once(info, {'DOWN', Ref, process, _Pid, _Reason}, #conn_data{owner_mon = Ref} = Data) -> {stop, normal, Data}; streaming_once(EventType, Event, Data) -> handle_common(EventType, Event, streaming_once, Data). %%==================================================================== %% State: closed - Connection terminated %%==================================================================== closed(enter, _OldState, #conn_data{socket = Socket, transport = Transport, pool_pid = PoolPid} = Data) -> %% Close socket if still open case Socket of undefined -> ok; _ -> Transport:close(Socket) end, %% Pooled connections used to stop immediately here, but that made %% late-arriving {call, From, {request, _}} messages from workers that %% raced the pool checkout race a terminating gen_statem — which %% surfaces as `exit:{normal, _}` in the caller (issue #836). Stay %% alive briefly so those late calls get a proper `{error, {closed, _}}` %% reply via handle_common's invalid_state fallback, then stop. case PoolPid of undefined -> {keep_state, Data#conn_data{socket = undefined}}; _ -> {keep_state, Data#conn_data{socket = undefined}, [{state_timeout, ?CLOSED_GRACE_MS, closed_grace_expired}]} end; closed(state_timeout, closed_grace_expired, Data) -> {stop, normal, Data}; closed({call, From}, connect, Data) -> %% Allow reconnection from closed state #conn_data{ host = Host, port = Port, transport = Transport, connect_timeout = Timeout, connect_options = ConnectOpts, ssl_options = SslOpts } = Data, %% Build connection options Opts = case Transport of hackney_ssl -> ConnectOpts ++ SslOpts; _ -> ConnectOpts end, %% Attempt connection case Transport:connect(Host, Port, Opts, Timeout) of {ok, Socket} -> %% Connection successful NewData = Data#conn_data{socket = Socket}, {next_state, connected, NewData, [{reply, From, ok}]}; {error, Reason} -> %% Connection failed, stay in closed state {keep_state_and_data, [{reply, From, {error, Reason}}]} end; closed({call, From}, get_state, _Data) -> {keep_state_and_data, [{reply, From, {ok, closed}}]}; closed(info, {'DOWN', Ref, process, _Pid, _Reason}, #conn_data{owner_mon = Ref} = Data) -> {stop, normal, Data}; closed(cast, {set_owner, _NewOwner}, #conn_data{pool_pid = PoolPid} = Data) when is_pid(PoolPid) -> %% #850: the pool tried to hand ownership to a pooled connection that %% already closed (async checkin/prewarm raced a server-side close). Stop %% now so the pool's monitor removes us from `available` promptly, instead %% of lingering through the grace window and being handed out again. {stop, normal, Data}; closed(EventType, Event, Data) -> handle_common(EventType, Event, closed, Data). %%==================================================================== %% Internal functions %%==================================================================== %% @private Handle common events across all states %% Handle enter events for state transitions (required when using state_enter callback mode) handle_common(enter, _OldState, _NewState, _Data) -> keep_state_and_data; handle_common(cast, stop, _State, Data) -> %% Async stop - used by pool to avoid deadlock during sync checkin {stop, normal, Data}; handle_common(cast, _Msg, _State, _Data) -> keep_state_and_data; %% Socket operations - available in any state with a socket handle_common({call, From}, {setopts, Opts}, _State, #conn_data{transport = Transport, socket = Socket} = _Data) when Socket =/= undefined -> Result = Transport:setopts(Socket, Opts), {keep_state_and_data, [{reply, From, Result}]}; handle_common({call, From}, peername, _State, #conn_data{transport = Transport, socket = Socket} = _Data) when Socket =/= undefined -> Result = Transport:peername(Socket), {keep_state_and_data, [{reply, From, Result}]}; handle_common({call, From}, peercert, _State, #conn_data{transport = Transport, socket = Socket} = _Data) when Socket =/= undefined -> Result = case erlang:function_exported(Transport, peercert, 1) of true -> Transport:peercert(Socket); false -> {error, not_ssl} end, {keep_state_and_data, [{reply, From, Result}]}; handle_common({call, From}, sockname, _State, #conn_data{transport = Transport, socket = Socket} = _Data) when Socket =/= undefined -> Result = Transport:sockname(Socket), {keep_state_and_data, [{reply, From, Result}]}; %% HTTP/3 socket operations — peername/sockname/peercert delegated to the %% underlying quic connection; setopts has no analog over QUIC. handle_common({call, From}, peername, _State, #conn_data{h3_conn = H3Conn} = _Data) when H3Conn =/= undefined -> {keep_state_and_data, [{reply, From, hackney_h3:peername(H3Conn)}]}; handle_common({call, From}, sockname, _State, #conn_data{h3_conn = H3Conn} = _Data) when H3Conn =/= undefined -> {keep_state_and_data, [{reply, From, hackney_h3:sockname(H3Conn)}]}; handle_common({call, From}, peercert, _State, #conn_data{h3_conn = H3Conn} = _Data) when H3Conn =/= undefined -> {keep_state_and_data, [{reply, From, hackney_h3:peercert(H3Conn)}]}; handle_common({call, From}, {setopts, _Opts}, _State, #conn_data{h3_conn = H3Conn} = _Data) when H3Conn =/= undefined -> {keep_state_and_data, [{reply, From, {error, not_supported}}]}; %% Low-level send operation handle_common({call, From}, {send, Data}, _State, #conn_data{transport = Transport, socket = Socket} = _Data) when Socket =/= undefined -> Result = Transport:send(Socket, Data), {keep_state_and_data, [{reply, From, Result}]}; %% Low-level recv operation handle_common({call, From}, {recv, Length, Timeout}, _State, #conn_data{transport = Transport, socket = Socket} = _Data) when Socket =/= undefined -> Result = Transport:recv(Socket, Length, Timeout), {keep_state_and_data, [{reply, From, Result}]}; %% Close socket but keep process handle_common({call, From}, close_socket, _State, #conn_data{transport = Transport, socket = Socket} = Data) when Socket =/= undefined -> Transport:close(Socket), {next_state, closed, Data#conn_data{socket = undefined}, [{reply, From, ok}]}; handle_common({call, From}, close_socket, _State, Data) -> {next_state, closed, Data, [{reply, From, ok}]}; handle_common({call, From}, {send, _SendData}, _State, _Data) -> {keep_state_and_data, [{reply, From, {error, not_connected}}]}; handle_common({call, From}, {recv, _Length, _Timeout}, _State, _Data) -> {keep_state_and_data, [{reply, From, {error, not_connected}}]}; handle_common({call, From}, {setopts, _Opts}, _State, _Data) -> {keep_state_and_data, [{reply, From, {error, not_connected}}]}; handle_common({call, From}, peername, _State, _Data) -> {keep_state_and_data, [{reply, From, {error, not_connected}}]}; handle_common({call, From}, peercert, _State, _Data) -> {keep_state_and_data, [{reply, From, {error, not_connected}}]}; handle_common({call, From}, sockname, _State, _Data) -> {keep_state_and_data, [{reply, From, {error, not_connected}}]}; handle_common({call, From}, response_headers, _State, #conn_data{response_headers = Headers}) -> {keep_state_and_data, [{reply, From, Headers}]}; handle_common({call, From}, get_location, _State, #conn_data{location = Location}) -> {keep_state_and_data, [{reply, From, Location}]}; handle_common({call, From}, {set_location, Location}, _State, Data) -> {keep_state, Data#conn_data{location = Location}, [{reply, From, ok}]}; handle_common({call, From}, is_upgraded_ssl, _State, #conn_data{upgraded_ssl = Upgraded}) -> {keep_state_and_data, [{reply, From, Upgraded}]}; handle_common({call, From}, is_no_reuse, _State, #conn_data{no_reuse = NoReuse}) -> {keep_state_and_data, [{reply, From, NoReuse}]}; handle_common({call, From}, get_protocol, _State, #conn_data{protocol = Protocol}) -> {keep_state_and_data, [{reply, From, Protocol}]}; handle_common({call, From}, _, _State, _Data) -> {keep_state_and_data, [{reply, From, {error, invalid_state}}]}; %% QUIC socket ready - drive event loop (common handler for all states) handle_common(info, {select, _Resource, _Ref, ready_input}, _State, #conn_data{h3_conn = ConnRef}) when ConnRef =/= undefined -> _ = hackney_h3:process(ConnRef), keep_state_and_data; %% HTTP/3 0-RTT/resumption: cache the session ticket in the pool so the next %% connection to this host can resume. Handled here (common to all H3 states) %% since the ticket can arrive at any point after the handshake. handle_common(info, {h3, ConnRef, {session_ticket, Ticket}}, _State, #conn_data{h3_conn = ConnRef} = Data) -> maybe_store_h3_session(Ticket, Data), {keep_state, Data#conn_data{h3_session_ticket = Ticket}}; %% HTTP/3 0-RTT rejection. Requests on this path are sent at 1-RTT, so this is %% not expected; if it arrives, invalidate the cached ticket and fail any %% matching in-flight stream so the caller can retry. handle_common(info, {h3, ConnRef, {early_data_rejected, StreamIds}}, _State, #conn_data{h3_conn = ConnRef, h3_streams = Streams} = Data) -> maybe_delete_h3_session(Data), {NewStreams, Actions} = fail_rejected_h3_streams(StreamIds, Streams), {keep_state, Data#conn_data{h3_streams = NewStreams}, Actions}; %% With trap_exit = true, an EXIT signal from any linked process (other than %% h2_conn, handled in connected/3) arrives here. Swallow it rather than %% propagating — avoids tearing down the gen_statem on stray links. handle_common(info, {'EXIT', _Pid, _Reason}, _State, _Data) -> keep_state_and_data; handle_common(info, _Msg, _State, _Data) -> keep_state_and_data. %% @private Reply to caller and stop reply_and_stop(undefined, _Reply, Data) -> {stop, normal, Data}; reply_and_stop(From, Reply, _Data) -> {stop_and_reply, normal, [{reply, From, Reply}]}. %% @private Reset async state reset_async(Data) -> Data#conn_data{ async = false, async_ref = undefined, stream_to = undefined }. %% @private Check if connection should be closed based on response headers should_close_connection(#conn_data{response_headers = undefined}) -> false; should_close_connection(#conn_data{response_headers = Headers}) -> case hackney_headers:get_value(<<"connection">>, Headers) of undefined -> false; Value -> hackney_bstr:to_lower(Value) =:= <<"close">> end. %% @private Finish async streaming - close or return to connected based on Connection header finish_async_streaming(Data) -> #conn_data{transport = Transport, socket = Socket, pool_pid = PoolPid} = Data, case should_close_connection(Data) of true -> %% Connection: close - close socket and stop process case Socket of undefined -> ok; _ -> Transport:close(Socket) end, {stop, normal, reset_async(Data#conn_data{socket = undefined})}; false when PoolPid =:= undefined -> %% No pool and no Connection: close - still stop since no reuse case Socket of undefined -> ok; _ -> Transport:close(Socket) end, {stop, normal, reset_async(Data#conn_data{socket = undefined})}; false -> %% Has pool or keep-alive - return to connected for reuse {next_state, connected, reset_async(Data)} end. %% @private Handle receiving response status and headers %% IncludePid determines whether to include pid in the response do_recv_response_impl(Data, IncludePid) -> Parser = hackney_http:parser([response]), DataWithParser = Data#conn_data{parser = Parser}, case recv_status_and_headers(DataWithParser) of {ok, Status, Headers, NewData} -> From = NewData#conn_data.request_from, HeadersList = hackney_headers:to_list(Headers), maybe_record_altsvc(HeadersList, NewData), Reply = case IncludePid of true -> {ok, Status, HeadersList, self()}; false -> {ok, Status, HeadersList} end, {keep_state, NewData#conn_data{ request_from = undefined, response_headers = Headers }, [{reply, From, Reply}]}; {error, Reason} -> From = Data#conn_data.request_from, {next_state, closed, Data, [{reply, From, {error, Reason}}]} end. %% @private Send HTTP request (shared by sync and async) do_send_request(Method, Path, Headers, Body, Data) -> #conn_data{ transport = Transport, socket = Socket, netloc = Netloc } = Data, %% Build request headers FinalHeaders = build_headers(Method, Headers, Body, Netloc), %% Build request line and headers Path1 = case Path of <<>> -> <<"/">>; _ -> Path end, RequestLine = <>, HeadersBin = headers_to_binary(FinalHeaders), RequestData = <>, %% Send request case Transport:send(Socket, RequestData) of ok -> %% Send body if present case send_body(Transport, Socket, Body) of ok -> {ok, Data}; {error, Reason} -> {error, Reason} end; {error, Reason} -> {error, Reason} end. %% @private Compute netloc for Host header compute_netloc(Host, Port, Transport) -> HostBin = hackney_bstr:to_binary(Host), case {Transport, Port} of {hackney_tcp, 80} -> HostBin; {hackney_ssl, 443} -> HostBin; _ -> PortBin = integer_to_binary(Port), <> end. %% @private Build request headers build_headers(_Method, Headers0, Body, Netloc) -> %% Start with user headers Headers1 = hackney_headers:new(Headers0), %% Add Host header if not present {_, Headers2} = hackney_headers:store_new(<<"Host">>, Netloc, Headers1), %% Add User-Agent if not present {_, Headers3} = hackney_headers:store_new(<<"User-Agent">>, default_ua(), Headers2), %% Add Content-Length for bodies case Body of <<>> -> Headers3; [] -> Headers3; _ when is_binary(Body) -> Len = byte_size(Body), case hackney_headers:is_key(<<"content-length">>, Headers3) of true -> Headers3; false -> hackney_headers:store(<<"Content-Length">>, integer_to_binary(Len), Headers3) end; _ when is_list(Body) -> Len = iolist_size(Body), case hackney_headers:is_key(<<"content-length">>, Headers3) of true -> Headers3; false -> hackney_headers:store(<<"Content-Length">>, integer_to_binary(Len), Headers3) end; _ -> %% Streaming body - expect user to have set Content-Length or Transfer-Encoding Headers3 end. %% @private Convert headers to binary headers_to_binary(Headers) -> hackney_headers:to_binary(Headers). %% @private Build request line build_request_line(Method, Path) -> [Method, <<" ">>, Path, <<" HTTP/1.1\r\n">>]. %% @private Encode a chunk for chunked transfer encoding encode_chunk(Data) when is_binary(Data) -> Size = byte_size(Data), SizeHex = integer_to_binary(Size, 16), [SizeHex, <<"\r\n">>, Data, <<"\r\n">>]; encode_chunk(Data) when is_list(Data) -> Size = iolist_size(Data), SizeHex = integer_to_binary(Size, 16), [SizeHex, <<"\r\n">>, Data, <<"\r\n">>]. %% @private Stream body from stateless function %% fun() -> {ok, Data} | eof | {error, Reason} stream_body_fun(Transport, Socket, Fun) when is_function(Fun, 0) -> case Fun() of {ok, Data} -> case Transport:send(Socket, encode_chunk(Data)) of ok -> stream_body_fun(Transport, Socket, Fun); Error -> Error end; eof -> ok; {error, _} = Error -> Error end; %% @private Stream body from stateful function %% fun(State) -> {ok, Data, NewState} | eof | {error, Reason} stream_body_fun(Transport, Socket, {Fun, State}) when is_function(Fun, 1) -> case Fun(State) of {ok, Data, NewState} -> case Transport:send(Socket, encode_chunk(Data)) of ok -> stream_body_fun(Transport, Socket, {Fun, NewState}); Error -> Error end; eof -> ok; {error, _} = Error -> Error end. %% @private Stream body from function for HTTP/3 stream_body_fun_h3(ConnRef, StreamId, Fun) when is_function(Fun, 0) -> case Fun() of {ok, Data} -> case hackney_h3:send_body_chunk(ConnRef, StreamId, iolist_to_binary(Data), false) of ok -> stream_body_fun_h3(ConnRef, StreamId, Fun); Error -> Error end; eof -> ok; {error, _} = Error -> Error end; stream_body_fun_h3(ConnRef, StreamId, {Fun, State}) when is_function(Fun, 1) -> case Fun(State) of {ok, Data, NewState} -> case hackney_h3:send_body_chunk(ConnRef, StreamId, iolist_to_binary(Data), false) of ok -> stream_body_fun_h3(ConnRef, StreamId, {Fun, NewState}); Error -> Error end; eof -> ok; {error, _} = Error -> Error end. %% @private Get default User-Agent default_ua() -> Version = case application:get_key(hackney, vsn) of {ok, FullVersion} -> list_to_binary(hd(string:tokens(FullVersion, "-"))); _ -> <<"0.0.0">> end, <<"hackney/", Version/binary>>. %% @private Send request body send_body(_Transport, _Socket, <<>>) -> ok; send_body(_Transport, _Socket, []) -> ok; send_body(_Transport, _Socket, stream) -> %% Stream body mode - body will be sent separately ok; send_body(Transport, Socket, Body) when is_binary(Body); is_list(Body) -> Transport:send(Socket, Body). %% @private Receive and parse response status and headers %% Handles 1XX informational responses per RFC 7231 recv_status_and_headers(Data) -> recv_status_and_headers_loop(Data). recv_status_and_headers_loop(Data) -> case recv_status(Data) of {ok, Status, Headers, NewData} when Status >= 100, Status < 200 -> %% 1XX informational response per RFC 7231 Section 6.2 #conn_data{inform_fun = InformFun, reason = Reason, parser = OldParser, stream_to = StreamTo, async_ref = AsyncRef} = NewData, HeadersList = hackney_headers:to_list(Headers), %% Handle differently for sync vs async mode case StreamTo of undefined -> %% Sync mode - call callback if provided case InformFun of undefined -> ok; Fun when is_function(Fun, 3) -> Fun(Status, Reason, HeadersList) end; _ -> %% Async mode - send message to stream_to StreamTo ! {hackney_response, AsyncRef, {informational, Status, Reason, HeadersList}} end, %% Reset parser for next response, preserving any buffered data %% The old parser's buffer may contain the start of the next response OldBuffer = hackney_http:get(OldParser, buffer), NewParser = hackney_http:parser([response]), recv_status_and_headers_loop(NewData#conn_data{parser = NewParser, buffer = OldBuffer}); Other -> Other end. recv_status(#conn_data{parser = Parser, buffer = Buffer} = Data) -> case hackney_http:execute(Parser, Buffer) of {more, NewParser} -> %% Check if parser has data in its internal buffer (e.g., after skipping 1XX response) %% If so, continue parsing; otherwise read from socket ParserBuffer = hackney_http:get(NewParser, buffer), case ParserBuffer of <<>> -> %% Parser buffer empty - need to read from socket case recv_data(Data) of {ok, RecvData} -> recv_status(Data#conn_data{parser = NewParser, buffer = RecvData}); {error, Reason} -> {error, Reason} end; _ -> %% Parser has data in buffer - continue parsing without reading recv_status(Data#conn_data{parser = NewParser, buffer = <<>>}) end; {response, Version, Status, Reason, NewParser} -> recv_headers(Data#conn_data{ parser = NewParser, buffer = <<>>, version = Version, status = Status, reason = Reason }, hackney_headers:new()); {error, Reason} -> {error, Reason} end. recv_headers(#conn_data{parser = Parser} = Data, Headers) -> case hackney_http:execute(Parser) of {more, NewParser} -> case recv_data(Data) of {ok, RecvData} -> case hackney_http:execute(NewParser, RecvData) of {header, {Key, Value}, NewParser2} -> Headers2 = hackney_headers:append(Key, Value, Headers), recv_headers(Data#conn_data{parser = NewParser2}, Headers2); {headers_complete, NewParser2} -> {ok, Data#conn_data.status, Headers, Data#conn_data{parser = NewParser2}}; {more, NewParser2} -> recv_headers(Data#conn_data{parser = NewParser2, buffer = <<>>}, Headers); {error, Reason} -> {error, Reason} end; {error, Reason} -> {error, Reason} end; {header, {Key, Value}, NewParser} -> Headers2 = hackney_headers:append(Key, Value, Headers), recv_headers(Data#conn_data{parser = NewParser}, Headers2); {headers_complete, NewParser} -> {ok, Data#conn_data.status, Headers, Data#conn_data{parser = NewParser}}; {error, Reason} -> {error, Reason} end. %% @private Read full response body read_full_body(#conn_data{method = <<"HEAD">>} = Data, Acc) -> %% HEAD requests have no body {ok, Acc, Data}; read_full_body(#conn_data{status = Status} = Data, Acc) when Status =:= 204; Status =:= 304 -> %% 204 No Content and 304 Not Modified have no body per RFC 7230 3.3.3 %% Force connection close to avoid corrupting subsequent requests if %% a misbehaving server sent Content-Length or body data {ok, Acc, Data#conn_data{socket = undefined}}; read_full_body(Data, Acc) -> case stream_body_chunk(Data) of {ok, Chunk, NewData} -> read_full_body(NewData, <>); {done, NewData} -> %% Body complete - apply decompression if needed maybe_decompress_body(Acc, NewData); {error, Reason} -> {error, Reason} end. %% @private Apply decompression if auto_decompress is enabled and Content-Encoding is set maybe_decompress_body(Body, #conn_data{auto_decompress = false} = Data) -> {ok, Body, Data}; maybe_decompress_body(Body, #conn_data{response_headers = undefined} = Data) -> {ok, Body, Data}; maybe_decompress_body(Body, #conn_data{response_headers = Headers, auto_decompress = true} = Data) -> %% Get Content-Encoding from response headers ContentEncoding = case hackney_headers:get_value(<<"content-encoding">>, Headers) of undefined -> <<>>; CE -> hackney_bstr:to_lower(CE) end, case ContentEncoding of <<"gzip">> -> case decompress_gzip(Body) of {ok, Decompressed} -> {ok, Decompressed, Data}; {error, Reason} -> {error, {decompress_error, gzip, Reason}} end; <<"deflate">> -> case decompress_deflate(Body) of {ok, Decompressed} -> {ok, Decompressed, Data}; {error, Reason} -> {error, {decompress_error, deflate, Reason}} end; <<"x-gzip">> -> %% x-gzip is an alias for gzip case decompress_gzip(Body) of {ok, Decompressed} -> {ok, Decompressed, Data}; {error, Reason} -> {error, {decompress_error, gzip, Reason}} end; _ -> %% No compression or unknown encoding - return as-is {ok, Body, Data} end. %% @private Decompress gzip-encoded data decompress_gzip(Data) -> try {ok, zlib:gunzip(Data)} catch error:Reason -> {error, Reason}; exit:Reason -> {error, Reason} end. %% @private Decompress deflate-encoded data %% Note: Some servers send raw deflate, others send zlib-wrapped deflate decompress_deflate(Data) -> %% Try zlib-wrapped first (RFC 1950), then raw deflate (RFC 1951) try Z = zlib:open(), try ok = zlib:inflateInit(Z), Decompressed = zlib:inflate(Z, Data), ok = zlib:inflateEnd(Z), {ok, iolist_to_binary(Decompressed)} after zlib:close(Z) end catch error:Reason -> {error, Reason}; exit:Reason -> {error, Reason} end. %% @private Stream a single body chunk stream_body_chunk(#conn_data{method = <<"HEAD">>} = Data) -> {done, Data}; stream_body_chunk(#conn_data{status = Status} = Data) when Status =:= 204; Status =:= 304 -> %% 204 No Content and 304 Not Modified have no body per RFC 7230 3.3.3 %% Force connection close to avoid corrupting subsequent requests if %% a misbehaving server sent Content-Length or body data {done, Data#conn_data{socket = undefined}}; stream_body_chunk(#conn_data{parser = Parser, transport = Transport, socket = Socket, recv_timeout = Timeout} = Data) -> case hackney_http:execute(Parser) of {more, NewParser, _Buffer} -> %% Need more data case Transport:recv(Socket, 0, Timeout) of {ok, RecvData} -> stream_body_chunk_result(hackney_http:execute(NewParser, RecvData), Data); {error, closed} -> %% Connection closed by server - mark socket as undefined {done, Data#conn_data{socket = undefined}}; {error, Reason} -> {error, Reason} end; {more, NewParser} -> %% Need more data case Transport:recv(Socket, 0, Timeout) of {ok, RecvData} -> %% Execute with new data and handle result stream_body_chunk_result(hackney_http:execute(NewParser, RecvData), Data); {error, closed} -> %% Connection closed by server - mark socket as undefined {done, Data#conn_data{socket = undefined}}; {error, Reason} -> {error, Reason} end; {ok, Chunk, NewParser} -> {ok, Chunk, Data#conn_data{parser = NewParser}}; {done, Rest} -> {done, Data#conn_data{buffer = Rest, parser = undefined}}; done -> {done, Data#conn_data{buffer = <<>>, parser = undefined}}; {error, Reason} -> {error, Reason} end. %% @private Handle result of parsing received data stream_body_chunk_result({ok, Chunk, NewParser}, Data) -> {ok, Chunk, Data#conn_data{parser = NewParser}}; stream_body_chunk_result({done, Rest}, Data) -> {done, Data#conn_data{buffer = Rest, parser = undefined}}; stream_body_chunk_result(done, Data) -> {done, Data#conn_data{buffer = <<>>, parser = undefined}}; stream_body_chunk_result({more, NewParser}, Data) -> stream_body_chunk(Data#conn_data{parser = NewParser}); stream_body_chunk_result({more, NewParser, _Buffer}, Data) -> stream_body_chunk(Data#conn_data{parser = NewParser}); stream_body_chunk_result({error, Reason}, _Data) -> {error, Reason}. %% @private Receive data from socket recv_data(#conn_data{transport = Transport, socket = Socket, recv_timeout = Timeout}) -> Transport:recv(Socket, 0, Timeout). %% @private Determine if we should enable active mode when entering connected state %% We only want active mode for close detection when the connection is truly idle %% (returning from a completed request), not on initial connection where the server %% might send data before we send our request (as in pipelining or some test scenarios). should_enable_active_mode(receiving) -> true; should_enable_active_mode(streaming) -> true; should_enable_active_mode(streaming_once) -> true; should_enable_active_mode(closed) -> true; %% Reconnection from closed state should_enable_active_mode(_) -> false. %% @private Determine if we should auto-release to pool when entering connected state %% We release when body reading is complete (coming from receiving state) %% but NOT from streaming states (user is still actively streaming) should_auto_release(receiving) -> true; should_auto_release(_) -> false. %% @private Auto-release connection back to pool %% Transfers ownership to pool and notifies it asynchronously auto_release_to_pool(#conn_data{pool_pid = PoolPid, owner_mon = OldMon} = Data) -> %% Transfer ownership to pool demonitor(OldMon, [flush]), NewMon = monitor(process, PoolPid), Data2 = Data#conn_data{owner = PoolPid, owner_mon = NewMon}, %% Notify pool asynchronously (avoid blocking the state machine) notify_pool_available(Data2), Data2. %% @private Check if socket is healthy (not closed by peer) %% Note: With active mode enabled on connected sockets, we rely on tcp_closed/ssl_closed %% messages to detect server closes. This function now uses peername to verify the %% socket is still valid without conflicting with active mode. check_socket_health(Transport, Socket) -> %% Use peername to check if socket is still connected %% This works regardless of active/passive mode Module = case Transport of hackney_ssl -> ssl; hackney_tcp -> inet; _ -> inet end, case Module:peername(Socket) of {ok, _} -> %% Socket is connected ok; {error, Reason} -> %% Socket error (closed, einval, etc.) {error, Reason} end. %% @private Check for pending close message in mailbox %% Returns true if a close message is waiting, false otherwise. has_pending_close(Socket) -> receive {tcp_closed, Socket} -> true; {ssl_closed, Socket} -> true; {tcp_error, Socket, _} -> true; {ssl_error, Socket, _} -> true after 0 -> false end. %% @private Flush any socket data messages that arrived while in active mode %% This is called after setting the socket to passive mode. %% Note: Close messages are checked separately via has_pending_close/1. flush_socket_messages(Socket) -> receive {tcp, Socket, _Data} -> flush_socket_messages(Socket); {ssl, Socket, _Data} -> flush_socket_messages(Socket) after 0 -> ok end. %% @private Notify pool that connection is available for reuse (async) notify_pool_available(#conn_data{pool_pid = undefined}) -> %% Not from a pool, nothing to do ok; notify_pool_available(#conn_data{pool_pid = PoolPid}) -> %% Tell the pool this connection is available gen_server:cast(PoolPid, {checkin, nil, self()}). %% @private Notify pool that connection is available for reuse (sync) %% This ensures the pool has processed the checkin before returning. %% We pass a "should close" flag to avoid deadlock (pool can't call back to us). %% The flag is true if connection was SSL upgraded or is a proxy tunnel (no_reuse). notify_pool_available_sync(#conn_data{pool_pid = undefined}) -> ok; notify_pool_available_sync(#conn_data{pool_pid = PoolPid, upgraded_ssl = UpgradedSsl, no_reuse = NoReuse}) -> ShouldClose = UpgradedSsl orelse NoReuse, gen_server:call(PoolPid, {checkin_sync, self(), ShouldClose}, 5000). %% @private Start an async request do_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, Data) -> %% Use self() (connection PID) as the async ref for message correlation Ref = self(), %% Issue #646: If StreamTo is different from the caller, set StreamTo as owner. %% This way the connection lifecycle is tied to the message recipient. %% If StreamTo dies, the connection will be terminated. {CallerPid, _} = From, Data2 = case StreamTo =:= CallerPid of true -> Data; false -> %% Change owner to StreamTo demonitor(Data#conn_data.owner_mon, [flush]), NewMon = monitor(process, StreamTo), Data#conn_data{owner = StreamTo, owner_mon = NewMon} end, NewData = Data2#conn_data{ request_from = From, method = Method, path = Path, parser = undefined, version = undefined, status = undefined, reason = undefined, response_headers = undefined, buffer = <<>>, async = AsyncMode, async_ref = Ref, stream_to = StreamTo, follow_redirect = FollowRedirect }, {next_state, sending, NewData, [{next_event, internal, {send_request_async, Method, Path, Headers, Body}}]}. %% @private Check if response is a redirect that should be handled maybe_handle_async_redirect(Status, _Method, Headers, true) when Status =:= 301; Status =:= 302; Status =:= 307; Status =:= 308 -> %% Redirect status - get location case hackney_headers:get_value(<<"location">>, Headers) of undefined -> no_redirect; Location -> {redirect, Location} end; maybe_handle_async_redirect(303, <<"POST">>, Headers, true) -> %% 303 See Other for POST - should redirect as GET case hackney_headers:get_value(<<"location">>, Headers) of undefined -> no_redirect; Location -> {see_other, Location} end; maybe_handle_async_redirect(_, _, _, _) -> no_redirect. %% @private Skip the response body skip_response_body(Data) -> case read_full_body(Data, <<>>) of {ok, _, NewData} -> NewData; {error, _} -> Data end. %%==================================================================== %% HTTP/2 Support Functions %%==================================================================== %% @private Try HTTP/3 connection via QUIC %% lsquic handles its own UDP socket creation and DNS resolution. try_h3_connect(Host, Port, Timeout, ConnectOpts, SslOpts) -> HostBin = if is_list(Host) -> list_to_binary(Host); true -> Host end, case hackney_h3:connect(HostBin, Port, h3_tls_opts(ConnectOpts, SslOpts), self()) of {ok, ConnRef} -> %% Drive event loop until connected wait_h3_connected(ConnRef, Timeout, erlang:monotonic_time(millisecond)); {error, _} = Error -> Error end. %% @private Map hackney's TLS options to the QUIC client verification. %% quic >= 1.4.4 verifies the server certificate. An insecure connection opts %% out; an explicitly configured CA (cacerts/cacertfile) is used as the trust %% store; otherwise quic verifies against its own default (OS) trust store. %% `family' (inet|inet6) and a `session_ticket' (for resumption) are forwarded %% when present in either the connect options or the ssl options. h3_tls_opts(ConnectOpts, SslOpts) -> Insecure = proplists:get_value(insecure, ConnectOpts, proplists:get_value(insecure, SslOpts, false)), Base = case Insecure of true -> #{verify => verify_none}; false -> h3_ca_opts(SslOpts) end, Base1 = case proplists:get_value(family, ConnectOpts, proplists:get_value(family, SslOpts, undefined)) of undefined -> Base; Family -> Base#{family => Family} end, case proplists:get_value(session_ticket, ConnectOpts, proplists:get_value(session_ticket, SslOpts, undefined)) of undefined -> Base1; Ticket -> Base1#{session_ticket => Ticket} end. %% @private Use an explicitly configured CA as the H3 trust store. quic only %% accepts DER cacerts, so a cacertfile is decoded here. With no CA configured %% the map is empty and quic falls back to its default trust store. h3_ca_opts(SslOpts) -> case proplists:get_value(cacerts, SslOpts) of undefined -> case proplists:get_value(cacertfile, SslOpts) of undefined -> #{}; File -> #{cacerts => cacertfile_ders(File)} end; CACerts -> #{cacerts => CACerts} end. %% @private Read a PEM cacertfile into a list of DER certificates. cacertfile_ders(File) -> case file:read_file(File) of {ok, Pem} -> [Der || {'Certificate', Der, _} <- public_key:pem_decode(Pem)]; {error, _} -> [] end. %% @private Drive QUIC event loop until connected wait_h3_connected(ConnRef, Timeout, StartTime) -> Elapsed = erlang:monotonic_time(millisecond) - StartTime, Remaining = max(0, Timeout - Elapsed), receive {select, _Resource, _Ref, ready_input} -> _ = hackney_h3:process(ConnRef), wait_h3_connected(ConnRef, Timeout, StartTime); {h3, ConnRef, {connected, _Info}} -> {ok, ConnRef}; {h3, ConnRef, {closed, Reason}} -> {error, {quic_closed, Reason}}; {h3, ConnRef, {transport_error, Code, Msg}} -> {error, {quic_error, Code, Msg}} after Remaining -> hackney_h3:close(ConnRef, timeout), {error, timeout} end. %% @private Standard TCP/TLS connection do_tcp_connect(From, Data) -> #conn_data{ host = Host, port = Port, transport = Transport, connect_timeout = Timeout, connect_options = ConnectOpts, ssl_options = SslOpts0 } = Data, %% Filter out hackney-specific options that are not valid for transport TransportOpts = proplists:delete(protocols, ConnectOpts), Opts = case Transport of hackney_ssl -> %% Use ssl_opts/2 to properly merge defaults with user options %% (handles cacertfile vs cacerts correctly) MergedSslOpts = hackney_ssl:ssl_opts(Host, [{ssl_options, SslOpts0}]), AlpnOpts = hackney_ssl:alpn_opts(ConnectOpts), FinalSslOpts = hackney_util:merge_opts(MergedSslOpts, AlpnOpts), TransportOpts ++ [{ssl_options, FinalSslOpts}]; _ -> TransportOpts end, case Transport:connect(Host, Port, Opts, Timeout) of {ok, Socket} -> Protocol = case Transport of hackney_ssl -> hackney_ssl:get_negotiated_protocol(Socket); _ -> http1 end, case Protocol of http2 -> init_h2_connection(Socket, Data#conn_data{socket = Socket, protocol = http2}, From); http1 -> NewData = Data#conn_data{socket = Socket, protocol = http1}, {next_state, connected, NewData, [{reply, From, ok}]} end; {error, Reason} -> {stop_and_reply, normal, [{reply, From, {error, Reason}}]} end. %% @private Initialize HTTP/2 connection via the h2 library. %% The h2_connection process takes ownership of the socket and delivers %% owner messages ({h2, Conn, Event}) to this gen_statem's mailbox. init_h2_connection(Socket, Data, From) -> start_h2_connection(Socket, Data, From, first_connect). %% @private Initialize HTTP/2 after SSL upgrade (e.g., after CONNECT proxy tunnel) init_h2_after_upgrade(SslSocket, Data, From) -> start_h2_connection(SslSocket, Data, From, after_upgrade). start_h2_connection(Socket, Data, From, Origin) -> #conn_data{transport = Transport} = Data, case h2_connection:start_link(client, Socket, self(), #{}) of {ok, H2Conn} -> %% Transfer socket ownership then activate handshake. _ = Transport:controlling_process(Socket, H2Conn), case h2_connection:activate(H2Conn) of ok -> case h2_connection:wait_connected(H2Conn, Data#conn_data.connect_timeout) of ok -> Mon = erlang:monitor(process, H2Conn), NewData = Data#conn_data{ h2_conn = H2Conn, h2_mon = Mon, h2_streams = #{} }, %% Cancel any pending idle_timeout armed by the %% TCP-first connected(enter): HTTP/2 connections %% multiplex and stay in `connected`, so the 2s %% pool default would kill a busy conn (#836). CancelIdle = {state_timeout, infinity, idle_timeout}, case Origin of first_connect -> {next_state, connected, NewData, [CancelIdle, {reply, From, ok}]}; after_upgrade -> {keep_state, NewData, [CancelIdle, {reply, From, ok}]} end; {error, WaitErr} -> close_h2(H2Conn), h2_start_failure(Origin, From, WaitErr) end; {error, ActivateErr} -> close_h2(H2Conn), h2_start_failure(Origin, From, ActivateErr) end; {error, Reason} -> h2_start_failure(Origin, From, Reason) end. h2_start_failure(first_connect, From, Reason) -> {stop_and_reply, normal, [{reply, From, {error, Reason}}]}; h2_start_failure(after_upgrade, From, Reason) -> {keep_state_and_data, [{reply, From, {error, Reason}}]}. %% @private Close an HTTP/2 connection, tolerating an already-closed one. close_h2(H2Conn) -> try h2_connection:close(H2Conn) catch _:_ -> ok end. %% @private Send an HTTP/2 request via the h2 library. do_h2_request(From, Method, Path, Headers, Body, Data) -> do_h2_send(From, Method, Path, Headers, Body, {sync, waiting_headers}, sync, Data). %% @private Send an HTTP/2 async request. do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, _FollowRedirect, Data) -> Ref = self(), StreamState = {async, AsyncMode, StreamTo, Ref, waiting_headers}, do_h2_send(From, Method, Path, Headers, Body, StreamState, {async, Ref, StreamTo, AsyncMode}, Data). do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, Data) -> #conn_data{h2_conn = H2Conn, transport = Transport, host = Host, port = Port} = Data, MethodBin = to_binary(Method), PathBin = case Path of <<>> -> <<"/">>; _ -> to_binary(Path) end, Scheme = case Transport of hackney_ssl -> <<"https">>; _ -> <<"http">> end, Authority = build_authority(Host, Port, Transport), PseudoHeaders = [ {<<":method">>, MethodBin}, {<<":scheme">>, Scheme}, {<<":authority">>, Authority}, {<<":path">>, PathBin} ], H2Headers = PseudoHeaders ++ normalize_headers(Headers), BodyBin = case Body of B when is_binary(B) -> B; L -> iolist_to_binary(L) end, %% h2_connection can die between pool checkout and this call; gen_statem:call %% on a dead pid raises exit:noproc. Catch that and normalise to an error %% so the caller sees {error, {closed, _}} instead of a gen_statem:call %% blowing up (issue #836). SendRes = try case BodyBin of <<>> -> h2_connection:send_request_headers(H2Conn, H2Headers, true); _ -> case h2_connection:send_request_headers(H2Conn, H2Headers, false) of {ok, SId} -> case h2_connection:send_data(H2Conn, SId, BodyBin, true) of ok -> {ok, SId}; {error, _} = E1 -> E1 end; Err -> Err end end catch exit:{ExitReason, _} -> {error, {closed, ExitReason}}; exit:ExitReason -> {error, {closed, ExitReason}} end, case SendRes of {ok, StreamId} -> Streams = maps:put(StreamId, {From, StreamState}, Data#conn_data.h2_streams), NewData0 = Data#conn_data{ h2_streams = Streams, method = MethodBin, path = PathBin }, NewData = case Mode of sync -> NewData0#conn_data{request_from = From}; {async, Ref, StreamTo, AsyncMode} -> NewData0#conn_data{ async = AsyncMode, async_ref = Ref, stream_to = StreamTo } end, case Mode of sync -> {keep_state, NewData}; {async, Ref1, _, _} -> {keep_state, NewData, [{reply, From, {ok, Ref1}}]} end; {error, Reason} -> {keep_state_and_data, [{reply, From, {error, Reason}}]} end. %% @private Build authority (host:port) for HTTP/2 :authority pseudo-header build_authority(Host, Port, Transport) -> HostBin = to_binary(Host), case {Transport, Port} of {hackney_ssl, 443} -> HostBin; {hackney_tcp, 80} -> HostBin; _ -> <> end. %% @private Convert value to binary to_binary(V) when is_binary(V) -> V; to_binary(V) when is_list(V) -> list_to_binary(V); to_binary(V) when is_atom(V) -> atom_to_binary(V, utf8). %% @private Normalize headers to binary key-value pairs for HTTP/2 %% Also filters out Host header since :authority pseudo-header is used instead. %% Having both Host and :authority causes protocol_error on strict servers (e.g. Google). normalize_headers(Headers) -> lists:filtermap(fun({K, V}) -> KeyLower = hackney_bstr:to_lower(to_binary(K)), case KeyLower of <<"host">> -> false; %% Skip Host header - use :authority instead _ -> {true, {KeyLower, to_binary(V)}} end end, Headers). %% @private Handle a {h2, Conn, Event} owner message from the h2 library. handle_h2_event({informational, _StreamId, _Status, _Headers}, Data) -> %% 1xx interim responses: ignore for now (same as previous HPACK path). {keep_state, Data}; handle_h2_event({response, StreamId, Status, Headers}, Data) -> h2_on_response(StreamId, Status, Headers, Data); handle_h2_event({data, StreamId, Body, EndStream}, Data) -> h2_on_data(StreamId, Body, EndStream, Data); handle_h2_event({trailers, _StreamId, _Headers}, Data) -> {keep_state, Data}; handle_h2_event({stream_reset, StreamId, ErrorCode}, Data) -> h2_on_stream_reset(StreamId, ErrorCode, Data); handle_h2_event({goaway, _LastStreamId, ErrorCode}, Data) -> h2_on_goaway(ErrorCode, Data); handle_h2_event({closed, Reason}, Data) -> h2_on_closed(Reason, Data); handle_h2_event(_Other, Data) -> {keep_state, Data}. h2_on_response(StreamId, Status, Headers, Data) -> maybe_record_altsvc(Headers, Data), #conn_data{h2_streams = Streams} = Data, case maps:get(StreamId, Streams, undefined) of {From, {sync, waiting_headers}} -> %% Headers alone with END_STREAM arrive here when the server %% already finished the response (body sent via empty data). Streams2 = maps:put(StreamId, {From, {sync, body, Status, Headers, <<>>}}, Streams), {keep_state, Data#conn_data{h2_streams = Streams2, status = Status, response_headers = Headers}}; {StreamTo, {async, AsyncMode, StreamTo, Ref, waiting_headers}} -> StreamTo ! {hackney_response, Ref, {status, Status, <<>>}}, StreamTo ! {hackney_response, Ref, {headers, Headers}}, Streams2 = maps:put(StreamId, {StreamTo, {async, AsyncMode, StreamTo, Ref, streaming, Status, Headers}}, Streams), {keep_state, Data#conn_data{h2_streams = Streams2, status = Status, response_headers = Headers}}; _ -> {keep_state, Data} end. h2_on_data(StreamId, Body, EndStream, Data) -> #conn_data{h2_streams = Streams} = Data, case maps:get(StreamId, Streams, undefined) of {From, {sync, body, Status, Headers, Acc}} -> NewAcc = <>, case EndStream of true -> Streams2 = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h2_streams = Streams2, request_from = undefined}, [{reply, From, {ok, Status, Headers, NewAcc}}]}; false -> Streams2 = maps:put(StreamId, {From, {sync, body, Status, Headers, NewAcc}}, Streams), {keep_state, Data#conn_data{h2_streams = Streams2}} end; {StreamTo, {async, AsyncMode, StreamTo, Ref, streaming, Status, Headers}} -> _ = case byte_size(Body) of 0 -> ok; _ -> StreamTo ! {hackney_response, Ref, Body} end, case EndStream of true -> StreamTo ! {hackney_response, Ref, done}, Streams2 = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h2_streams = Streams2, async = false, async_ref = undefined, stream_to = undefined}}; false -> NewState = {async, AsyncMode, StreamTo, Ref, streaming, Status, Headers}, Streams2 = maps:put(StreamId, {StreamTo, NewState}, Streams), {keep_state, Data#conn_data{h2_streams = Streams2}} end; _ -> {keep_state, Data} end. h2_on_stream_reset(StreamId, ErrorCode, Data) -> #conn_data{h2_streams = Streams} = Data, case maps:get(StreamId, Streams, undefined) of {From, {sync, _}} -> Streams2 = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h2_streams = Streams2, request_from = undefined}, [{reply, From, {error, {stream_error, ErrorCode}}}]}; {StreamTo, {async, _, StreamTo, Ref, _, _, _}} -> StreamTo ! {hackney_response, Ref, {error, {stream_error, ErrorCode}}}, Streams2 = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h2_streams = Streams2}}; {StreamTo, {async, _, StreamTo, Ref, _}} -> StreamTo ! {hackney_response, Ref, {error, {stream_error, ErrorCode}}}, Streams2 = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h2_streams = Streams2}}; _ -> {keep_state, Data} end. h2_on_goaway(ErrorCode, Data) -> {Replies, Data1} = collect_h2_aborts({goaway, ErrorCode}, Data), {keep_state, Data1, Replies}. h2_on_closed(Reason, Data) -> {Replies, Data1} = collect_h2_aborts({closed, Reason}, Data), Stripped = Data1#conn_data{h2_conn = undefined, h2_mon = undefined, socket = undefined}, %% Transition to closed. For pooled conns, closed(enter,...) keeps the %% process alive for ?CLOSED_GRACE_MS so calls from workers that raced %% the pool checkout get a proper error reply (issue #836). {next_state, closed, Stripped, Replies}. collect_h2_aborts(Err, #conn_data{h2_streams = Streams} = Data) -> Replies = maps:fold(fun (_SId, {From, {sync, _}}, Acc) -> [{reply, From, {error, Err}} | Acc]; (_SId, {From, {sync, body, _, _, _}}, Acc) -> [{reply, From, {error, Err}} | Acc]; (_SId, {StreamTo, {async, _, StreamTo, Ref, _}}, Acc) -> StreamTo ! {hackney_response, Ref, {error, Err}}, Acc; (_SId, {StreamTo, {async, _, StreamTo, Ref, _, _, _}}, Acc) -> StreamTo ! {hackney_response, Ref, {error, Err}}, Acc; (_, _, Acc) -> Acc end, [], Streams), {Replies, Data#conn_data{h2_streams = #{}, request_from = undefined}}. %%==================================================================== %% HTTP/3 Support Functions %%==================================================================== %% @private Send an HTTP/3 request %% Opens a stream, sends headers and optionally body do_h3_request(From, Method, Path, Headers, Body, Data) -> #conn_data{ host = Host, h3_conn = ConnRef, h3_streams = Streams } = Data, MethodBin = to_binary(Method), PathBin = case Path of <<>> -> <<"/">>; _ -> to_binary(Path) end, HostBin = to_binary(Host), %% Use hackney_h3 to send request case hackney_h3:send_request(ConnRef, MethodBin, HostBin, PathBin, normalize_headers(Headers), Body) of {ok, StreamId, _NewStreams} -> %% Track stream for response UpdatedStreams = maps:put(StreamId, {From, waiting_headers}, Streams), NewData = Data#conn_data{ h3_streams = UpdatedStreams, method = MethodBin, path = PathBin, request_from = From }, %% Stay in connected state, wait for response via QUIC messages {keep_state, NewData}; {error, Reason} -> {keep_state_and_data, [{reply, From, {error, Reason}}]} end. %% @private Send an HTTP/3 async request %% Opens a stream, sends headers and body, sets up async streaming do_h3_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, Data) -> #conn_data{ host = Host, h3_conn = ConnRef, h3_streams = Streams } = Data, MethodBin = to_binary(Method), PathBin = case Path of <<>> -> <<"/">>; _ -> to_binary(Path) end, HostBin = to_binary(Host), %% Use connection PID as async ref Ref = self(), %% Use hackney_h3 to send request case hackney_h3:send_request(ConnRef, MethodBin, HostBin, PathBin, normalize_headers(Headers), Body) of {ok, StreamId, _NewStreams} -> %% Track stream with async info StreamState = {waiting_headers_async, AsyncMode, StreamTo, Ref}, UpdatedStreams = maps:put(StreamId, {From, StreamState}, Streams), NewData = Data#conn_data{ h3_streams = UpdatedStreams, method = MethodBin, path = PathBin, request_from = From, async = AsyncMode, async_ref = Ref, stream_to = StreamTo }, %% Reply immediately with {ok, Ref} for async {keep_state, NewData, [{reply, From, {ok, Ref}}]}; {error, Reason} -> {keep_state_and_data, [{reply, From, {error, Reason}}]} end. %% @private Send an HTTP/3 request with streaming body reads %% Returns {ok, Status, Headers} and allows subsequent stream_body calls do_h3_request_streaming(From, Method, Path, Headers, Body, Data) -> #conn_data{ host = Host, h3_conn = ConnRef, h3_streams = Streams } = Data, MethodBin = to_binary(Method), PathBin = case Path of <<>> -> <<"/">>; _ -> to_binary(Path) end, HostBin = to_binary(Host), %% Use hackney_h3 to send request case hackney_h3:send_request(ConnRef, MethodBin, HostBin, PathBin, normalize_headers(Headers), Body) of {ok, StreamId, _NewStreams} -> %% Track stream for streaming response StreamState = {waiting_headers_streaming, From}, UpdatedStreams = maps:put(StreamId, {From, StreamState}, Streams), NewData = Data#conn_data{ h3_streams = UpdatedStreams, method = MethodBin, path = PathBin, request_from = From }, %% Stay in connected state, wait for headers {keep_state, NewData}; {error, Reason} -> {keep_state_and_data, [{reply, From, {error, Reason}}]} end. %% @private Send HTTP/3 request headers only (for streaming body mode) %% Returns ok and transitions to streaming_body state do_h3_send_headers(From, Method, Path, Headers, Data) -> #conn_data{ host = Host, h3_conn = ConnRef, h3_streams = Streams } = Data, MethodBin = to_binary(Method), PathBin = case Path of <<>> -> <<"/">>; _ -> to_binary(Path) end, HostBin = to_binary(Host), %% Use hackney_h3 to send headers only (Fin=false) case hackney_h3:send_request_headers(ConnRef, MethodBin, HostBin, PathBin, normalize_headers(Headers)) of {ok, StreamId, _NewStreams} -> %% Track stream for response (will wait after body is sent) StreamState = {sending_body, From}, UpdatedStreams = maps:put(StreamId, {From, StreamState}, Streams), NewData = Data#conn_data{ h3_streams = UpdatedStreams, h3_stream_id = StreamId, method = MethodBin, path = PathBin, request_from = undefined }, %% Transition to streaming_body state {next_state, streaming_body, NewData, [{reply, From, ok}]}; {error, Reason} -> {keep_state_and_data, [{reply, From, {error, Reason}}]} end. %% @private Handle HTTP/3 stream_body call %% Returns buffered chunk, waits for data, or returns done handle_h3_stream_body(From, Streams, Data) -> %% Find the streaming stream case find_streaming_stream(Streams) of {ok, StreamId, {streaming_body, Status, Headers, Buffer, undefined}} -> %% No pending caller, check buffer case Buffer of <<>> -> %% No data buffered - wait for QUIC data NewStreamState = {streaming_body, Status, Headers, <<>>, From}, UpdatedStreams = maps:put(StreamId, {undefined, NewStreamState}, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}}; _ -> %% Return buffered data NewStreamState = {streaming_body, Status, Headers, <<>>, undefined}, UpdatedStreams = maps:put(StreamId, {undefined, NewStreamState}, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}, [{reply, From, {ok, Buffer}}]} end; {ok, StreamId, {streaming_body_done, _Status, _Headers}} -> %% Stream complete, no more data UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}, [{reply, From, done}]}; {ok, StreamId, {streaming_body_final, _Status, _Headers, FinalBuffer}} -> %% Stream complete with final chunk case FinalBuffer of <<>> -> UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}, [{reply, From, done}]}; _ -> %% Return final chunk, next call gets done NewStreamState = {streaming_body_done, _Status, _Headers}, UpdatedStreams = maps:put(StreamId, {undefined, NewStreamState}, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}, [{reply, From, {ok, FinalBuffer}}]} end; {ok, _StreamId, _OtherState} -> %% Stream in wrong state {keep_state_and_data, [{reply, From, {error, not_streaming}}]}; none -> %% No streaming stream found {keep_state_and_data, [{reply, From, {error, no_stream}}]} end. %% @private Find a stream in streaming mode find_streaming_stream(Streams) -> Result = maps:fold(fun (StreamId, {_, {streaming_body, _, _, _, _} = State}, none) -> {ok, StreamId, State}; (StreamId, {_, {streaming_body_done, _, _} = State}, none) -> {ok, StreamId, State}; (StreamId, {_, {streaming_body_final, _, _, _} = State}, none) -> {ok, StreamId, State}; (_, _, Acc) -> Acc end, none, Streams), Result. %% @private Handle HTTP/3 response headers handle_h3_headers(StreamId, Headers, Fin, Streams, Data) -> case hackney_h3:parse_response_headers(Headers) of {ok, Status, RespHeaders} -> %% RespHeaders from hackney_h3:parse_response_headers is already a list HeadersList = RespHeaders, maybe_record_altsvc(HeadersList, Data), case maps:get(StreamId, Streams, undefined) of {From, waiting_headers} -> case Fin of true -> %% Headers with Fin=true means no body (redirect, 204, 304, HEAD response) %% Reply immediately with empty body UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{ h3_streams = UpdatedStreams, status = Status, response_headers = RespHeaders, request_from = undefined }, [{reply, From, {ok, Status, HeadersList, <<>>}}]}; false -> %% Sync mode - update stream state to receiving body UpdatedStreams = maps:put(StreamId, {From, {receiving_body, Status, RespHeaders, <<>>}}, Streams), {keep_state, Data#conn_data{ h3_streams = UpdatedStreams, status = Status, response_headers = RespHeaders }} end; {_From, {waiting_headers_async, AsyncMode, StreamTo, Ref}} -> %% Async mode - send status and headers to StreamTo StreamTo ! {hackney_response, Ref, {status, Status, <<"">>}}, StreamTo ! {hackney_response, Ref, {headers, HeadersList}}, %% Update stream state for async body streaming NewStreamState = {streaming_body_async, AsyncMode, StreamTo, Ref, Status, RespHeaders}, UpdatedStreams = maps:put(StreamId, {undefined, NewStreamState}, Streams), {keep_state, Data#conn_data{ h3_streams = UpdatedStreams, status = Status, response_headers = RespHeaders }}; {From, {waiting_headers_streaming, From}} -> %% Streaming mode - reply with headers, then allow stream_body calls NewStreamState = {streaming_body, Status, RespHeaders, <<>>, undefined}, UpdatedStreams = maps:put(StreamId, {undefined, NewStreamState}, Streams), {keep_state, Data#conn_data{ h3_streams = UpdatedStreams, status = Status, response_headers = RespHeaders }, [{reply, From, {ok, Status, HeadersList}}]}; {From, {sending_body, From}} -> %% Response arrived while sending body - update state %% If request_from is set (from start_response), reply to it case Data#conn_data.request_from of undefined -> %% No one waiting yet - just update state UpdatedStreams = maps:put(StreamId, {From, {receiving_body, Status, RespHeaders, <<>>}}, Streams), {keep_state, Data#conn_data{ h3_streams = UpdatedStreams, status = Status, response_headers = RespHeaders }}; WaitingFrom -> %% Someone called start_response - reply with status/headers UpdatedStreams = maps:put(StreamId, {WaitingFrom, {receiving_body, Status, RespHeaders, <<>>}}, Streams), {keep_state, Data#conn_data{ h3_streams = UpdatedStreams, status = Status, response_headers = RespHeaders, request_from = undefined }, [{reply, WaitingFrom, {ok, Status, HeadersList, self()}}]} end; _ -> %% Unknown stream or wrong state {keep_state, Data} end; {error, Reason} -> %% Failed to parse headers case maps:get(StreamId, Streams, undefined) of {From, _State} when From =/= undefined -> UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}, [{reply, From, {error, Reason}}]}; {_, {waiting_headers_async, _AsyncMode, StreamTo, Ref}} -> %% Async error StreamTo ! {hackney_response, Ref, {error, Reason}}, UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}}; _ -> {keep_state, Data} end end. %% @private Handle HTTP/3 response body data handle_h3_data(StreamId, RecvData, Fin, Streams, Data) -> case maps:get(StreamId, Streams, undefined) of {From, {receiving_body, Status, Headers, AccBody}} -> %% Sync mode - buffer body NewBody = <>, case Fin of true -> %% Stream complete - reply with full response UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams, request_from = undefined}, [{reply, From, {ok, Status, Headers, NewBody}}]}; false -> %% More data expected UpdatedStreams = maps:put(StreamId, {From, {receiving_body, Status, Headers, NewBody}}, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}} end; {_, {streaming_body_async, _AsyncMode, StreamTo, Ref, Status, Headers}} -> %% Async mode - send chunk to StreamTo _ = case byte_size(RecvData) > 0 of true -> StreamTo ! {hackney_response, Ref, RecvData}; false -> ok end, case Fin of true -> %% Stream complete - send done StreamTo ! {hackney_response, Ref, done}, UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{ h3_streams = UpdatedStreams, request_from = undefined, async = false, async_ref = undefined, stream_to = undefined }}; false -> %% More data expected - keep streaming NewStreamState = {streaming_body_async, _AsyncMode, StreamTo, Ref, Status, Headers}, UpdatedStreams = maps:put(StreamId, {undefined, NewStreamState}, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}} end; {_, {streaming_body, Status, Headers, Buffer, PendingFrom}} -> %% Pull-based streaming mode case Fin of true -> %% Stream complete case PendingFrom of undefined when Buffer =:= <<>>, RecvData =:= <<>> -> %% No pending caller, no data - mark done NewStreamState = {streaming_body_done, Status, Headers}, UpdatedStreams = maps:put(StreamId, {undefined, NewStreamState}, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}}; undefined -> %% No pending caller, has data - store final chunk for next stream_body call NewBuffer = <>, FinalStreams = maps:put(StreamId, {undefined, {streaming_body_final, Status, Headers, NewBuffer}}, Streams), {keep_state, Data#conn_data{h3_streams = FinalStreams}}; _ -> %% Has pending caller - reply with last chunk or done NewBuffer = <>, case NewBuffer of <<>> -> %% No data - reply done UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}, [{reply, PendingFrom, done}]}; _ -> %% Has data - reply with it, next call gets done NewStreamState = {streaming_body_done, Status, Headers}, UpdatedStreams = maps:put(StreamId, {undefined, NewStreamState}, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}, [{reply, PendingFrom, {ok, NewBuffer}}]} end end; false -> %% More data expected case PendingFrom of undefined -> %% No pending caller - buffer data NewBuffer = <>, NewStreamState = {streaming_body, Status, Headers, NewBuffer, undefined}, UpdatedStreams = maps:put(StreamId, {undefined, NewStreamState}, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}}; _ -> %% Has pending caller - reply with data NewBuffer = <>, NewStreamState = {streaming_body, Status, Headers, <<>>, undefined}, UpdatedStreams = maps:put(StreamId, {undefined, NewStreamState}, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}, [{reply, PendingFrom, {ok, NewBuffer}}]} end end; {_, {streaming_body_final, Status, Headers, FinalBuffer}} -> %% Already got final data, just buffer more (shouldn't happen) NewBuffer = <>, UpdatedStreams = maps:put(StreamId, {undefined, {streaming_body_final, Status, Headers, NewBuffer}}, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams}}; {From, waiting_headers} -> %% Got data before headers - unusual but handle it {keep_state, Data#conn_data{h3_streams = maps:put(StreamId, {From, {receiving_body, 0, [], RecvData}}, Streams)}}; _ -> %% Unknown stream {keep_state, Data} end. %% @private Handle HTTP/3 stream reset handle_h3_stream_reset(StreamId, ErrorCode, Streams, Data) -> case maps:get(StreamId, Streams, undefined) of {From, _State} when is_pid(From) -> %% Sync stream - reply with error UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams, request_from = undefined}, [{reply, From, {error, {stream_reset, ErrorCode}}}]}; {_, {streaming_body_async, _AsyncMode, StreamTo, Ref, _Status, _Headers}} -> %% Async stream - send error to StreamTo StreamTo ! {hackney_response, Ref, {error, {stream_reset, ErrorCode}}}, UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams, request_from = undefined}}; {_, {waiting_headers_async, _AsyncMode, StreamTo, Ref}} -> %% Async stream waiting for headers StreamTo ! {hackney_response, Ref, {error, {stream_reset, ErrorCode}}}, UpdatedStreams = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h3_streams = UpdatedStreams, request_from = undefined}}; _ -> {keep_state, Data} end. %% @private Cache the H3 session ticket in the pool (best effort, guarded so a %% custom pool handler without the callback degrades to no caching). maybe_store_h3_session(_Ticket, #conn_data{pool_handler = undefined}) -> ok; maybe_store_h3_session(Ticket, #conn_data{pool_handler = PoolHandler, host = Host, port = Port, transport = Transport, pool_name = PoolName}) -> case erlang:function_exported(PoolHandler, store_h3_session, 5) of true -> try PoolHandler:store_h3_session(Host, Port, Transport, Ticket, [{pool, PoolName}]) catch _:_ -> ok end, ok; false -> ok end. %% @private Drop a cached H3 session ticket (e.g. after 0-RTT rejection). maybe_delete_h3_session(#conn_data{pool_handler = undefined}) -> ok; maybe_delete_h3_session(#conn_data{pool_handler = PoolHandler, host = Host, port = Port, transport = Transport, pool_name = PoolName}) -> case erlang:function_exported(PoolHandler, delete_h3_session, 4) of true -> try PoolHandler:delete_h3_session(Host, Port, Transport, [{pool, PoolName}]) catch _:_ -> ok end, ok; false -> ok end. %% @private Fail any in-flight stream whose 0-RTT data the server rejected. %% Returns the updated stream map plus gen_statem reply actions for sync %% callers; async callers are notified via their StreamTo mailbox directly. fail_rejected_h3_streams(StreamIds, Streams) -> Ids = case is_list(StreamIds) of true -> StreamIds; false -> sets:to_list(StreamIds) end, lists:foldl(fun(StreamId, {AccStreams, AccActions}) -> fail_rejected_h3_stream(StreamId, AccStreams, AccActions) end, {Streams, []}, Ids). fail_rejected_h3_stream(StreamId, Streams, Actions) -> Err = {error, early_data_rejected}, case maps:get(StreamId, Streams, undefined) of undefined -> {Streams, Actions}; {From, waiting_headers} -> %% Sync request awaiting headers - reply via the state machine. {maps:remove(StreamId, Streams), [{reply, From, Err} | Actions]}; {From, {sending_body, _}} -> {maps:remove(StreamId, Streams), [{reply, From, Err} | Actions]}; {From, {waiting_headers_streaming, _}} -> {maps:remove(StreamId, Streams), [{reply, From, Err} | Actions]}; {_, {streaming_body_async, _AsyncMode, StreamTo, Ref, _Status, _Headers}} -> StreamTo ! {hackney_response, Ref, Err}, {maps:remove(StreamId, Streams), Actions}; {_, {waiting_headers_async, _AsyncMode, StreamTo, Ref}} -> StreamTo ! {hackney_response, Ref, Err}, {maps:remove(StreamId, Streams), Actions}; {_, {async, _AsyncMode, StreamTo, Ref, _SubState}} -> StreamTo ! {hackney_response, Ref, Err}, {maps:remove(StreamId, Streams), Actions}; _ -> {maps:remove(StreamId, Streams), Actions} end. %% @private Handle HTTP/3 connection closed handle_h3_conn_closed(Reason, Data) -> handle_h3_termination({connection_closed, Reason}, Data). %% @private Handle HTTP/3 transport error handle_h3_error(Error, Data) -> handle_h3_termination(Error, Data). %% @private Shared helper for HTTP/3 connection termination. %% Notifies all pending streams and transitions to closed state. handle_h3_termination(Error, Data) -> #conn_data{h3_streams = Streams, request_from = From} = Data, %% Notify all pending streams (sync and async) Actions = maps:fold(fun(_StreamId, StreamState, Acc) -> case StreamState of {StreamFrom, _State} when is_pid(StreamFrom) -> %% Sync stream [{reply, StreamFrom, {error, Error}} | Acc]; {_, {streaming_body_async, _AsyncMode, StreamTo, Ref, _Status, _Headers}} -> %% Async stream - send error to StreamTo (not gen_statem action) StreamTo ! {hackney_response, Ref, {error, Error}}, Acc; {_, {waiting_headers_async, _AsyncMode, StreamTo, Ref}} -> %% Async stream waiting for headers StreamTo ! {hackney_response, Ref, {error, Error}}, Acc; _ -> Acc end end, [], Streams), %% Clear connection state NewData = Data#conn_data{ h3_conn = undefined, h3_streams = #{}, request_from = undefined, async = false, async_ref = undefined, stream_to = undefined }, case From of undefined -> {next_state, closed, NewData, Actions}; _ -> %% Don't duplicate reply if From is already in Actions case lists:any(fun({reply, F, _}) -> F =:= From end, Actions) of true -> {next_state, closed, NewData, Actions}; false -> {next_state, closed, NewData, [{reply, From, {error, Error}} | Actions]} end end. %% @private Hand response headers to the Alt-Svc cache so server-advertised %% HTTP/3 endpoints get recorded for future requests. Fires for every %% protocol so the cache TTL stays fresh while h3 is in use and `clear' %% directives are honored even on h3 responses. maybe_record_altsvc(Headers, #conn_data{host = Host, port = Port}) when is_list(Headers) -> _ = (try hackney_altsvc:parse_and_cache(Host, Port, Headers) catch _:_ -> ok end), ok; maybe_record_altsvc(_Headers, _Data) -> ok.