Support processing HTTP/1.1 headers as they arrive

lib/mint/http.ex

@@ -844,8 +844,11 @@ defmodule Mint.HTTP do
     * `{:headers, request_ref, headers}` - returned when the server replied
       with a list of headers. Headers are in the form `{header_name, header_value}`
       with `header_name` and `header_value` being strings. A single `:headers` response
-      will come after the `:status` response. A single `:headers` response may come
-      after all the `:data` responses if **trailer headers** are present.
+      will come after the `:status` response and a single `:headers` response may come
+      after all the `:data` responses if **trailer headers** are present unless
+      `:stream_headers` is enabled (only available for HTTP/1.1 connections), in which
+      case any number of `:headers` responses (including none) may come after the
+      `:status` response and/or after all the `:data` responses.
 
     * `{:data, request_ref, binary}` - returned when the server replied with
       a chunk of response body (as a binary). The request shouldn't be considered done

lib/mint/http1.ex

@@ -104,6 +104,7 @@ defmodule Mint.HTTP1 do
     :case_sensitive_headers,
     :skip_target_validation,
     :max_header_list_size,
+    :stream_headers,
     requests: :queue.new(),
     state: :closed,
     buffer: "",
@@ -150,6 +151,11 @@ defmodule Mint.HTTP1 do
       bytes allowed in a response header section or chunked trailer section. This includes
       header names, values, and line delimiters. Defaults to 256 KiB. *Available since 1.9.2*.
 
