%%% -*- erlang -*- %%% %%% This file is part of hackney released under the Apache 2 license. %%% See the NOTICE for more information. %%% %%% Copyright (c) 2026 Benoît Chesneau %%% %%% @doc gen_statem process for WebTransport client connections. %%% %%% This module mirrors {@link hackney_ws} so an application can move from %%% WebSocket to WebTransport by swapping the `ws_' API prefix for `wt_'. %%% It wraps an `erlang-webtransport' client session (HTTP/3 by default, %%% HTTP/2 optional) and owns that session process. %%% %%% == Connection reuse and multiplexing == %%% %%% A WebTransport session is the analog of an HTTP/2 connection: one %%% `hackney_wt' process owns a single session and many streams are %%% multiplexed over it. `open_stream/2' opens as many bidirectional or %%% unidirectional streams as the peer's flow control allows; each stream %%% has its own send (`stream_send/3,4') and receive (`stream_recv/2,3') %%% channel, keyed by stream id, exactly like the `h2_streams' map in %%% `hackney_conn'. %%% %%% == Default message channel == %%% %%% WebTransport has no message framing of its own, so to stay %%% interoperable with any server we do NOT invent a wire format. The %%% `send/2' / `recv/1' convenience channel maps onto a single persistent %%% bidirectional stream opened at connect time: `send/2' writes bytes to %%% it and `recv/1' returns the next chunk received on it as %%% `{binary, Data}'. Datagrams and data from server-opened streams are %%% also surfaced on this channel as `{datagram, Data}' and %%% `{stream, Id, Data}'. Because there is no framing, chunks are not %%% guaranteed to align with send boundaries; an application that needs %%% message boundaries must self-delimit. %%% %%% == Delivery modes == %%% %%% In passive mode (the default) data is buffered per channel and read %%% with `recv'/`stream_recv'. In active mode every event is forwarded to %%% the owner uniformly tagged with its stream id: %%% `{hackney_wt, Conn, {binary, Data}}', `{hackney_wt, Conn, {datagram, %%% Data}}', `{hackney_wt, Conn, {stream, Id, Data}}', `{hackney_wt, Conn, %%% {stream_fin, Id, Data}}', `{hackney_wt, Conn, {stream_closed, Id, %%% Reason}}' and `{hackney_wt, Conn, closed}'. %%% %%% States: %%% -module(hackney_wt). -behaviour(gen_statem). %% API -export([ start_link/1, connect/1, connect/2, send/2, recv/1, recv/2, setopts/2, close/1, close/2, controlling_process/2, open_stream/2, stream_send/3, stream_send/4, stream_recv/2, stream_recv/3, close_stream/2, reset_stream/3, stop_sending/3, send_datagram/2, session_info/1 ]). %% gen_statem callbacks -export([ init/1, callback_mode/0, terminate/3, code_change/4 ]). %% State functions -export([ idle/3, connected/3, closed/3 ]). -define(CONNECT_TIMEOUT, 8000). -define(RECV_TIMEOUT, infinity). %% Bound the bytes buffered across all passive receive queues so a hostile %% server cannot drive the client to OOM by flooding stream/datagram data %% that is never consumed. `infinity' disables the cap. This complements %% WebTransport flow control (which bounds in-flight bytes per stream) by %% bounding the already-decoded queues; same intent as the WebSocket %% client's GHSA-q8jg caps. -define(DEFAULT_MAX_RECV_BUFFER, 16#4000000). %% 64 MiB %% A normalized message on the primary channel, handed to recv/2 or, in %% active mode, to the owner. -type wt_msg() :: {binary, binary()} | {datagram, binary()} | {stream, non_neg_integer(), binary()} | {stream_fin, non_neg_integer(), binary()} | {stream_closed, non_neg_integer(), term()}. %% Accepted by send/2. -type wt_frame() :: {text, iodata()} | {binary, iodata()} | iodata() | {datagram, iodata()} | {stream, non_neg_integer(), iodata()} | {stream, non_neg_integer(), iodata(), fin | nofin}. %% Returned by stream_recv/2,3. -type stream_msg() :: {ok, binary()} | {ok, {fin, binary()}} | {error, term()}. -export_type([wt_msg/0, wt_frame/0, stream_msg/0]). %% Per-stream receive state for a client-opened stream (mirrors the role %% of an HTTP/2 stream's pending caller in hackney_conn). -record(stream, { q = queue:new() :: queue:queue(), from :: {pid(), reference()} | undefined, %% undefined while open; set when the peer ends the stream closed :: undefined | term() }). -record(wt_data, { %% Connection owner (linked via start_link, trap_exit handles death) owner :: pid(), %% Connection identity host :: string() | binary(), port :: inet:port_number(), transport = h3 :: h2 | h3, path :: binary(), %% webtransport:connect/4 options map (TLS, headers, compat_mode, ...) connect_opts = #{} :: map(), connect_timeout = ?CONNECT_TIMEOUT :: timeout(), recv_timeout = ?RECV_TIMEOUT :: timeout(), %% Delivery mode active = false :: false | true | once, %% Underlying webtransport session and the persistent default stream session :: pid() | undefined, default_stream :: non_neg_integer() | undefined, %% Per-stream state for client-opened streams: StreamId => #stream{} streams = #{} :: #{non_neg_integer() => #stream{}}, %% Primary channel buffering (default stream + datagrams + server streams) recv_q = queue:new() :: queue:queue(), recv_from :: {pid(), reference()} | undefined, %% Total buffered bytes across every queue, for the OOM cap recv_bytes = 0 :: non_neg_integer(), max_recv_buffer = ?DEFAULT_MAX_RECV_BUFFER :: non_neg_integer() | infinity, %% Reason recorded once the session ends closed_reason :: term() }). %%==================================================================== %% API %%==================================================================== %% @doc Start a WebTransport connection process. %% Options: %% -spec start_link(map()) -> {ok, pid()} | {error, term()}. start_link(Opts) when is_map(Opts) -> gen_statem:start_link(?MODULE, [self(), Opts], []). %% @doc Establish the WebTransport session. Blocks until the CONNECT %% completes or fails. -spec connect(pid()) -> ok | {error, term()}. connect(Pid) -> connect(Pid, ?CONNECT_TIMEOUT). -spec connect(pid(), timeout()) -> ok | {error, term()}. connect(Pid, Timeout) -> %% The handshake is driven inside the process; the internal timeout %% below bounds it, so the call itself waits indefinitely. gen_statem:call(Pid, {connect, Timeout}, infinity). %% @doc Send on the connection. %% Frame forms: %% -spec send(pid(), wt_frame()) -> ok | {error, term()}. send(Pid, Frame) -> gen_statem:call(Pid, {send, Frame}). %% @doc Receive the next message on the primary channel (passive mode only). -spec recv(pid()) -> {ok, wt_msg()} | {error, term()}. recv(Pid) -> gen_statem:call(Pid, {recv, default}, infinity). -spec recv(pid(), timeout()) -> {ok, wt_msg()} | {error, term()}. recv(Pid, Timeout) -> gen_statem:call(Pid, {recv, Timeout}, infinity). %% @doc Set options. Supported: [{active, true|false|once}] -spec setopts(pid(), list()) -> ok | {error, term()}. setopts(Pid, Opts) -> gen_statem:call(Pid, {setopts, Opts}). %% @doc Close the session gracefully (error code 0, no reason). -spec close(pid()) -> ok. close(Pid) -> close(Pid, {0, <<>>}). -spec close(pid(), {non_neg_integer(), binary()}) -> ok. close(Pid, {Code, Reason}) -> gen_statem:cast(Pid, {close, Code, Reason}). %% @doc Assign a new controlling process. -spec controlling_process(pid(), pid()) -> ok | {error, term()}. controlling_process(Pid, NewOwner) -> gen_statem:call(Pid, {controlling_process, NewOwner}). %% @doc Open a new stream multiplexed over the session. Returns its id. -spec open_stream(pid(), bidi | uni) -> {ok, non_neg_integer()} | {error, term()}. open_stream(Pid, Type) when Type =:= bidi; Type =:= uni -> gen_statem:call(Pid, {open_stream, Type}). %% @doc Write to a stream (no FIN). -spec stream_send(pid(), non_neg_integer(), iodata()) -> ok | {error, term()}. stream_send(Pid, StreamId, Data) -> gen_statem:call(Pid, {stream_send, StreamId, Data, nofin}). %% @doc Write to a stream, optionally closing the write side (FIN). -spec stream_send(pid(), non_neg_integer(), iodata(), fin | nofin) -> ok | {error, term()}. stream_send(Pid, StreamId, Data, Fin) when Fin =:= fin; Fin =:= nofin -> gen_statem:call(Pid, {stream_send, StreamId, Data, Fin}). %% @doc Receive the next chunk on a client-opened stream (passive mode). %% Returns `{ok, Data}', `{ok, {fin, Data}}' when the peer ends the %% stream, or `{error, Reason}'. -spec stream_recv(pid(), non_neg_integer()) -> stream_msg(). stream_recv(Pid, StreamId) -> gen_statem:call(Pid, {stream_recv, StreamId, default}, infinity). -spec stream_recv(pid(), non_neg_integer(), timeout()) -> stream_msg(). stream_recv(Pid, StreamId, Timeout) -> gen_statem:call(Pid, {stream_recv, StreamId, Timeout}, infinity). %% @doc Close a stream gracefully (send FIN). -spec close_stream(pid(), non_neg_integer()) -> ok | {error, term()}. close_stream(Pid, StreamId) -> gen_statem:call(Pid, {close_stream, StreamId}). %% @doc Abruptly terminate a stream with an error code. -spec reset_stream(pid(), non_neg_integer(), non_neg_integer()) -> ok | {error, term()}. reset_stream(Pid, StreamId, ErrorCode) -> gen_statem:call(Pid, {reset_stream, StreamId, ErrorCode}). %% @doc Ask the peer to stop sending on a stream. -spec stop_sending(pid(), non_neg_integer(), non_neg_integer()) -> ok | {error, term()}. stop_sending(Pid, StreamId, ErrorCode) -> gen_statem:call(Pid, {stop_sending, StreamId, ErrorCode}). %% @doc Send an unreliable datagram. -spec send_datagram(pid(), iodata()) -> ok | {error, term()}. send_datagram(Pid, Data) -> gen_statem:call(Pid, {send_datagram, Data}). %% @doc Return session information (transport, stream count, flow control). -spec session_info(pid()) -> {ok, map()} | {error, term()}. session_info(Pid) -> gen_statem:call(Pid, session_info). %%==================================================================== %% gen_statem callbacks %%==================================================================== %% @private callback_mode() -> [state_functions, state_enter]. %% @private init([Owner, Opts]) -> process_flag(trap_exit, true), Data = #wt_data{ owner = Owner, host = maps:get(host, Opts), port = maps:get(port, Opts), transport = maps:get(transport, Opts, h3), path = maps:get(path, Opts, <<"/">>), connect_opts = maps:get(connect_opts, Opts, #{}), connect_timeout = maps:get(connect_timeout, Opts, ?CONNECT_TIMEOUT), recv_timeout = maps:get(recv_timeout, Opts, ?RECV_TIMEOUT), active = maps:get(active, Opts, false), max_recv_buffer = maps:get(max_recv_buffer, Opts, ?DEFAULT_MAX_RECV_BUFFER) }, {ok, idle, Data}. %% @private terminate(_Reason, _State, #wt_data{session = undefined}) -> ok; terminate(_Reason, _State, #wt_data{} = Data) -> close_session_safe(Data), ok. %% @private code_change(_OldVsn, State, Data, _Extra) -> {ok, State, Data}. %%==================================================================== %% State: idle %%==================================================================== idle(enter, _OldState, _Data) -> keep_state_and_data; idle({call, From}, {connect, Timeout}, Data) -> #wt_data{host = Host, port = Port, path = Path, transport = Transport, connect_opts = COpts0} = Data, COpts = COpts0#{transport => Transport, timeout => Timeout}, case webtransport:connect(Host, Port, Path, COpts) of {ok, Session} -> %% Open the persistent default bidi stream eagerly; if the %% server has not granted bidi credit yet, fall back to opening %% it lazily on the first send. DefaultStream = case webtransport:open_stream(Session, bidi) of {ok, Sid} -> Sid; {error, _} -> undefined end, Data1 = Data#wt_data{session = Session, default_stream = DefaultStream}, {next_state, connected, Data1, [{reply, From, ok}]}; {error, Reason} -> {stop_and_reply, normal, [{reply, From, {error, Reason}}]} end; idle({call, From}, _Request, _Data) -> {keep_state_and_data, [{reply, From, {error, not_connected}}]}; idle(info, {'EXIT', Owner, _Reason}, #wt_data{owner = Owner}) -> {stop, normal}; idle(_, _, _) -> keep_state_and_data. %%==================================================================== %% State: connected %%==================================================================== connected(enter, _OldState, _Data) -> keep_state_and_data; %% --- send ------------------------------------------------------------- connected({call, From}, {send, Frame}, Data) -> {Reply, Data1} = do_send(Frame, Data), {keep_state, Data1, [{reply, From, Reply}]}; %% --- primary channel recv -------------------------------------------- connected({call, From}, {recv, _Timeout}, #wt_data{active = Active}) when Active =/= false -> {keep_state_and_data, [{reply, From, {error, {active_mode, Active}}}]}; connected({call, From}, {recv, _Timeout}, #wt_data{recv_from = RF}) when RF =/= undefined -> {keep_state_and_data, [{reply, From, {error, recv_busy}}]}; connected({call, From}, {recv, Timeout0}, Data) -> Timeout = recv_timeout(Timeout0, Data), case dequeue(Data#wt_data.recv_q, Data) of {value, Msg, Q1, Data1} -> {keep_state, Data1#wt_data{recv_q = Q1}, [{reply, From, {ok, Msg}}]}; {empty, _} -> case Data#wt_data.closed_reason of undefined -> {keep_state, Data#wt_data{recv_from = From}, recv_timer(recv, Timeout)}; Reason -> {keep_state_and_data, [{reply, From, {error, Reason}}]} end end; %% --- per-stream recv -------------------------------------------------- connected({call, From}, {stream_recv, _Sid, _Timeout}, #wt_data{active = Active}) when Active =/= false -> {keep_state_and_data, [{reply, From, {error, {active_mode, Active}}}]}; connected({call, From}, {stream_recv, Sid, Timeout0}, Data) -> case maps:find(Sid, Data#wt_data.streams) of error -> {keep_state_and_data, [{reply, From, {error, unknown_stream}}]}; {ok, #stream{from = F}} when F =/= undefined -> {keep_state_and_data, [{reply, From, {error, recv_busy}}]}; {ok, S} -> do_stream_recv(From, Sid, S, recv_timeout(Timeout0, Data), Data) end; %% --- options ---------------------------------------------------------- connected({call, From}, {setopts, Opts}, Data) -> case proplists:get_value(active, Opts) of undefined -> {keep_state_and_data, [{reply, From, ok}]}; NewActive when NewActive =:= true; NewActive =:= false; NewActive =:= once -> Data1 = apply_active(NewActive, Data), {keep_state, Data1, [{reply, From, ok}]}; _ -> {keep_state_and_data, [{reply, From, {error, badarg}}]} end; connected({call, From}, {controlling_process, NewOwner}, #wt_data{owner = OldOwner} = Data) -> unlink(OldOwner), link(NewOwner), {keep_state, Data#wt_data{owner = NewOwner}, [{reply, From, ok}]}; %% --- native stream / datagram API ------------------------------------ connected({call, From}, {open_stream, Type}, #wt_data{session = S} = Data) -> case webtransport:open_stream(S, Type) of {ok, Sid} = Ok -> Streams = maps:put(Sid, #stream{}, Data#wt_data.streams), {keep_state, Data#wt_data{streams = Streams}, [{reply, From, Ok}]}; {error, _} = Err -> {keep_state_and_data, [{reply, From, Err}]} end; connected({call, From}, {stream_send, StreamId, SData, Fin}, #wt_data{session = S}) -> {keep_state_and_data, [{reply, From, webtransport:send(S, StreamId, SData, Fin)}]}; connected({call, From}, {close_stream, StreamId}, #wt_data{session = S}) -> {keep_state_and_data, [{reply, From, webtransport:close_stream(S, StreamId)}]}; connected({call, From}, {reset_stream, StreamId, Code}, #wt_data{session = S}) -> {keep_state_and_data, [{reply, From, webtransport:reset_stream(S, StreamId, Code)}]}; connected({call, From}, {stop_sending, StreamId, Code}, #wt_data{session = S}) -> {keep_state_and_data, [{reply, From, webtransport:stop_sending(S, StreamId, Code)}]}; connected({call, From}, {send_datagram, SData}, #wt_data{session = S}) -> {keep_state_and_data, [{reply, From, webtransport:send_datagram(S, SData)}]}; connected({call, From}, session_info, #wt_data{session = S}) -> {keep_state_and_data, [{reply, From, webtransport:session_info(S)}]}; connected({call, From}, _Request, _Data) -> {keep_state_and_data, [{reply, From, {error, badrequest}}]}; %% --- close ------------------------------------------------------------ connected(cast, {close, Code, Reason}, #wt_data{session = S} = Data) -> _ = webtransport:close_session(S, Code, Reason), {next_state, closed, Data#wt_data{closed_reason = closed}}; %% --- recv timeouts ---------------------------------------------------- connected({timeout, recv}, recv, #wt_data{recv_from = From} = Data) when From =/= undefined -> {keep_state, Data#wt_data{recv_from = undefined}, [{reply, From, {error, timeout}}]}; connected({timeout, recv}, recv, _Data) -> keep_state_and_data; connected({timeout, {srecv, Sid}}, {srecv, Sid}, Data) -> case maps:find(Sid, Data#wt_data.streams) of {ok, #stream{from = From} = S} when From =/= undefined -> Streams = maps:put(Sid, S#stream{from = undefined}, Data#wt_data.streams), {keep_state, Data#wt_data{streams = Streams}, [{reply, From, {error, timeout}}]}; _ -> keep_state_and_data end; %% --- session events --------------------------------------------------- connected(info, {webtransport, Session, Event}, #wt_data{session = Session} = Data) -> handle_wt_event(Event, Data); connected(info, {'EXIT', Session, Reason}, #wt_data{session = Session} = Data) -> R = case Reason of normal -> closed; _ -> {session_down, Reason} end, session_ended(R, Data); connected(info, {'EXIT', Owner, _Reason}, #wt_data{owner = Owner} = Data) -> close_session_safe(Data), {stop, normal}; connected(_, _, _) -> keep_state_and_data. %%==================================================================== %% State: closed %%==================================================================== closed(enter, _OldState, _Data) -> keep_state_and_data; %% Allow draining whatever was buffered before the session ended. closed({call, From}, {recv, _Timeout}, Data) -> case dequeue(Data#wt_data.recv_q, Data) of {value, Msg, Q1, Data1} -> {keep_state, Data1#wt_data{recv_q = Q1}, [{reply, From, {ok, Msg}}]}; {empty, _} -> {keep_state_and_data, [{reply, From, {error, closed_reason(Data)}}]} end; closed({call, From}, {stream_recv, Sid, _Timeout}, Data) -> case maps:find(Sid, Data#wt_data.streams) of {ok, S} -> case queue:out(S#stream.q) of {{value, Item}, Q1} -> Data1 = sub_bytes(item_size(Item), Data), Streams = maps:put(Sid, S#stream{q = Q1}, Data1#wt_data.streams), {keep_state, Data1#wt_data{streams = Streams}, [{reply, From, stream_item_result(Item)}]}; {empty, _} -> {keep_state_and_data, [{reply, From, {error, closed_reason(Data)}}]} end; error -> {keep_state_and_data, [{reply, From, {error, closed}}]} end; closed({call, From}, {controlling_process, NewOwner}, #wt_data{owner = OldOwner} = Data) -> unlink(OldOwner), link(NewOwner), {keep_state, Data#wt_data{owner = NewOwner}, [{reply, From, ok}]}; closed({call, From}, _Request, _Data) -> {keep_state_and_data, [{reply, From, {error, closed}}]}; closed(cast, {close, _Code, _Reason}, _Data) -> keep_state_and_data; closed(info, {'EXIT', Owner, _Reason}, #wt_data{owner = Owner}) -> {stop, normal}; closed(_, _, _) -> keep_state_and_data. %%==================================================================== %% Internal: send %%==================================================================== %% @private Map a send frame onto the webtransport session. do_send({text, D}, Data) -> send_default(D, Data); do_send({binary, D}, Data) -> send_default(D, Data); do_send({datagram, D}, #wt_data{session = S} = Data) -> {webtransport:send_datagram(S, D), Data}; do_send({stream, StreamId, D}, #wt_data{session = S} = Data) -> {webtransport:send(S, StreamId, D, nofin), Data}; do_send({stream, StreamId, D, Fin}, #wt_data{session = S} = Data) when Fin =:= fin; Fin =:= nofin -> {webtransport:send(S, StreamId, D, Fin), Data}; do_send(Ping, Data) when Ping =:= ping; Ping =:= pong -> {{error, {unsupported_frame, Ping}}, Data}; do_send({ping, _}, Data) -> {{error, {unsupported_frame, ping}}, Data}; do_send({pong, _}, Data) -> {{error, {unsupported_frame, pong}}, Data}; do_send(D, Data) when is_binary(D); is_list(D) -> send_default(D, Data); do_send(_Other, Data) -> {{error, badarg}, Data}. %% @private Write to the persistent default stream, opening it lazily if it %% was not granted at connect time. send_default(D, #wt_data{session = S, default_stream = undefined} = Data) -> case webtransport:open_stream(S, bidi) of {ok, StreamId} -> {webtransport:send(S, StreamId, D, nofin), Data#wt_data{default_stream = StreamId}}; {error, _} = Err -> {Err, Data} end; send_default(D, #wt_data{session = S, default_stream = StreamId} = Data) -> {webtransport:send(S, StreamId, D, nofin), Data}. %%==================================================================== %% Internal: inbound events %%==================================================================== %% @private Classify a webtransport handler event and route it. Data on %% the default stream becomes the ws-shaped `{binary, Data}'; data on a %% client-opened stream goes to that stream's per-stream channel; %% everything else (datagrams, server-opened streams) stays on the primary %% channel with its native shape. handle_wt_event(closed, Data) -> session_ended(closed, Data); handle_wt_event(Event, #wt_data{active = Active} = Data) when Active =/= false -> %% Active mode: forward everything to the owner, uniformly tagged. Data1 = clear_default_on_fin(Event, Data), case active_msg(Event, Data) of ignore -> {keep_state, Data1}; Msg -> (Data1#wt_data.owner) ! {hackney_wt, self(), Msg}, case Active of once -> {keep_state, Data1#wt_data{active = false}}; true -> {keep_state, Data1} end end; handle_wt_event(Event, Data) -> %% Passive mode: buffer per channel. route_passive(Event, Data). %% @private Owner-facing message for active mode (ignore = drop). active_msg({stream, Sid, _Type, D}, #wt_data{default_stream = Sid}) -> nonempty_binary(D); active_msg({stream_fin, Sid, _Type, D}, #wt_data{default_stream = Sid}) -> nonempty_binary(D); active_msg({stream, Sid, _Type, D}, _Data) -> {stream, Sid, D}; active_msg({stream_fin, Sid, _Type, D}, _Data) -> {stream_fin, Sid, D}; active_msg({datagram, D}, _Data) -> {datagram, D}; active_msg({stream_closed, Sid, Reason}, _Data) -> {stream_closed, Sid, Reason}; active_msg(_Other, _Data) -> ignore. nonempty_binary(<<>>) -> ignore; nonempty_binary(D) -> {binary, D}. %% @private Route an inbound event into the right passive queue. route_passive({stream, Sid, _Type, D}, #wt_data{default_stream = Sid} = Data) -> deliver_primary_nonempty({binary, D}, Data); route_passive({stream_fin, Sid, _Type, D}, #wt_data{default_stream = Sid} = Data) -> %% Server closed the default stream; drop our id so the next send opens %% a fresh one. deliver_primary_nonempty({binary, D}, Data#wt_data{default_stream = undefined}); route_passive({stream, Sid, _Type, D}, Data) -> case maps:is_key(Sid, Data#wt_data.streams) of true -> deliver_stream(Sid, {data, D}, Data); false -> deliver_primary({stream, Sid, D}, Data) end; route_passive({stream_fin, Sid, _Type, D}, Data) -> case maps:is_key(Sid, Data#wt_data.streams) of true -> deliver_stream(Sid, {fin, D}, Data); false -> deliver_primary({stream_fin, Sid, D}, Data) end; route_passive({datagram, D}, Data) -> deliver_primary({datagram, D}, Data); route_passive({stream_closed, Sid, Reason}, Data) -> case maps:is_key(Sid, Data#wt_data.streams) of true -> deliver_stream(Sid, {closed, Reason}, Data); false -> deliver_primary({stream_closed, Sid, Reason}, Data) end; route_passive(_Other, _Data) -> keep_state_and_data. %% @private In active mode the default stream still has to be forgotten on %% FIN so a later send reopens it. clear_default_on_fin({stream_fin, Sid, _Type, _D}, #wt_data{default_stream = Sid} = Data) -> Data#wt_data{default_stream = undefined}; clear_default_on_fin(_Event, Data) -> Data. %%==================================================================== %% Internal: primary channel buffering %%==================================================================== deliver_primary_nonempty({binary, <<>>}, _Data) -> keep_state_and_data; deliver_primary_nonempty(Msg, Data) -> deliver_primary(Msg, Data). %% @private Hand a primary-channel message to a waiting reader or buffer it. deliver_primary(Msg, #wt_data{recv_from = undefined} = Data) -> enqueue_primary(Msg, Data); deliver_primary(Msg, #wt_data{recv_from = From} = Data) -> {keep_state, Data#wt_data{recv_from = undefined}, [{reply, From, {ok, Msg}}, {{timeout, recv}, cancel}]}. enqueue_primary(Msg, #wt_data{recv_q = Q} = Data) -> case add_bytes(msg_size(Msg), Data) of {ok, Data1} -> {keep_state, Data1#wt_data{recv_q = queue:in(Msg, Q)}}; overflow -> overflow(Data) end. %%==================================================================== %% Internal: per-stream buffering %%==================================================================== %% @private Serve a per-stream recv from the buffer or wait for data. do_stream_recv(From, Sid, #stream{q = Q} = S, Timeout, Data) -> case queue:out(Q) of {{value, Item}, Q1} -> Data1 = sub_bytes(item_size(Item), Data), Streams = maps:put(Sid, S#stream{q = Q1}, Data1#wt_data.streams), {keep_state, Data1#wt_data{streams = Streams}, [{reply, From, stream_item_result(Item)}]}; {empty, _} -> case S#stream.closed of undefined -> Streams = maps:put(Sid, S#stream{from = From}, Data#wt_data.streams), {keep_state, Data#wt_data{streams = Streams}, recv_timer({srecv, Sid}, Timeout)}; Reason -> {keep_state_and_data, [{reply, From, {error, map_stream_reason(Reason)}}]} end end. %% @private Hand a per-stream terminal event to a waiting reader or record it. deliver_stream(Sid, {closed, Reason}, Data) -> S = maps:get(Sid, Data#wt_data.streams), case S#stream.from of undefined -> buffer_stream(Sid, S, {closed, Reason}, Data); From -> S1 = S#stream{from = undefined, closed = Reason}, Streams = maps:put(Sid, S1, Data#wt_data.streams), {keep_state, Data#wt_data{streams = Streams}, [{reply, From, {error, map_stream_reason(Reason)}}, {{timeout, {srecv, Sid}}, cancel}]} end; %% @private Hand a per-stream data/fin item to a waiting reader or buffer it. deliver_stream(Sid, Item, Data) -> S = maps:get(Sid, Data#wt_data.streams), case S#stream.from of undefined -> buffer_stream(Sid, S, Item, Data); From -> S1 = S#stream{from = undefined, closed = closed_after(Item, S#stream.closed)}, Streams = maps:put(Sid, S1, Data#wt_data.streams), {keep_state, Data#wt_data{streams = Streams}, [{reply, From, stream_item_result(Item)}, {{timeout, {srecv, Sid}}, cancel}]} end. buffer_stream(Sid, S, {closed, Reason}, Data) -> %% No bytes to buffer; just record the terminal state. Streams = maps:put(Sid, S#stream{closed = Reason}, Data#wt_data.streams), {keep_state, Data#wt_data{streams = Streams}}; buffer_stream(Sid, S, Item, Data) -> case add_bytes(item_size(Item), Data) of {ok, Data1} -> S1 = S#stream{q = queue:in(Item, S#stream.q), closed = closed_after(Item, S#stream.closed)}, Streams = maps:put(Sid, S1, Data1#wt_data.streams), {keep_state, Data1#wt_data{streams = Streams}}; overflow -> overflow(Data) end. %% @private A delivered FIN means the stream is half-closed for reading %% once the queue drains. closed_after({fin, _}, undefined) -> normal; closed_after(_Item, Closed) -> Closed. stream_item_result({data, D}) -> {ok, D}; stream_item_result({fin, D}) -> {ok, {fin, D}}. map_stream_reason(normal) -> closed; map_stream_reason(Reason) -> Reason. item_size({data, D}) -> byte_size(D); item_size({fin, D}) -> byte_size(D); item_size({closed, _}) -> 0. %%==================================================================== %% Internal: buffer accounting and lifecycle %%==================================================================== %% @private Pop a primary-channel message, adjusting the byte counter. dequeue(Q, Data) -> case queue:out(Q) of {{value, Msg}, Q1} -> {value, Msg, Q1, sub_bytes(msg_size(Msg), Data)}; {empty, _} = E -> E end. add_bytes(_Size, #wt_data{max_recv_buffer = infinity} = Data) -> {ok, Data}; add_bytes(Size, #wt_data{recv_bytes = B, max_recv_buffer = Max} = Data) -> NewBytes = B + Size, case NewBytes > Max of true -> overflow; false -> {ok, Data#wt_data{recv_bytes = NewBytes}} end. sub_bytes(Size, #wt_data{recv_bytes = B} = Data) -> Data#wt_data{recv_bytes = max(0, B - Size)}. %% @private Receive buffer cap exceeded: tear the session down. overflow(Data) -> close_session_safe(Data), _ = maybe_notify_error(recv_buffer_overflow, Data), {next_state, closed, Data#wt_data{closed_reason = recv_buffer_overflow}}. msg_size({binary, B}) -> byte_size(B); msg_size({datagram, B}) -> byte_size(B); msg_size({stream, _, B}) -> byte_size(B); msg_size({stream_fin, _, B}) -> byte_size(B); msg_size({stream_closed, _, _}) -> 0. %% @private Apply a new active mode, flushing buffered messages into active %% delivery as we switch. apply_active(false, Data) -> Data#wt_data{active = false}; apply_active(true, Data) -> flush_all(Data#wt_data{active = true}); apply_active(once, #wt_data{owner = Owner} = Data) -> case dequeue(Data#wt_data.recv_q, Data) of {value, Msg, Q1, Data1} -> Owner ! {hackney_wt, self(), Msg}, %% A single buffered message satisfies `once'; stay passive. Data1#wt_data{recv_q = Q1, active = false}; {empty, _} -> Data#wt_data{active = once} end. %% @private Forward every buffered message (primary + per-stream) to the %% owner, then run with empty queues. flush_all(#wt_data{owner = Owner, recv_q = Q, streams = Streams} = Data) -> lists:foreach(fun(Msg) -> Owner ! {hackney_wt, self(), Msg} end, queue:to_list(Q)), Streams1 = maps:map( fun(Sid, #stream{q = SQ} = S) -> lists:foreach( fun(Item) -> Owner ! {hackney_wt, self(), stream_active_msg(Sid, Item)} end, queue:to_list(SQ)), S#stream{q = queue:new()} end, Streams), Data#wt_data{recv_q = queue:new(), recv_bytes = 0, streams = Streams1}. stream_active_msg(Sid, {data, D}) -> {stream, Sid, D}; stream_active_msg(Sid, {fin, D}) -> {stream_fin, Sid, D}; stream_active_msg(Sid, {closed, R}) -> {stream_closed, Sid, R}. %% @private Handle the session ending (graceful close or crash). Reply to %% any waiting reader, notify the owner in active mode, and move to closed %% while keeping buffered data drainable. session_ended(Reason, Data0) -> {Data1, Actions1} = fail_primary_reader(Reason, Data0), {Data2, Actions2} = fail_stream_readers(Reason, Data1), _ = case Data2#wt_data.active of false -> ok; _ -> maybe_notify_error(Reason, Data2) end, {next_state, closed, Data2#wt_data{closed_reason = Reason}, Actions1 ++ Actions2}. fail_primary_reader(_Reason, #wt_data{recv_from = undefined} = Data) -> {Data, []}; fail_primary_reader(Reason, #wt_data{recv_from = From} = Data) -> {Data#wt_data{recv_from = undefined}, [{reply, From, {error, Reason}}, {{timeout, recv}, cancel}]}. fail_stream_readers(Reason, #wt_data{streams = Streams} = Data) -> maps:fold( fun(Sid, #stream{from = From} = S, {DAcc, AAcc}) when From =/= undefined -> Streams1 = maps:put(Sid, S#stream{from = undefined}, DAcc#wt_data.streams), {DAcc#wt_data{streams = Streams1}, [{reply, From, {error, Reason}}, {{timeout, {srecv, Sid}}, cancel} | AAcc]}; (_Sid, _S, Acc) -> Acc end, {Data, []}, Streams). %% @private In active mode, surface a graceful close as `closed' and any %% other reason as an error. maybe_notify_error(closed, #wt_data{owner = Owner}) -> Owner ! {hackney_wt, self(), closed}; maybe_notify_error(Reason, #wt_data{owner = Owner}) -> Owner ! {hackney_wt_error, self(), Reason}. closed_reason(#wt_data{closed_reason = undefined}) -> closed; closed_reason(#wt_data{closed_reason = R}) -> R. %% @private gen_statem timeout action for a recv, or none for infinity. recv_timer(_Name, infinity) -> []; recv_timer(Name, Timeout) -> [{{timeout, Name}, Timeout, Name}]. %% @private Resolve the effective recv timeout (`default' uses the option). recv_timeout(default, #wt_data{recv_timeout = T}) -> T; recv_timeout(T, _Data) -> T. %% @private Best-effort session close. close_session_safe(#wt_data{session = undefined}) -> ok; close_session_safe(#wt_data{session = Session}) -> try webtransport:close_session(Session) catch _:_ -> ok end, ok.