HTTP1 stream

lib/finch.ex

@@ -307,6 +307,28 @@ defmodule Finch do
     end
   end
 
+  @doc """
+  Like `stream/5`, but returns valid `Stream` structure, safe for passing between processes, but working __only__
+  on local node. Works only for HTTP1, HTTP2 not supported currently
+  """
+  @spec actual_stream(Request.t(), name(), request_opts()) ::
+          {:ok, Enumerable.t()} | {:error, Exception.t()}
+  def actual_stream(request, name, opts \\ []) do
+    {pool, pool_mod} = get_pool(request, name)
+    pool_mod.stream(pool, request, name, opts)
+  end
+
+  @doc """
+  Raising version of `actual_stream/3`
+  """
+  @spec actual_stream!(Request.t(), name(), request_opts()) :: Enumerable.t()
+  def actual_stream!(request, name, opts \\ []) do
+    case actual_stream(request, name, opts) do
+      {:ok, stream} -> stream
+      exception -> raise exception
+    end
+  end
+
   @doc """
   Builds an HTTP request to be sent with `request/3` or `stream/4`.
 

lib/finch/http1/conn.ex

@@ -100,6 +100,90 @@ defmodule Finch.HTTP1.Conn do
     end
   end
 
+  def stream(%{mint: nil} = conn, _, _, _, _, _), do: {:error, conn, "Could not connect"}
+
+  def stream(conn, req, name, ref, handler, receive_timeout, request_timeout, idle_time) do
+    full_path = Finch.Request.request_path(req)
+
+    metadata = %{request: req, name: name}
+
+    extra_measurements = %{idle_time: idle_time}
+
+    start_time = Telemetry.start(:send, metadata, extra_measurements)
+
+    try do
+      case Mint.HTTP.request(
+             conn.mint,
+             req.method,
+             full_path,
+             req.headers,
+             stream_or_body(req.body)
+           ) do
+        {:ok, mint, mint_ref} ->
+          case maybe_stream_request_body(mint, mint_ref, req.body) do
+            {:ok, mint} ->
+              Telemetry.stop(:send, start_time, metadata, extra_measurements)
+              start_time = Telemetry.start(:recv, metadata, extra_measurements)
+              resp_metadata = %{status: nil, headers: [], trailers: []}
+              timeouts = %{receive_timeout: receive_timeout, request_timeout: request_timeout}
+
+              stream =
+                Stream.resource(
+                  fn -> {[], mint, mint_ref, timeouts, resp_metadata} end,
+                  &receive_stream_response/1,
+                  fn
+                    {mint, error, resp_metadata} ->
+                      conn = %{conn | mint: mint}
+
+                      state =
+                        if open?(conn) do
+                          {:ok, conn}
+                        else
+                          :closed
+                        end
+
+                      if error do
+                        metadata = Map.merge(metadata, Map.put(resp_metadata, :error, error))
+                        Telemetry.stop(:recv, start_time, metadata, extra_measurements)
+                        raise error
+                      else
+                        metadata = Map.merge(metadata, resp_metadata)
+                        Telemetry.stop(:recv, start_time, metadata, extra_measurements)
+                      end
+
+                      send(handler, {ref, :stop, state})
+
+                    {_entries, mint, _mint_ref, _timeout, _resp_metadata} ->
+                      # In case some exception occured, we close the connection
+                      Mint.HTTP.close(mint)
+                      send(handler, {ref, :stop, :closed})
+                  end
+                )
+
+              {:ok, stream}
+
+            {:error, mint, error} ->
+              handle_request_error(
+                conn,
+                mint,
+                error,
+                metadata,
+                start_time,
+                extra_measurements
+              )
+          end
+
+        {:error, mint, error} ->
+          handle_request_error(conn, mint, error, metadata, start_time, extra_measurements)
+      end
+    catch
+      kind, error ->
+        close(conn)
+        Telemetry.exception(:recv, start_time, kind, error, __STACKTRACE__, metadata)
+        :erlang.raise(kind, error, __STACKTRACE__)
+    end
+  end
+
   def request(%{mint: nil} = conn, _, _, _, _, _, _, _), do: {:error, conn, "Could not connect"}
 
   def request(conn, req, acc, fun, name, receive_timeout, request_timeout, idle_time) do
@@ -368,4 +452,86 @@ defmodule Finch.HTTP1.Conn do
         {:error, mint, error, resp_metadata}
     end
   end
+
+  defp receive_stream_response({
+         [{:done, _minf_ref} | _],
+         mint,
+         _mint_ref,
+         _timeouts,
+         resp_metadata
+       }) do
+    {:halt, {mint, nil, resp_metadata}}
+  end
+
+  defp receive_stream_response({
+         _,
+         mint,
+         _mint_ref,
+         timeouts,
+         resp_metadata
+       })
+       when timeouts.request_timeout < 0 do
+    {:ok, mint} = Mint.HTTP1.close(mint)
+    {:halt, {mint, %Mint.TransportError{reason: :timeout}, resp_metadata}}
+  end
+
+  defp receive_stream_response({
+         [],
+         mint,
+         mint_ref,
+         timeouts,
+         resp_metadata
+       }) do
+    start_time = System.monotonic_time(:millisecond)
+    %{} = timeouts
+
+    case Mint.HTTP.recv(mint, 0, timeouts.receive_timeout) do
+      {:ok, mint, entries} ->
+        timeouts =
+          if is_integer(timeouts.request_timeout) do
+            elapsed_time = System.monotonic_time(:millisecond) - start_time
+            update_in(timeouts.request_timeout, &(&1 - elapsed_time))
+          else
+            timeouts
+          end
+
+        {[], {entries, mint, mint_ref, timeouts, resp_metadata}}
+
+      {:error, mint, error, _responses} ->
+        {:halt, {mint, error, resp_metadata}}
+    end
+  end
+
+  defp receive_stream_response({
+         [entry | entries],
+         mint,
+         mint_ref,
+         timeouts,
+         resp_metadata
+       }) do
+    case entry do
+      {key, ^mint_ref, value} when key in ~w[status headers data]a ->
+        resp_metadata =
+          case key do
+            :headers ->
+              Map.update(resp_metadata, :headers, value, &(&1 ++ value))
+
+            _ ->
+              resp_metadata
+          end
+
+        acc = {
+          entries,
+          mint,
+          mint_ref,
+          timeouts,
+          %{resp_metadata | status: value}
+        }
+
+        {[{key, value}], acc}
+
+      {:error, ^mint_ref, error} ->
+        {:halt, {mint, error, resp_metadata}}
+    end
+  end
 end

lib/finch/http1/pool.ex

@@ -38,6 +38,112 @@ defmodule Finch.HTTP1.Pool do
     )
   end
 
+  @impl Finch.Pool
+  def stream(pool, req, name, opts) do
+    pool_timeout = Keyword.get(opts, :pool_timeout, 5_000)
+    receive_timeout = Keyword.get(opts, :receive_timeout, 15_000)
+    request_timeout = Keyword.get(opts, :request_timeout, 30_000)
+
+    metadata = %{request: req, pool: pool, name: name}
+
+    start_time = Telemetry.start(:queue, metadata)
+    owner = self()
+    ref = make_ref()
+
+    holder =
+      spawn_link(fn ->
+        try do
+          NimblePool.checkout!(
+            pool,
+            :checkout,
+            fn from, {state, conn, idle_time} ->
+              Telemetry.stop(:queue, start_time, metadata, %{idle_time: idle_time})
+              send(owner, {ref, :ok, {conn, from, state, idle_time}})
+
+              receive do
+                {^ref, :stop, state} -> {:ok, state}
+              end
+            end,
+            pool_timeout
+          )
+        catch
+          :exit, data ->
+            Telemetry.exception(:queue, start_time, :exit, data, __STACKTRACE__, metadata)
+            send(owner, {ref, :exit, {data, __STACKTRACE__}})
+        end
+      end)
+
+    receive do
+      {^ref, :ok, conn_from} ->
+        {conn, from, state, idle_time} = conn_from
+
+        with {:ok, conn} <- Conn.connect(conn, name),
+             {:ok, conn} <- transfer_if_open(conn, state, from),
+             {:ok, stream} <-
+               Conn.stream(
+                 conn,
+                 req,
+                 name,
+                 ref,
+                 holder,
+                 receive_timeout,
+                 request_timeout,
+                 idle_time
+               ) do
+          {:ok, stream}
+        else
+          :closed ->
+            send(holder, {ref, :stop, state})
+            # FIXME
+            {:error, :closed}
+
+          {:error, conn, error} ->
+            state =
+              if Conn.open?(conn) do
+                {:ok, conn}
+              else
+                :closed
+              end
+
+            send(holder, {ref, :stop, state})
+            {:error, error}
+        end
+
+      {^ref, :error, reason} ->
+        {:error, reason}
+
+      {^ref, :exit, data_trace} ->
+        {data, trace} = data_trace
+
+        case data do
+          {:timeout, {NimblePool, :checkout, _affected_pids}} ->
+            # Provide helpful error messages for known errors
+            reraise(
+              """
+              Finch was unable to provide a connection within the timeout due to excess queuing \
+              for connections. Consider adjusting the pool size, count, timeout or reducing the \
+              rate of requests if it is possible that the downstream service is unable to keep up \
+              with the current rate.
+              """,
+              trace
+            )
+
+          _ ->
+            exit(data)
+        end
+    after
+      pool_timeout + request_timeout ->
+        # Cleanup late messages
+        receive do
+          {^ref, _, _} -> :ok
+        after
+          0 -> :ok
+        end
+
+        raise "Has not received message from pool yet"
+    end
+  end
+
   @impl Finch.Pool
   def request(pool, req, acc, fun, name, opts) do
     pool_timeout = Keyword.get(opts, :pool_timeout, 5_000)
@@ -280,16 +386,22 @@ defmodule Finch.HTTP1.Pool do
   def handle_cancelled(:queued, _pool_state), do: :ok
 
   defp transfer_if_open(conn, state, {pid, _} = from) do
+    transfer_if_open(conn, state, from, pid)
+  end
+
+  defp transfer_if_open(conn, state, from, pid) do
     if Conn.open?(conn) do
-      if state == :fresh do
-        NimblePool.update(from, conn)
+      case state do
+        :fresh ->
+          NimblePool.update(from, conn)
 
-        case Conn.transfer(conn, pid) do
-          {:ok, conn} -> {:ok, conn}
-          {:error, _, _} -> :closed
-        end
-      else
-        {:ok, conn}
+          case Conn.transfer(conn, pid) do
+            {:ok, conn} -> {:ok, conn}
+            {:error, _, _} -> :closed
+          end
+
+        _ ->
+          {:ok, conn}
       end
     else
       :closed

lib/finch/http2/pool.ex

@@ -27,6 +27,11 @@ defmodule Finch.HTTP2.Pool do
     }
   end
 
+  @impl Finch.Pool
+  def stream(_pool, _request, _name, _opts) do
+    {:error, %RuntimeError{message: "Streaming is not supported for HTTP2"}}
+  end
+
   # Call the pool with the request. The pool will multiplex multiple requests
   # and stream the result set back to the calling process using `send`
   @impl Finch.Pool

lib/finch/pool.ex

@@ -4,6 +4,9 @@ defmodule Finch.Pool do
 
   @type request_ref :: {pool_mod :: module(), cancel_ref :: term()}
 
+  @callback stream(pid(), Finch.Request.t(), finch_name :: atom(), list()) ::
+              {:ok, Stream.t()} | {:error, Exception.t()}
+
   @callback request(
               pid(),
               Finch.Request.t(),