Merge pull request #876 from benoitc/feature/http2-bidi-streams

Add HTTP/2 full-duplex bidirectional stream API (h2_*)

Author: benoitc

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:

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;
+        _ -> <<Path0/binary, "?", Query/binary>>
+      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).

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{