feat: polymorphic helper for request body streaming

lib/mint/core/conn.ex

@@ -62,4 +62,6 @@ defmodule Mint.Core.Conn do
   @callback put_proxy_headers(conn(), Mint.Types.headers()) :: conn()
 
   @callback put_log(conn(), boolean()) :: conn()
+
+  @callback get_send_window(conn(), term()) :: non_neg_integer()
 end

lib/mint/http.ex

@@ -126,7 +126,7 @@ defmodule Mint.HTTP do
 
   @behaviour Mint.Core.Conn
 
-  @opaque t() :: Mint.HTTP1.t() | Mint.HTTP2.t()
+  @type t() :: Mint.HTTP1.t() | Mint.HTTP2.t() | Mint.UnsafeProxy.t()
 
   defguardp is_data_message(message)
             when elem(message, 0) in [:ssl, :tcp] and tuple_size(message) == 3
@@ -1065,6 +1065,12 @@ defmodule Mint.HTTP do
   @impl true
   def put_proxy_headers(conn, headers), do: conn_apply(conn, :put_proxy_headers, [conn, headers])
 
+  @impl true
+  def get_send_window(conn, ref), do: conn_apply(conn, :get_send_window, [conn, ref])
+
+  @spec next_body_chunk_size(t(), term(), binary()) :: non_neg_integer()
+  def next_body_chunk_size(conn, ref, body), do: min(get_send_window(conn, ref), byte_size(body))
+
   ## Helpers
 
   defp conn_apply(%UnsafeProxy{}, fun, args), do: apply(UnsafeProxy, fun, args)

lib/mint/http1.ex

@@ -102,7 +102,8 @@ defmodule Mint.HTTP1 do
     proxy_headers: [],
     private: %{},
     log: false,
-    optional_responses: []
+    optional_responses: [],
+    sndbuf: nil
   ]
 
   defmacrop log(conn, level, message) do
@@ -204,6 +205,7 @@ defmodule Mint.HTTP1 do
     end
 
     with :ok <- Util.inet_opts(transport, socket),
+         {:ok, sndbuf_opts} <- transport.getopts(socket, [:sndbuf]),
          :ok <- if(mode == :active, do: transport.setopts(socket, active: :once), else: :ok) do
       conn = %__MODULE__{
         transport: transport,
@@ -216,7 +218,8 @@ defmodule Mint.HTTP1 do
         log: log?,
         case_sensitive_headers: Keyword.get(opts, :case_sensitive_headers, false),
         skip_target_validation: Keyword.get(opts, :skip_target_validation, false),
-        optional_responses: validate_optional_response_values(opts)
+        optional_responses: validate_optional_response_values(opts),
+        sndbuf: sndbuf_opts[:sndbuf]
       }
 
       {:ok, conn}
@@ -667,6 +670,10 @@ defmodule Mint.HTTP1 do
     %{conn | proxy_headers: headers}
   end
 
+  @impl true
+  def get_send_window(%__MODULE__{streaming_request: %{ref: ref}, sndbuf: sndbuf}, ref),
+    do: sndbuf
+
   ## Helpers
 
   defp decode(:status, %{request: request} = conn, data, responses) do

lib/mint/http2.ex

@@ -1039,6 +1039,10 @@ defmodule Mint.HTTP2 do
     %{conn | proxy_headers: headers}
   end
 
+  @impl true
+  def get_send_window(%__MODULE__{} = conn, ref),
+    do: min(get_window_size(conn, :connection), get_window_size(conn, {:request, ref}))
+
   ## Helpers
 
   defp handle_closed(conn) do

lib/mint/unsafe_proxy.ex

@@ -199,4 +199,9 @@ defmodule Mint.UnsafeProxy do
   def put_proxy_headers(%__MODULE__{}, _headers) do
     raise "invalid function for proxy unsafe proxy connections"
   end
+
+  @impl true
+  def get_send_window(%__MODULE__{module: module, state: state}, ref) do
+    module.get_send_window(state, ref)
+  end
 end

