Add support for async requests

.github/workflows/elixir.yml

@@ -5,59 +5,62 @@ on:
     branches:
       - main
 jobs:
- mix_test:
-   strategy:
-     fail-fast: false
-     matrix:
-       include:
-         - pair:
-             elixir: '1.7.4'
-             otp: '22.x'
-         - pair:
-             elixir: '1.14.1'
-             otp: '25.x'
-           lint: lint
-
-   runs-on: ubuntu-20.04
-   steps:
-     - uses: actions/checkout@v2
-     - uses: actions/cache@v2
-       with:
-         path: deps
-         key: deps-${{ runner.os }}-${{ matrix.pair.otp }}-${{ matrix.pair.elixir }}-${{ hashFiles('**/mix.lock') }}
-         restore-keys: deps-${{ runner.os }}-${{ matrix.pair.otp }}-${{ matrix.pair.elixir }}-
-     - uses: actions/cache@v2
-       with:
-         path: _build
-         key: build-${{ runner.os }}-${{ matrix.pair.otp }}-${{ matrix.pair.elixir }}-${{ hashFiles('**/mix.lock') }}
-         restore-keys: build-${{ runner.os }}-${{ matrix.pair.otp }}-${{ matrix.pair.elixir }}-
-     - uses: erlef/setup-beam@v1
-       with:
-         otp-version: ${{matrix.pair.otp}}
-         elixir-version: ${{matrix.pair.elixir}}
-
-     - name: Install Dependencies
-       run: mix deps.get
-
-     - run: mix format --check-formatted
-       if: ${{ matrix.lint }}
-
-     - run: mix deps.unlock --check-unused
-       if: ${{ matrix.lint }}
-
-     - run: mix deps.compile
-
-     - run: mix compile --warnings-as-errors
-       if: ${{ matrix.lint }}
-
-     - name: Run Credo
-       run: mix credo
-       if: ${{ matrix.lint }}
-
-     - name: Run Tests
-       run: mix test
-       if: ${{ ! matrix.lint }}
-
-     - name: Run Tests
-       run: mix test --warnings-as-errors
-       if: ${{ matrix.lint }}
+  mix_test:
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - pair:
+              elixir: '1.7.4'
+              otp: '22.x'
+          - pair:
+              elixir: '1.14.5'
+              otp: '25.x'
+            lint: lint
+          - pair:
+              elixir: '1.14.5'
+              otp: '26'
+
+    runs-on: ubuntu-20.04
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/cache@v2
+        with:
+          path: deps
+          key: deps-${{ runner.os }}-${{ matrix.pair.otp }}-${{ matrix.pair.elixir }}-${{ hashFiles('**/mix.lock') }}
+          restore-keys: deps-${{ runner.os }}-${{ matrix.pair.otp }}-${{ matrix.pair.elixir }}-
+      - uses: actions/cache@v2
+        with:
+          path: _build
+          key: build-${{ runner.os }}-${{ matrix.pair.otp }}-${{ matrix.pair.elixir }}-${{ hashFiles('**/mix.lock') }}
+          restore-keys: build-${{ runner.os }}-${{ matrix.pair.otp }}-${{ matrix.pair.elixir }}-
+      - uses: erlef/setup-beam@v1
+        with:
+          otp-version: ${{matrix.pair.otp}}
+          elixir-version: ${{matrix.pair.elixir}}
+
+      - name: Install Dependencies
+        run: mix deps.get
+
+      - run: mix format --check-formatted
+        if: ${{ matrix.lint }}
+
+      - run: mix deps.unlock --check-unused
+        if: ${{ matrix.lint }}
+
+      - run: mix deps.compile
+
+      - run: mix compile --warnings-as-errors
+        if: ${{ matrix.lint }}
+
+      - name: Run Credo
+        run: mix credo
+        if: ${{ matrix.lint }}
+
+      - name: Run Tests
+        run: mix test
+        if: ${{ ! matrix.lint }}
+
+      - name: Run Tests
+        run: mix test --warnings-as-errors
+        if: ${{ matrix.lint }}

lib/finch.ex

@@ -6,6 +6,7 @@ defmodule Finch do
              |> Enum.fetch!(1)
 
   alias Finch.{PoolManager, Request, Response}
+  require Finch.Pool
 
   use Supervisor
 
@@ -83,6 +84,19 @@ defmodule Finch do
   """
   @type name() :: atom()
 
+  @typedoc """
+  Options used by request functions.
+  """
+  @type request_opts() :: [
+          {:pool_timeout, pos_integer()}
+          | {:receive_timeout, pos_integer()}
+        ]
+
+  @typedoc """
+  The reference used to identify a request sent using `async_request/3`.
+  """
+  @opaque request_ref() :: Finch.Pool.request_ref()
+
   @typedoc """
   The stream function given to `stream/5`.
   """
@@ -263,14 +277,9 @@ defmodule Finch do
 
   ## Options
 
-    * `:pool_timeout` - This timeout is applied when we check out a connection from the pool.
-      Default value is `5_000`.
-
-    * `:receive_timeout` - The maximum time to wait for a response before returning an error.
-      Default value is `15_000`.
-
+  Shares the same options as `request/3`.
   """
-  @spec stream(Request.t(), name(), acc, stream(acc), keyword) ::
+  @spec stream(Request.t(), name(), acc, stream(acc), request_opts()) ::
           {:ok, acc} | {:error, Exception.t()}
         when acc: term()
   def stream(%Request{} = req, name, acc, fun, opts \\ []) when is_function(fun, 2) do
@@ -280,20 +289,10 @@ defmodule Finch do
   end
 
   defp __stream__(%Request{} = req, name, acc, fun, opts) when is_function(fun, 2) do