+    * `:stream_headers` - (`t:boolean/0`) if set to `true`, response headers and trailer headers
+      will be emitted as they are parsed, rather than buffered until the complete header section
+      is received. When enabled, you may receive multiple `{:headers, ref, headers}` responses
+      for a single request. Defaults to `false`. *Available since v1.10.0*.
+
   """
   @spec connect(Types.scheme(), Types.address(), :inet.port_number(), keyword()) ::
           {:ok, t()} | {:error, Types.error()}
@@ -238,6 +244,7 @@ defmodule Mint.HTTP1 do
         case_sensitive_headers: Keyword.get(opts, :case_sensitive_headers, false),
         skip_target_validation: Keyword.get(opts, :skip_target_validation, false),
         max_header_list_size: max_header_list_size,
+        stream_headers: Keyword.get(opts, :stream_headers, false),
         optional_responses: validate_optional_response_values(opts)
       }
 
@@ -758,7 +765,12 @@ defmodule Mint.HTTP1 do
       {:ok, :eof, rest} ->
         case add_header_bytes(conn, request, byte_size(data) - byte_size(rest)) do
           {:ok, request} ->
-            responses = [{:headers, request.ref, Enum.reverse(headers)} | responses]
+            responses =
+              case {conn.stream_headers, headers} do
+                {true, []} -> responses
+                {_, headers} -> [{:headers, request.ref, Enum.reverse(headers)} | responses]
+              end
+
             request = %{request | state: :body, headers_buffer: [], headers_size: 0}
             conn = %{conn | buffer: "", request: request}
             decode(:body, conn, rest, responses)
@@ -770,7 +782,19 @@ defmodule Mint.HTTP1 do
       :more ->
         case check_header_section_size(conn, request.headers_size + byte_size(data)) do
           :ok ->
-            request = %{request | headers_buffer: headers}
+            {responses, headers_buffer} =
+              cond do
+                not conn.stream_headers ->
+                  {responses, headers}
+
+                headers != [] ->
+                  {[{:headers, request.ref, Enum.reverse(headers)} | responses], []}
+
+                true ->
+                  {responses, []}
+              end
+
+            request = %{request | headers_buffer: headers_buffer}
             conn = %{conn | buffer: data, request: request}
             {:ok, conn, responses}
 
@@ -950,7 +974,24 @@ defmodule Mint.HTTP1 do
       :more ->
         case check_header_section_size(conn, conn.request.headers_size + byte_size(data)) do
           :ok ->
-            request = %{conn.request | body: {:chunked, :trailer}, headers_buffer: headers}
+            {responses, headers_buffer} =
+              cond do
+                not conn.stream_headers ->
+                  {responses, headers}
+
+                headers != [] ->
+                  responses =
+                    headers
+                    |> Headers.remove_unallowed_trailer()
+                    |> add_trailer_headers(conn.request.ref, responses)
+
+                  {responses, []}
+
+                true ->
+                  {responses, []}
+              end
+
+            request = %{conn.request | body: {:chunked, :trailer}, headers_buffer: headers_buffer}
             conn = %{conn | buffer: data, request: request}
             {:ok, conn, responses}
 

test/mint/http1/conn_test.exs

@@ -1273,6 +1273,106 @@ defmodule Mint.HTTP1Test do
     end
   end
 
+  describe "stream_headers option" do
+    setup %{port: port} do
+      assert {:ok, conn} = HTTP1.connect(:http, "localhost", port, stream_headers: true)
+      assert_receive {_server_ref, server_socket}
+      [conn: conn, server_socket: server_socket]
+    end
+
+    test "emits all complete headers from a chunk at once when stream_headers is true", %{conn: conn} do
+      {:ok, conn, ref} = HTTP1.request(conn, "GET", "/", [], nil)
+
+      assert {:ok, conn, [_status]} = HTTP1.stream(conn, {:tcp, conn.socket, "HTTP/1.1 200 OK\r\n"})
+
+      # Send two complete headers plus start of third header in one chunk
+      # This should emit the two complete headers together
+      assert {:ok, conn, [headers1]} =
+               HTTP1.stream(conn, {:tcp, conn.socket, "Foo: Bar\r\nBaz: Boz\r\nQux"})
+
+      assert {:headers, ^ref, [{"foo", "Bar"}, {"baz", "Boz"}]} = headers1
+
+      # Complete the third header and end headers section
+      assert {:ok, _conn, [headers2]} =
+               HTTP1.stream(conn, {:tcp, conn.socket, ": Quux\r\n\r\n"})
+
+      assert {:headers, ^ref, [{"qux", "Quux"}]} = headers2
+    end
+
+    test "emits multiple headers from one packet together", %{conn: conn} do
+      {:ok, conn, ref} = HTTP1.request(conn, "GET", "/", [], nil)
+
+      assert {:ok, conn, [_status]} = HTTP1.stream(conn, {:tcp, conn.socket, "HTTP/1.1 200 OK\r\n"})
+
+      assert {:ok, _conn, responses} =
+               HTTP1.stream(conn, {:tcp, conn.socket, "Foo: Bar\r\nBaz: Boz\r\n\r\n"})
+
+      assert [headers] = responses
+      assert {:headers, ^ref, [{"foo", "Bar"}, {"baz", "Boz"}]} = headers
+    end
+
+    test "handles partial headers with streaming", %{conn: conn} do
+      {:ok, conn, ref} = HTTP1.request(conn, "GET", "/", [], nil)
+
+      assert {:ok, conn, [_status]} = HTTP1.stream(conn, {:tcp, conn.socket, "HTTP/1.1 200 OK\r\n"})
+
+      # Send first header with partial second header
+      assert {:ok, conn, [header1]} = HTTP1.stream(conn, {:tcp, conn.socket, "Foo: Bar\r\nB"})
+      assert {:headers, ^ref, [{"foo", "Bar"}]} = header1
+
+      # Complete second header and end headers
+      assert {:ok, _conn, [header2]} = HTTP1.stream(conn, {:tcp, conn.socket, "az: Boz\r\n\r\n"})
+      assert {:headers, ^ref, [{"baz", "Boz"}]} = header2
+    end
+
+    test "streams trailer headers from same chunk together", %{conn: conn} do
+      {:ok, conn, ref} = HTTP1.request(conn, "GET", "/", [], nil)
+
+      assert {:ok, conn, [_status]} =
+               HTTP1.stream(conn, {:tcp, conn.socket, "HTTP/1.1 200 OK\r\n"})
+
+      assert {:ok, conn, [_headers]} =
+               HTTP1.stream(conn, {:tcp, conn.socket, "Transfer-Encoding: chunked\r\n\r\n"})
+
+      assert {:ok, conn, responses} = HTTP1.stream(conn, {:tcp, conn.socket, "5\r\nhello\r\n"})
+      assert [{:data, ^ref, "hello"}] = responses
+
+      # Send last chunk and trailer headers in one chunk
+      assert {:ok, _conn, responses} =
+               HTTP1.stream(
+                 conn,
+                 {:tcp, conn.socket, "0\r\nX-Trailer-1: value1\r\nX-Trailer-2: value2\r\n\r\n"}
+               )
+
+      assert [trailers, done] = responses
+      assert {:headers, ^ref, [{"x-trailer-1", "value1"}, {"x-trailer-2", "value2"}]} = trailers
+      assert {:done, ^ref} = done
+    end
+
+    test "filters unallowed trailer headers when streaming", %{conn: conn} do
+      {:ok, conn, ref} = HTTP1.request(conn, "GET", "/", [], nil)
+
+      assert {:ok, conn, [_status]} =
+               HTTP1.stream(conn, {:tcp, conn.socket, "HTTP/1.1 200 OK\r\n"})
+
+      assert {:ok, conn, [_headers]} =
+               HTTP1.stream(conn, {:tcp, conn.socket, "Transfer-Encoding: chunked\r\n\r\n"})
+
+      # Send last chunk with allowed and unallowed trailer headers
+      assert {:ok, _conn, responses} =
+               HTTP1.stream(
+                 conn,
+                 {:tcp, conn.socket,
+                  "0\r\nContent-Length: 100\r\nX-Custom-Trailer: allowed\r\n\r\n"}
+               )
+
+      # Content-Length should be filtered out, only X-Custom-Trailer should appear
+      assert [trailer, done] = responses
+      assert {:headers, ^ref, [{"x-custom-trailer", "allowed"}]} = trailer
+      assert {:done, ^ref} = done
+    end
+  end
+
   @mint_user_agent "mint/#{Mix.Project.config()[:version]}"
   defp mint_user_agent, do: @mint_user_agent
 end