test/http_test.exs

@@ -1,4 +1,44 @@
 defmodule Mint.HTTPTest do
   use ExUnit.Case, async: true
   doctest Mint.HTTP
+
+  alias Mint.{HTTP, HTTP1.TestServer}
+
+  setup do
+    {:ok, port, server_ref} = TestServer.start()
+    assert {:ok, conn} = HTTP.connect(:http, "localhost", port)
+    assert_receive {^server_ref, server_socket}
+
+    [conn: conn, server_socket: server_socket]
+  end
+
+  describe "get_send_window/2" do
+    test "returns a positive integer for an active streaming request", %{conn: conn} do
+      {:ok, conn, ref} = HTTP.request(conn, "GET", "/", [], :stream)
+
+      send_window = HTTP.get_send_window(conn, ref)
+      assert is_integer(send_window)
+      assert send_window > 0
+    end
+  end
+
+  describe "next_body_chunk_size/3" do
+    test "returns body size when body is smaller than send window", %{conn: conn} do
+      {:ok, conn, ref} = HTTP.request(conn, "GET", "/", [], :stream)
+      small_body = "hello"
+      assert HTTP.next_body_chunk_size(conn, ref, small_body) == byte_size(small_body)
+    end
+
+    test "returns send window when body is larger than send window", %{conn: conn} do
+      {:ok, conn, ref} = HTTP.request(conn, "GET", "/", [], :stream)
+      send_window = HTTP.get_send_window(conn, ref)
+      large_body = :binary.copy(<<0>>, send_window + 1000)
+      assert HTTP.next_body_chunk_size(conn, ref, large_body) == send_window
+    end
+
+    test "returns 0 for empty body", %{conn: conn} do
+      {:ok, conn, ref} = HTTP.request(conn, "GET", "/", [], :stream)
+      assert HTTP.next_body_chunk_size(conn, ref, "") == 0
+    end
+  end
 end

test/mint/http1/conn_test.exs

@@ -1073,6 +1073,21 @@ defmodule Mint.HTTP1Test do
     {:ok, conn, responses}
   end
 
+  describe "get_send_window/2" do
+    test "returns a positive integer for an active streaming request", %{conn: conn} do
+      {:ok, conn, ref} = HTTP1.request(conn, "GET", "/", [], :stream)
+
+      send_window = HTTP1.get_send_window(conn, ref)
+      assert is_integer(send_window)
+      assert send_window > 0
+    end
+
+    test "returns the cached sndbuf value", %{conn: conn} do
+      {:ok, conn, ref} = HTTP1.request(conn, "GET", "/", [], :stream)
+      assert HTTP1.get_send_window(conn, ref) == conn.sndbuf
+    end
+  end
+
   @mint_user_agent "mint/#{Mix.Project.config()[:version]}"
   defp mint_user_agent, do: @mint_user_agent
 end

test/mint/http2/conn_test.exs

@@ -1772,6 +1772,29 @@ defmodule Mint.HTTP2Test do
         HTTP2.get_window_size(conn, {:request, make_ref()})
       end
     end
+
+    test "get_send_window/2 returns the minimum of connection and request window sizes",
+         %{conn: conn} do
+      {conn, ref} = open_request(conn, :stream)
+
+      send_window = HTTP2.get_send_window(conn, ref)
+      conn_window = HTTP2.get_window_size(conn, :connection)
+      request_window = HTTP2.get_window_size(conn, {:request, ref})
+
+      assert send_window == min(conn_window, request_window)
+    end
+
+    test "get_send_window/2 decreases after streaming body data", %{conn: conn} do
+      {conn, ref} = open_request(conn, :stream)
+
+      initial_send_window = HTTP2.get_send_window(conn, ref)
+      assert initial_send_window > 0
+
+      body_chunk = "hello"
+      {:ok, conn} = HTTP2.stream_request_body(conn, ref, body_chunk)
+
+      assert HTTP2.get_send_window(conn, ref) == initial_send_window - byte_size(body_chunk)
+    end
   end
 
   describe "settings" do