diff --git a/guides/http2_guide.md b/guides/http2_guide.md index cd10b3ee..4cef5e52 100644 --- a/guides/http2_guide.md +++ b/guides/http2_guide.md @@ -283,6 +283,43 @@ Each chunk is sent as a DATA frame and the request stream is closed with END_STREAM on `finish_send_body/1`. The `h2` connection buffers beyond the peer's flow-control window and drains as WINDOW_UPDATEs arrive. +## Bidirectional Streaming (gRPC-style) + +For full-duplex streams, where the client sends and receives on the same +stream interleaved (as gRPC bidi RPCs do), use the `h2_*` API. It mirrors the +`ws_*` / `wt_*` APIs: `h2_open` returns a pid, `h2_send` writes DATA frames, +`h2_recv` reads inbound messages, and `h2_send_trailers` / `h2_send(_, _, fin)` +half-close the send side. The URL must be `https` (HTTP/2 is negotiated over +ALPN), and each `h2_open` uses its own dedicated connection. + +```erlang +{ok, S} = hackney:h2_open(<<"https://host/pkg.Service/BidiMethod">>, + [{<<"content-type">>, <<"application/grpc">>}, + {<<"te">>, <<"trailers">>}], + [{ssl_options, [...]}]), + +{ok, {response, 200, _Headers}} = hackney:h2_recv(S), +ok = hackney:h2_send(S, Frame1), +{ok, {data, Reply1}} = hackney:h2_recv(S), +ok = hackney:h2_send(S, Frame2), %% keep sending while receiving +{ok, {data, Reply2}} = hackney:h2_recv(S), +ok = hackney:h2_send(S, <<>>, fin), %% half-close the request +{ok, {trailers, Trailers}} = hackney:h2_recv(S), +{ok, done} = hackney:h2_recv(S), +ok = hackney:h2_close(S). +``` + +`h2_recv/1,2` returns `{response, Status, Headers}`, `{data, Data}`, +`{trailers, Trailers}`, or `done` (the peer ended the stream); after `done` it +returns `{error, closed}`. With `{active, true | once}` the same messages are +delivered to the owner as `{hackney_h2, Pid, Msg}` instead (errors as +`{hackney_h2_error, Pid, Reason}`). + +Open with `{flow_control, manual}` to apply receive backpressure: the window is +only replenished when you call `h2_consume(Pid, NBytes)` for the bytes you have +processed. The API carries raw bytes; gRPC message framing is the caller's +responsibility. + ## Flow Control HTTP/2 has built-in flow control to prevent fast senders from overwhelming slow receivers. Hackney handles this automatically: diff --git a/src/hackney.erl b/src/hackney.erl index d59f211b..ad006517 100644 --- a/src/hackney.erl +++ b/src/hackney.erl @@ -42,6 +42,15 @@ wt_send_datagram/2, wt_session_info/1]). +%% HTTP/2 bidirectional (gRPC-style) stream API +-export([h2_open/2, h2_open/3, h2_open/4, + h2_send/2, h2_send/3, + h2_send_trailers/2, + h2_recv/1, h2_recv/2, + h2_consume/2, + h2_setopts/2, + h2_close/1]). + -export([redirect_location/1, location/1]). -export([get_version/0]). @@ -1023,6 +1032,131 @@ wt_session_info(WtPid) when is_pid(WtPid) -> shutdown_wt(WtPid) -> try exit(WtPid, shutdown) catch _:_ -> ok end. +%%==================================================================== +%% HTTP/2 bidirectional (gRPC-style) stream API +%%==================================================================== + +%% @doc Open a full-duplex HTTP/2 stream (gRPC-style bidirectional streaming). +%% Establishes a dedicated HTTP/2 connection (ALPN, so an https URL) and opens +%% one stream on it. Returns a pid driven with h2_send/h2_recv etc. The method +%% defaults to POST. +%% +%% Options: connect_timeout, recv_timeout, connect_options, ssl_options, +%% {flow_control, auto | manual}, {active, true | false | once}, +%% {max_recv_buffer, bytes | infinity}. +-spec h2_open(binary() | string(), list()) -> {ok, pid()} | {error, term()}. +h2_open(URL, Opts) -> + h2_open(post, URL, [], Opts). + +-spec h2_open(binary() | string(), list(), list()) -> {ok, pid()} | {error, term()}. +h2_open(URL, Headers, Opts) -> + h2_open(post, URL, Headers, Opts). + +-spec h2_open(atom() | binary() | string(), binary() | string(), list(), list()) -> + {ok, pid()} | {error, term()}. +h2_open(Method, URL, Headers, Opts) -> + #hackney_url{ + transport = Transport, + scheme = Scheme, + host = Host, + port = Port, + path = Path0, + qs = Query + } = hackney_url:parse_url(URL), + case Transport of + hackney_ssl -> + Path = case Query of + <<>> -> Path0; + _ -> <> + end, + H2Opts = #{ + method => h2_method_bin(Method), + host => Host, + port => Port, + transport => Transport, + path => Path, + headers => Headers, + connect_timeout => proplists:get_value(connect_timeout, Opts, 8000), + recv_timeout => proplists:get_value(recv_timeout, Opts, infinity), + connect_options => proplists:get_value(connect_options, Opts, []), + ssl_options => proplists:get_value(ssl_options, Opts, []), + flow_control => proplists:get_value(flow_control, Opts, auto), + active => proplists:get_value(active, Opts, false), + max_recv_buffer => proplists:get_value(max_recv_buffer, Opts, 16#4000000) + }, + case hackney_h2_stream:start_link(H2Opts) of + {ok, Pid} -> + Timeout = maps:get(connect_timeout, H2Opts), + try hackney_h2_stream:connect(Pid, Timeout) of + ok -> + {ok, Pid}; + {error, Reason} -> + shutdown_h2(Pid), + {error, Reason} + catch + exit:{timeout, _} -> + shutdown_h2(Pid), + {error, connect_timeout}; + exit:{noproc, _} -> + {error, {h2_process_died, noproc}} + end; + {error, Reason} -> + {error, Reason} + end; + _ -> + {error, {scheme_not_supported, Scheme}} + end. + +%% @doc Send a DATA frame on the stream (no END_STREAM). +-spec h2_send(pid(), iodata()) -> ok | {error, term()}. +h2_send(Pid, Data) when is_pid(Pid) -> + hackney_h2_stream:send(Pid, Data). + +%% @doc Send a DATA frame, optionally half-closing the send side (`fin'). +-spec h2_send(pid(), iodata(), fin | nofin) -> ok | {error, term()}. +h2_send(Pid, Data, Fin) when is_pid(Pid) -> + hackney_h2_stream:send(Pid, Data, Fin). + +%% @doc Send trailing HEADERS, half-closing the send side (gRPC trailers). +-spec h2_send_trailers(pid(), list()) -> ok | {error, term()}. +h2_send_trailers(Pid, Trailers) when is_pid(Pid) -> + hackney_h2_stream:send_trailers(Pid, Trailers). + +%% @doc Receive the next inbound message: {response, Status, Headers} | +%% {data, Data} | {trailers, Trailers} | done. After done, returns +%% {error, closed}. Passive mode only. +-spec h2_recv(pid()) -> {ok, hackney_h2_stream:h2_msg()} | {error, term()}. +h2_recv(Pid) when is_pid(Pid) -> + hackney_h2_stream:recv(Pid). + +-spec h2_recv(pid(), timeout()) -> {ok, hackney_h2_stream:h2_msg()} | {error, term()}. +h2_recv(Pid, Timeout) when is_pid(Pid) -> + hackney_h2_stream:recv(Pid, Timeout). + +%% @doc Acknowledge N consumed bytes (manual flow control only). +-spec h2_consume(pid(), non_neg_integer()) -> ok | {error, term()}. +h2_consume(Pid, NBytes) when is_pid(Pid) -> + hackney_h2_stream:consume(Pid, NBytes). + +%% @doc Set options. Supported: [{active, true | false | once}]. +-spec h2_setopts(pid(), list()) -> ok | {error, term()}. +h2_setopts(Pid, Opts) when is_pid(Pid) -> + hackney_h2_stream:setopts(Pid, Opts). + +%% @doc Cancel the stream and tear down its connection. +-spec h2_close(pid()) -> ok. +h2_close(Pid) when is_pid(Pid) -> + hackney_h2_stream:close(Pid). + +%% @private Normalize an HTTP method to an uppercase binary. +h2_method_bin(M) when is_binary(M) -> M; +h2_method_bin(M) when is_atom(M) -> list_to_binary(string:to_upper(atom_to_list(M))); +h2_method_bin(M) when is_list(M) -> list_to_binary(string:to_upper(M)). + +%% @private Signal the HTTP/2 stream process to shut down, ignoring errors. +shutdown_h2(Pid) -> + try exit(Pid, shutdown) catch _:_ -> ok end. + %% @private Reject CR/LF/NUL in the authority, request path, or any %% caller-supplied header used in the WebTransport CONNECT request %% (GHSA-f9vr analog). diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index c24e784a..e951a444 100644 --- a/src/hackney_conn.erl +++ b/src/hackney_conn.erl @@ -42,6 +42,8 @@ send_body_chunk/2, finish_send_body/1, start_response/1, + %% HTTP/2 bidirectional stream (handler-routed, for hackney_h2_stream) + open_h2_stream/6, %% Async streaming request_async/6, request_async/7, @@ -361,6 +363,18 @@ finish_send_body(Pid) -> start_response(Pid) -> safe_call(Pid, start_response, infinity). +%% @doc Open an HTTP/2 stream whose events are routed to HandlerPid (the gRPC +%% bidi model), returning the underlying h2_connection pid and stream id so the +%% handler can drive send_data/send_trailers/consume directly. Used by +%% hackney_h2_stream; the stream is not tracked in this gen_statem. +-spec open_h2_stream(pid(), binary(), binary(), list(), pid(), map()) -> + {ok, pid(), pos_integer()} | {error, term()}. +open_h2_stream(Pid, Method, Path, Headers, HandlerPid, Opts) -> + case valid_request_target(Path) of + ok -> safe_call(Pid, {open_h2_stream, Method, Path, Headers, HandlerPid, Opts}, infinity); + Err -> Err + end. + %% @doc Get the full response body. -spec body(pid()) -> {ok, binary()} | {error, term()}. body(Pid) -> @@ -989,6 +1003,25 @@ connected({call, From}, {send_headers, Method, Path, Headers}, #conn_data{protoc %% chunks via send_body_chunk/finish_send_body. Mirrors do_h3_send_headers/5. do_h2_send_headers(From, Method, Path, Headers, Data); +connected({call, From}, {open_h2_stream, Method, Path, Headers, HandlerPid, Opts}, + #conn_data{protocol = http2, h2_conn = H2Conn} = Data) -> + %% Open a stream routed to HandlerPid (gRPC bidi). The handler owns the + %% stream end to end; we do not track it in h2_streams. Returns the + %% h2_connection pid + stream id so the handler drives it directly. + {_, _, H2Headers} = build_h2_request_headers(Method, Path, Headers, Data), + FlowControl = maps:get(flow_control, Opts, auto), + StreamOpts = #{handler => HandlerPid, flow_control => FlowControl}, + Reply = try + case h2_connection:send_request_headers(H2Conn, H2Headers, false, StreamOpts) of + {ok, StreamId} -> {ok, H2Conn, StreamId}; + {error, _} = E -> E + end + catch + exit:{ExitReason, _} -> {error, {closed, ExitReason}}; + exit:ExitReason -> {error, {closed, ExitReason}} + end, + {keep_state_and_data, [{reply, From, Reply}]}; + connected({call, From}, {send_headers, Method, Path, Headers}, Data) -> %% Send only headers for streaming body mode (HTTP/1.1) NewData = Data#conn_data{ diff --git a/src/hackney_h2_stream.erl b/src/hackney_h2_stream.erl new file mode 100644 index 00000000..7ce71937 --- /dev/null +++ b/src/hackney_h2_stream.erl @@ -0,0 +1,592 @@ +%%% -*- 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 a single full-duplex HTTP/2 stream +%%% (gRPC-style bidirectional streaming). +%%% +%%% This module mirrors {@link hackney_wt} / {@link hackney_ws}: an +%%% application gets a pid and uses `send'/`recv' (passive) or active-mode +%%% messages. It owns one dedicated HTTP/2 connection (negotiated via ALPN) +%%% and opens one stream on it whose events are routed here directly by the +%%% `h2' library's per-stream handler mechanism, so the request body and the +%%% response can be sent and read interleaved on the same stream. +%%% +%%% == Send / receive == +%%% +%%% `send/2,3' writes a DATA frame (optionally with END_STREAM via `fin'). +%%% `send_trailers/2' closes the send side with trailing HEADERS (gRPC +%%% carries grpc-status there). `recv/1,2' returns the next inbound message: +%%% `{response, Status, Headers}', `{data, Data}', `{trailers, Trailers}' or +%%% `done' (the peer ended the stream). After `done', recv returns +%%% `{error, closed}'. +%%% +%%% == Delivery modes == +%%% +%%% Passive (default): messages are buffered and read with `recv'. Active +%%% (`{active, true|once}'): every message is forwarded to the owner as +%%% `{hackney_h2, Pid, Msg}', and errors as `{hackney_h2_error, Pid, Reason}'. +%%% +%%% == Flow control == +%%% +%%% With `{flow_control, manual}' the receive window is only replenished by +%%% `consume/2', so a slow reader applies backpressure to the peer. The +%%% default `auto' replenishes automatically. +%%% +%%% States: idle (not yet connected), connected (ready), closed (stream/conn +%%% ended, buffered data still drainable). +-module(hackney_h2_stream). +-behaviour(gen_statem). + +%% API +-export([ + start_link/1, + connect/1, connect/2, + send/2, send/3, + send_trailers/2, + recv/1, recv/2, + consume/2, + setopts/2, + controlling_process/2, + close/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 decoded bytes buffered in the passive recv queue so a hostile +%% peer cannot drive the client to OOM; complements HTTP/2 flow control. +-define(DEFAULT_MAX_RECV_BUFFER, 16#4000000). %% 64 MiB + +%% A normalized inbound message handed to recv/2 or, in active mode, the owner. +-type h2_msg() :: {response, non_neg_integer(), list()} + | {data, binary()} + | {trailers, list()} + | done. + +-export_type([h2_msg/0]). + +-record(h2s_data, { + %% Stream owner (linked via start_link, trap_exit handles death) + owner :: pid(), + + %% Request identity used to open the stream + method :: binary(), + host :: string() | binary(), + port :: inet:port_number(), + transport :: module(), + path :: binary(), + headers = [] :: list(), + + %% Connection setup + connect_options = [] :: list(), + ssl_options = [] :: list(), + connect_timeout = ?CONNECT_TIMEOUT :: timeout(), + recv_timeout = ?RECV_TIMEOUT :: timeout(), + flow_control = auto :: auto | manual, + active = false :: false | true | once, + + %% Underlying hackney_conn (dedicated, owned here) and the h2 stream + conn_pid :: pid() | undefined, + conn_mon :: reference() | undefined, + h2_conn :: pid() | undefined, + stream_id :: pos_integer() | undefined, + + %% Passive recv buffering + recv_q = queue:new() :: queue:queue(), + recv_from :: gen_statem:from() | undefined, + recv_bytes = 0 :: non_neg_integer(), + max_recv_buffer = ?DEFAULT_MAX_RECV_BUFFER :: non_neg_integer() | infinity, + + %% Peer END_STREAM seen (no more inbound after the queue drains) + ended = false :: boolean(), + closed_reason :: term() +}). + +%%==================================================================== +%% API +%%==================================================================== + +%% @doc Start a stream process. See hackney:h2_open/3,4 for option docs. +-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 connection and open the stream. Blocks until ready. +-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}, infinity). + +%% @doc Send a DATA frame on the stream (no END_STREAM). +-spec send(pid(), iodata()) -> ok | {error, term()}. +send(Pid, Data) -> + send(Pid, Data, nofin). + +%% @doc Send a DATA frame, optionally half-closing the send side (`fin'). +-spec send(pid(), iodata(), fin | nofin) -> ok | {error, term()}. +send(Pid, Data, Fin) when Fin =:= fin; Fin =:= nofin -> + gen_statem:call(Pid, {send, Data, Fin}). + +%% @doc Send trailing HEADERS, half-closing the send side (gRPC trailers). +-spec send_trailers(pid(), list()) -> ok | {error, term()}. +send_trailers(Pid, Trailers) -> + gen_statem:call(Pid, {send_trailers, Trailers}). + +%% @doc Receive the next inbound message (passive mode only). +-spec recv(pid()) -> {ok, h2_msg()} | {error, term()}. +recv(Pid) -> + gen_statem:call(Pid, {recv, default}, infinity). + +-spec recv(pid(), timeout()) -> {ok, h2_msg()} | {error, term()}. +recv(Pid, Timeout) -> + gen_statem:call(Pid, {recv, Timeout}, infinity). + +%% @doc Acknowledge N consumed bytes (manual flow control only). +-spec consume(pid(), non_neg_integer()) -> ok | {error, term()}. +consume(Pid, NBytes) when is_integer(NBytes), NBytes >= 0 -> + gen_statem:call(Pid, {consume, NBytes}). + +%% @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 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 Cancel the stream and tear down the connection. +-spec close(pid()) -> ok. +close(Pid) -> + gen_statem:cast(Pid, close). + +%%==================================================================== +%% gen_statem callbacks +%%==================================================================== + +%% @private +callback_mode() -> + [state_functions, state_enter]. + +%% @private +init([Owner, Opts]) -> + process_flag(trap_exit, true), + Data = #h2s_data{ + owner = Owner, + method = maps:get(method, Opts, <<"POST">>), + host = maps:get(host, Opts), + port = maps:get(port, Opts), + transport = maps:get(transport, Opts), + path = maps:get(path, Opts, <<"/">>), + headers = maps:get(headers, Opts, []), + connect_options = maps:get(connect_options, Opts, []), + ssl_options = maps:get(ssl_options, Opts, []), + connect_timeout = maps:get(connect_timeout, Opts, ?CONNECT_TIMEOUT), + recv_timeout = maps:get(recv_timeout, Opts, ?RECV_TIMEOUT), + flow_control = maps:get(flow_control, Opts, auto), + 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, #h2s_data{conn_pid = undefined}) -> + ok; +terminate(_Reason, _State, #h2s_data{conn_pid = ConnPid}) -> + stop_conn(ConnPid), + 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) -> + case establish(Timeout, Data) of + {ok, Data1} -> + {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}, #h2s_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, SData, Fin}, + #h2s_data{h2_conn = H2Conn, stream_id = Sid}) -> + EndStream = (Fin =:= fin), + Reply = h2_call(fun() -> + h2_connection:send_data(H2Conn, Sid, iolist_to_binary(SData), EndStream) + end), + {keep_state_and_data, [{reply, From, Reply}]}; + +connected({call, From}, {send_trailers, Trailers}, + #h2s_data{h2_conn = H2Conn, stream_id = Sid}) -> + Reply = h2_call(fun() -> h2_connection:send_trailers(H2Conn, Sid, Trailers) end), + {keep_state_and_data, [{reply, From, Reply}]}; + +connected({call, From}, {consume, NBytes}, + #h2s_data{h2_conn = H2Conn, stream_id = Sid}) -> + Reply = h2_call(fun() -> h2_connection:consume(H2Conn, Sid, NBytes) end), + {keep_state_and_data, [{reply, From, Reply}]}; + +%% --- recv ------------------------------------------------------------ +connected({call, From}, {recv, _Timeout}, #h2s_data{active = Active}) + when Active =/= false -> + {keep_state_and_data, [{reply, From, {error, {active_mode, Active}}}]}; + +connected({call, From}, {recv, _Timeout}, #h2s_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) of + {value, Msg, Data1} -> + {keep_state, Data1, [{reply, From, {ok, Msg}}]}; + empty -> + case Data#h2s_data.ended of + true -> + {keep_state_and_data, [{reply, From, {error, closed}}]}; + false -> + {keep_state, Data#h2s_data{recv_from = From}, recv_timer(Timeout)} + end + 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 -> + {keep_state, apply_active(NewActive, Data), [{reply, From, ok}]}; + _ -> + {keep_state_and_data, [{reply, From, {error, badarg}}]} + end; + +connected({call, From}, {controlling_process, NewOwner}, #h2s_data{owner = OldOwner} = Data) -> + unlink(OldOwner), + link(NewOwner), + {keep_state, Data#h2s_data{owner = NewOwner}, [{reply, From, ok}]}; + +connected({call, From}, _Request, _Data) -> + {keep_state_and_data, [{reply, From, {error, badrequest}}]}; + +%% --- close ----------------------------------------------------------- +connected(cast, close, Data) -> + cancel_stream_safe(Data), + stop_conn(Data#h2s_data.conn_pid), + {next_state, closed, Data#h2s_data{closed_reason = closed}}; + +%% --- recv timeout ---------------------------------------------------- +connected({timeout, recv}, recv, #h2s_data{recv_from = From} = Data) + when From =/= undefined -> + {keep_state, Data#h2s_data{recv_from = undefined}, [{reply, From, {error, timeout}}]}; +connected({timeout, recv}, recv, _Data) -> + keep_state_and_data; + +%% --- stream events from the h2 connection ---------------------------- +connected(info, {h2, H2Conn, Event}, #h2s_data{h2_conn = H2Conn} = Data) -> + handle_h2_event(Event, Data); + +%% The hackney_conn owning the h2 connection died. +connected(info, {'DOWN', Mon, process, _Pid, Reason}, #h2s_data{conn_mon = Mon} = Data) -> + stream_ended(down_reason(Reason), Data#h2s_data{conn_pid = undefined, conn_mon = undefined}); + +connected(info, {'EXIT', Owner, _Reason}, #h2s_data{owner = Owner} = Data) -> + cancel_stream_safe(Data), + stop_conn(Data#h2s_data.conn_pid), + {stop, normal}; + +connected(_, _, _) -> + keep_state_and_data. + +%%==================================================================== +%% State: closed +%%==================================================================== + +closed(enter, _OldState, _Data) -> + keep_state_and_data; + +%% Drain whatever was buffered before the stream/connection ended. +closed({call, From}, {recv, _Timeout}, Data) -> + case dequeue(Data) of + {value, Msg, Data1} -> + {keep_state, Data1, [{reply, From, {ok, Msg}}]}; + empty -> + {keep_state_and_data, [{reply, From, {error, closed_reason(Data)}}]} + end; + +%% Once the stream has ended there is no receive window left to manage, so +%% consume is a harmless no-op (a slow reader may still be draining + acking +%% buffered chunks after the peer's END_STREAM arrived). +closed({call, From}, {consume, _NBytes}, _Data) -> + {keep_state_and_data, [{reply, From, ok}]}; + +closed({call, From}, {controlling_process, NewOwner}, #h2s_data{owner = OldOwner} = Data) -> + unlink(OldOwner), + link(NewOwner), + {keep_state, Data#h2s_data{owner = NewOwner}, [{reply, From, ok}]}; + +closed({call, From}, _Request, _Data) -> + {keep_state_and_data, [{reply, From, {error, closed}}]}; + +closed(cast, close, _Data) -> + keep_state_and_data; + +closed(info, {'EXIT', Owner, _Reason}, #h2s_data{owner = Owner}) -> + {stop, normal}; + +closed(_, _, _) -> + keep_state_and_data. + +%%==================================================================== +%% Internal: connection + stream setup +%%==================================================================== + +%% @private Open a dedicated HTTP/2 connection and one handler-routed stream. +establish(Timeout, Data) -> + #h2s_data{host = Host, port = Port, transport = Transport, + connect_options = COpts, ssl_options = SslOpts, + recv_timeout = RT, method = Method, path = Path, + headers = Headers, flow_control = FC} = Data, + ConnOpts = #{ + host => Host, + port => Port, + transport => Transport, + connect_timeout => Timeout, + recv_timeout => RT, + connect_options => [{protocols, [http2]} | COpts], + ssl_options => SslOpts + }, + case hackney_conn_sup:start_conn(ConnOpts) of + {ok, ConnPid} -> + case hackney_conn:connect(ConnPid, Timeout) of + ok -> + case hackney_conn:get_protocol(ConnPid) of + http2 -> + open_stream(ConnPid, Method, Path, Headers, FC, Data); + Other -> + stop_conn(ConnPid), + {error, {protocol_not_negotiated, Other}} + end; + {error, Reason} -> + stop_conn(ConnPid), + {error, Reason} + end; + {error, Reason} -> + {error, Reason} + end. + +open_stream(ConnPid, Method, Path, Headers, FC, Data) -> + case hackney_conn:open_h2_stream(ConnPid, Method, Path, Headers, self(), + #{flow_control => FC}) of + {ok, H2Conn, StreamId} -> + Mon = erlang:monitor(process, ConnPid), + {ok, Data#h2s_data{conn_pid = ConnPid, conn_mon = Mon, + h2_conn = H2Conn, stream_id = StreamId}}; + {error, Reason} -> + stop_conn(ConnPid), + {error, Reason} + end. + +%%==================================================================== +%% Internal: inbound stream events +%%==================================================================== + +%% @private Map an {h2, Conn, Event} stream event onto recv messages. +handle_h2_event({response, _Sid, Status, Headers}, Data) -> + deliver({response, Status, Headers}, Data); +handle_h2_event({informational, _Sid, _Status, _Headers}, Data) -> + {keep_state, Data}; +handle_h2_event({data, _Sid, Body, false}, Data) -> + case Body of + <<>> -> {keep_state, Data}; + _ -> deliver({data, Body}, Data) + end; +handle_h2_event({data, _Sid, Body, true}, Data) -> + Data1 = case Body of + <<>> -> Data; + _ -> deliver_acc({data, Body}, Data) + end, + finish(Data1); +handle_h2_event({trailers, _Sid, Trailers}, Data) -> + Data1 = deliver_acc({trailers, Trailers}, Data), + finish(Data1); +handle_h2_event({stream_reset, _Sid, ErrorCode}, Data) -> + stream_ended({stream_error, ErrorCode}, Data); +handle_h2_event({goaway, _LastSid, _ErrorCode}, #h2s_data{ended = true}) -> + keep_state_and_data; +handle_h2_event({goaway, _LastSid, ErrorCode}, Data) -> + stream_ended({goaway, ErrorCode}, Data); +handle_h2_event({closed, Reason}, Data) -> + stream_ended(down_reason(Reason), Data); +handle_h2_event(_Other, Data) -> + {keep_state, Data}. + +%% @private Deliver a message, returning the gen_statem transition. +deliver(Msg, Data) -> + {keep_state, deliver_acc(Msg, Data)}. + +%% @private Deliver/buffer a message and return the updated #h2s_data only, +%% so callers can chain (e.g. data then done). +deliver_acc(Msg, #h2s_data{active = Active} = Data) when Active =/= false -> + _ = (Data#h2s_data.owner) ! {hackney_h2, self(), Msg}, + case Active of + once -> Data#h2s_data{active = false}; + true -> Data + end; +deliver_acc(Msg, #h2s_data{recv_from = From} = Data) when From =/= undefined -> + gen_statem:reply(From, {ok, Msg}), + %% A pending {timeout, recv} may still fire; it is a no-op once recv_from + %% is cleared (see the connected/closed {timeout, recv} clauses). + Data#h2s_data{recv_from = undefined}; +deliver_acc(Msg, Data) -> + enqueue(Msg, Data). + +%% @private Mark the stream ended and surface a terminal `done', then move to +%% closed (buffered data stays drainable). +finish(Data0) -> + Data1 = deliver_acc(done, Data0#h2s_data{ended = true}), + {next_state, closed, Data1#h2s_data{closed_reason = closed}}. + +%% @private The stream/connection ended abnormally: fail a waiting reader, +%% notify the owner in active mode, move to closed. +stream_ended(Reason, #h2s_data{recv_from = From} = Data) when From =/= undefined -> + _ = maybe_notify_error(Reason, Data), + {next_state, closed, + Data#h2s_data{recv_from = undefined, ended = true, closed_reason = Reason}, + [{reply, From, {error, Reason}}]}; +stream_ended(Reason, Data) -> + _ = maybe_notify_error(Reason, Data), + {next_state, closed, Data#h2s_data{ended = true, closed_reason = Reason}}. + +maybe_notify_error(_Reason, #h2s_data{active = false}) -> + ok; +maybe_notify_error(closed, #h2s_data{owner = Owner}) -> + Owner ! {hackney_h2, self(), {closed, closed}}; +maybe_notify_error(Reason, #h2s_data{owner = Owner}) -> + Owner ! {hackney_h2_error, self(), Reason}. + +%%==================================================================== +%% Internal: passive buffering +%%==================================================================== + +enqueue(Msg, #h2s_data{recv_q = Q} = Data) -> + case add_bytes(msg_size(Msg), Data) of + {ok, Data1} -> + Data1#h2s_data{recv_q = queue:in(Msg, Q)}; + overflow -> + %% Tear the connection down; recv surfaces the overflow reason. + cancel_stream_safe(Data), + stop_conn(Data#h2s_data.conn_pid), + Data#h2s_data{ended = true, closed_reason = recv_buffer_overflow} + end. + +dequeue(#h2s_data{recv_q = Q} = Data) -> + case queue:out(Q) of + {{value, Msg}, Q1} -> + {value, Msg, sub_bytes(msg_size(Msg), Data#h2s_data{recv_q = Q1})}; + {empty, _} -> + empty + end. + +add_bytes(_Size, #h2s_data{max_recv_buffer = infinity} = Data) -> + {ok, Data}; +add_bytes(Size, #h2s_data{recv_bytes = B, max_recv_buffer = Max} = Data) -> + case B + Size > Max of + true -> overflow; + false -> {ok, Data#h2s_data{recv_bytes = B + Size}} + end. + +sub_bytes(Size, #h2s_data{recv_bytes = B} = Data) -> + Data#h2s_data{recv_bytes = max(0, B - Size)}. + +msg_size({data, B}) -> byte_size(B); +msg_size(_) -> 0. + +%% @private Apply a new active mode, flushing buffered messages on the way. +apply_active(false, Data) -> + Data#h2s_data{active = false}; +apply_active(true, #h2s_data{owner = Owner, recv_q = Q} = Data) -> + lists:foreach(fun(Msg) -> Owner ! {hackney_h2, self(), Msg} end, queue:to_list(Q)), + Data#h2s_data{active = true, recv_q = queue:new(), recv_bytes = 0}; +apply_active(once, #h2s_data{owner = Owner} = Data) -> + case dequeue(Data) of + {value, Msg, Data1} -> + Owner ! {hackney_h2, self(), Msg}, + Data1#h2s_data{active = false}; + empty -> + Data#h2s_data{active = once} + end. + +%%==================================================================== +%% Internal: helpers +%%==================================================================== + +recv_timer(infinity) -> []; +recv_timer(Timeout) -> [{{timeout, recv}, Timeout, recv}]. + +recv_timeout(default, #h2s_data{recv_timeout = T}) -> T; +recv_timeout(T, _Data) -> T. + +closed_reason(#h2s_data{closed_reason = undefined}) -> closed; +closed_reason(#h2s_data{closed_reason = R}) -> R. + +down_reason(normal) -> closed; +down_reason(shutdown) -> closed; +down_reason({shutdown, _}) -> closed; +down_reason(Reason) -> {closed, Reason}. + +%% @private Run an h2_connection call, normalising a dead-connection exit. +h2_call(Fun) -> + try Fun() + catch + exit:{ExitReason, _} -> {error, {closed, ExitReason}}; + exit:ExitReason -> {error, {closed, ExitReason}} + end. + +cancel_stream_safe(#h2s_data{h2_conn = undefined}) -> + ok; +cancel_stream_safe(#h2s_data{h2_conn = H2Conn, stream_id = Sid}) -> + _ = try h2_connection:cancel_stream(H2Conn, Sid) catch _:_ -> ok end, + ok. + +stop_conn(undefined) -> + ok; +stop_conn(ConnPid) -> + try hackney_conn:stop(ConnPid) catch _:_ -> ok end, + ok. diff --git a/test/hackney_h2_bidi_tests.erl b/test/hackney_h2_bidi_tests.erl new file mode 100644 index 00000000..eb220288 --- /dev/null +++ b/test/hackney_h2_bidi_tests.erl @@ -0,0 +1,172 @@ +%%% Tests for the full-duplex HTTP/2 bidirectional stream API (h2_*). +%%% +%%% The server handler does a gRPC-shaped bidi echo: respond 200, echo every +%%% client DATA frame back, then send grpc-status trailers when the client +%%% half-closes. This exercises sending and receiving interleaved on one +%%% stream, trailers, active mode, and manual receive flow control. +-module(hackney_h2_bidi_tests). + +-include_lib("eunit/include/eunit.hrl"). + +cert_dir() -> + BeamDir = filename:dirname(code:which(?MODULE)), + Root = filename:join([BeamDir, "..", "..", "..", "..", ".."]), + filename:join([filename:absname(Root), "test", "certs"]). + +%%==================================================================== +%% Fixture +%%==================================================================== + +h2_bidi_test_() -> + {setup, + fun setup/0, + fun cleanup/1, + fun(Ctx) -> + [{"full duplex: interleaved send/recv + trailers", + {timeout, 30, fun() -> t_full_duplex(Ctx) end}}, + {"active mode delivers messages in order", + {timeout, 30, fun() -> t_active(Ctx) end}}, + {"manual flow control round-trips a large payload", + {timeout, 30, fun() -> t_manual_flow_control(Ctx) end}}, + {"close makes recv return closed", + {timeout, 30, fun() -> t_close(Ctx) end}}] + end}. + +setup() -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + Certs = cert_dir(), + {ok, Server} = h2:start_server(0, #{ + cert => filename:join(Certs, "server.pem"), + key => filename:join(Certs, "server.key"), + handler => fun bidi_echo_handler/5, + settings => #{max_concurrent_streams => unlimited} + }), + #{server => Server, port => h2:server_port(Server)}. + +cleanup(#{server := Server}) -> + try h2:stop_server(Server) catch _:_ -> ok end, + ok. + +%%==================================================================== +%% Server: gRPC-shaped bidi echo +%%==================================================================== + +bidi_echo_handler(Conn, Sid, _Method, _Path, _Headers) -> + ok = h2:set_stream_handler(Conn, Sid, self()), + ok = h2:send_response(Conn, Sid, 200, + [{<<"content-type">>, <<"application/grpc">>}]), + case echo_loop(Conn, Sid) of + complete -> + ok = h2:send_trailers(Conn, Sid, [{<<"grpc-status">>, <<"0">>}]); + aborted -> + %% Client cancelled/closed the stream; nothing more to send. + ok + end. + +echo_loop(Conn, Sid) -> + receive + {h2, Conn, {data, Sid, Data, true}} -> + _ = (Data =/= <<>>) andalso h2:send_data(Conn, Sid, Data, false), + complete; + {h2, Conn, {data, Sid, Data, false}} -> + _ = (Data =/= <<>>) andalso h2:send_data(Conn, Sid, Data, false), + echo_loop(Conn, Sid); + {h2, Conn, {trailers, Sid, _T}} -> + complete; + {h2, Conn, {stream_reset, Sid, _}} -> + aborted; + {h2, Conn, {closed, _}} -> + aborted + after 15000 -> + aborted + end. + +%%==================================================================== +%% Tests +%%==================================================================== + +t_full_duplex(#{port := Port}) -> + {ok, S} = hackney:h2_open(url(Port, <<"/grpc">>), + [{<<"te">>, <<"trailers">>}], opts()), + %% Server responds 200 before we finish sending: full-duplex. + {ok, {response, 200, _Hdrs}} = hackney:h2_recv(S), + ok = hackney:h2_send(S, <<"msg1">>), + {ok, {data, <<"msg1">>}} = hackney:h2_recv(S), + %% Send more after having read a response chunk. + ok = hackney:h2_send(S, <<"msg2">>), + {ok, {data, <<"msg2">>}} = hackney:h2_recv(S), + ok = hackney:h2_send(S, <<>>, fin), + {ok, {trailers, T}} = hackney:h2_recv(S), + ?assertEqual(<<"0">>, proplists:get_value(<<"grpc-status">>, T)), + {ok, done} = hackney:h2_recv(S), + {error, closed} = hackney:h2_recv(S), + ok = hackney:h2_close(S). + +t_active(#{port := Port}) -> + {ok, S} = hackney:h2_open(url(Port, <<"/grpc">>), [], + [{active, true} | opts()]), + ok = hackney:h2_send(S, <<"ping">>), + ok = hackney:h2_send(S, <<>>, fin), + Msgs = collect_active(S, []), + ?assertMatch({response, 200, _}, hd(Msgs)), + ?assert(lists:member({data, <<"ping">>}, Msgs)), + ?assert(lists:any(fun({trailers, _}) -> true; (_) -> false end, Msgs)), + ?assertEqual(done, lists:last(Msgs)), + ok = hackney:h2_close(S). + +t_manual_flow_control(#{port := Port}) -> + Payload = crypto:strong_rand_bytes(120000), + {ok, S} = hackney:h2_open(url(Port, <<"/grpc">>), [], + [{flow_control, manual} | opts()]), + {ok, {response, 200, _}} = hackney:h2_recv(S), + ok = hackney:h2_send(S, Payload), + ok = hackney:h2_send(S, <<>>, fin), + Echoed = drain_consume(S, <<>>), + ?assertEqual(Payload, Echoed), + ok = hackney:h2_close(S). + +t_close(#{port := Port}) -> + {ok, S} = hackney:h2_open(url(Port, <<"/grpc">>), [], opts()), + {ok, {response, 200, _}} = hackney:h2_recv(S), + ok = hackney:h2_close(S), + ?assertEqual({error, closed}, hackney:h2_recv(S)). + +%%==================================================================== +%% Helpers +%%==================================================================== + +opts() -> + [{recv_timeout, 10000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}]. + +url(Port, Path) -> + iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), Path]). + +collect_active(S, Acc) -> + receive + {hackney_h2, S, done} -> + lists:reverse([done | Acc]); + {hackney_h2, S, Msg} -> + collect_active(S, [Msg | Acc]); + {hackney_h2_error, S, Reason} -> + lists:reverse([{error, Reason} | Acc]) + after 10000 -> + lists:reverse([timeout | Acc]) + end. + +%% Drain a manual-flow-control stream, acknowledging each chunk via consume so +%% the peer's send window keeps opening. +drain_consume(S, Acc) -> + case hackney:h2_recv(S) of + {ok, {data, Chunk}} -> + %% Ack the window; once the stream has ended this is a no-op. + _ = hackney:h2_consume(S, byte_size(Chunk)), + drain_consume(S, <>); + {ok, {trailers, _}} -> + drain_consume(S, Acc); + {ok, done} -> + Acc; + {ok, {response, _, _}} -> + drain_consume(S, Acc) + end.