-    shp = build_shp(req)
-    {pool, pool_mod} = PoolManager.get_pool(name, shp)
+    {pool, pool_mod} = get_pool(req, name)
     pool_mod.request(pool, req, acc, fun, opts)
   end
 
-  defp build_shp(%Request{scheme: scheme, unix_socket: unix_socket})
-       when is_binary(unix_socket) do
-    {scheme, {:local, unix_socket}, 0}
-  end
-
-  defp build_shp(%Request{scheme: scheme, host: host, port: port}) do
-    {scheme, host, port}
-  end
-
   @doc """
   Sends an HTTP request and returns a `Finch.Response` struct.
 
@@ -306,7 +305,7 @@ defmodule Finch do
       Default value is `15_000`.
 
   """
-  @spec request(Request.t(), name(), keyword()) ::
+  @spec request(Request.t(), name(), request_opts()) ::
           {:ok, Response.t()}
           | {:error, Exception.t()}
   def request(req, name, opts \\ [])
@@ -351,12 +350,84 @@ defmodule Finch do
 
   See `request/3` for more detailed information.
   """
-  @spec request!(Request.t(), name(), keyword()) ::
+  @spec request!(Request.t(), name(), request_opts()) ::
           Response.t()
   def request!(%Request{} = req, name, opts \\ []) do
     case request(req, name, opts) do
       {:ok, resp} -> resp
       {:error, exception} -> raise exception
     end
   end
+
+  @doc """
+  Sends an HTTP request asynchronously, returning a request reference.
+
+  ## Responses
+
+  Response information is sent to the calling process as it is received
+  in `{ref, response}` tuples. Responses include:
+
+    * `{:status, status}` - HTTP response status
+    * `{:headers, headers}` - HTTP response headers
+    * `{:data, data}` - section of the HTTP response body
+    * `{:error, exception}` - an error occurred during the request
+    * `:done` - request has completed successfully
+
+  On a successful request, a single `:status` message will be followed
+  by a single `:headers` message, after which one or more `:data`
+  messages may be sent. If trailing headers are present, it is possible
+  that a final `:headers` message may be sent. A `:done` or an `:error`
+  message indicates that the request has succeeded or failed and no
+  further messages are expected.
+
+  ## Example
+
+      iex> Stream.resource(
+      ...>   fn ->
+      ...>     Finch.build(:get, "https://httpbin.org/stream/5")
+      ...>     |> Finch.async_request(MyFinch)
+      ...>   end,
+      ...>   fn ref ->
+      ...>     receive do
+      ...>       {^ref, :done} -> {:halt, ref}
+      ...>       {^ref, response} -> {[response], ref}
+      ...>     end
+      ...>   end,
+      ...>   fn _ref -> :ok end
+      ...> ) |> Enum.to_list()
+      [
+        {:status, 200},
+        {:headers, [...]},
+        {:data, "..."},
+        ...
+        :done
+      ]
+
+  ## Options
+
+  Shares the same options as `request/3`.
+  """
+  @spec async_request(Request.t(), name(), request_opts()) :: request_ref()
+  def async_request(%Request{} = req, name, opts \\ []) do
+    {pool, pool_mod} = get_pool(req, name)
+    pool_mod.async_request(pool, req, opts)
+  end
+
+  @doc """
+  Cancels a request sent with `async_request/3`.
+  """
+  @spec cancel_async_request(request_ref()) :: :ok
+  def cancel_async_request(request_ref) when Finch.Pool.is_request_ref(request_ref) do
+    {_ref, _pool, pool_mod, _state} = request_ref
+    pool_mod.cancel_async_request(request_ref)
+  end
+
+  defp get_pool(%Request{scheme: scheme, unix_socket: unix_socket}, name)
+       when is_binary(unix_socket) do
+    PoolManager.get_pool(name, {scheme, {:local, unix_socket}, 0})
+  end
+
+  defp get_pool(%Request{scheme: scheme, host: host, port: port}, name) do
+    PoolManager.get_pool(name, {scheme, host, port})
+  end
 end

lib/finch/http1/pool.ex

@@ -74,6 +74,56 @@ defmodule Finch.HTTP1.Pool do
     end
   end
 
+  @impl Finch.Pool
+  def async_request(pool, req, opts) do
+    owner = self()
+
+    pid =
+      spawn(fn ->
+        monitor = Process.monitor(owner)
+        request_ref = receive_next_within!(10)
+
+        case request(pool, req, {owner, monitor, request_ref}, &send_async_response/2, opts) do
+          {:ok, _} -> send(owner, {request_ref, :done})
+          {:error, error} -> send(owner, {request_ref, {:error, error}})
+        end
+      end)
+
+    request_ref = {make_ref(), pool, __MODULE__, pid}
+    send(pid, request_ref)
+  end
+
+  defp receive_next_within!(timeout) do
+    receive do
+      value -> value
+    after
+      timeout -> raise "no message received within #{timeout}"
+    end
+  end
+
+  defp send_async_response(response, {owner, monitor, request_ref}) do
+    if process_down?(monitor) do
+      exit(:shutdown)
+    end
+
+    send(owner, {request_ref, response})
+    {owner, monitor, request_ref}
+  end
+
+  defp process_down?(monitor) do
+    receive do
+      {:DOWN, ^monitor, _, _, _} -> true
+    after
+      0 -> false
+    end
+  end
+
+  @impl Finch.Pool
+  def cancel_async_request({_, _, _, pid}) do
+    Process.exit(pid, :shutdown)
+    :ok
+  end
+
   @impl NimblePool
   def init_pool({registry, shp, opts}) do
     # Register our pool with our module name as the key. This allows the caller