Current section

Files

Jump to
electric_client lib electric client fetch pool.ex
Raw

lib/electric/client/fetch/pool.ex

defmodule Electric.Client.Fetch.Pool do
@moduledoc """
Coaleses requests so that multiple client instances making the same
(potentially long-polling) request will all use the same request process.
"""
alias Electric.Client
alias Electric.Client.Fetch
require Logger
@callback request(Client.t(), Fetch.Request.t(), opts :: Keyword.t()) ::
Fetch.Response.t() | {:error, Fetch.Response.t() | term()}
@behaviour __MODULE__
@impl Electric.Client.Fetch.Pool
def request(%Client{} = client, %Fetch.Request{} = request, opts) do
request_id = request_id(client, request)
# The monitor process is unique to the request and launches the actual
# request as a linked process.
#
# This coalesces requests, so no matter how many simultaneous
# clients we have, we only ever make one request to the backend.
{:ok, monitor_pid} = start_monitor(request_id, request, client)
try do
ref = Fetch.Monitor.register(monitor_pid, self())
Fetch.Monitor.wait(ref)
catch
:exit, {reason, _} ->
Logger.debug(fn ->
"Request process ended with reason #{inspect(reason)} before we could register. Re-attempting."
end)
request(client, request, opts)
end
end
defp start_monitor(request_id, request, client) do
DynamicSupervisor.start_child(
Electric.Client.RequestSupervisor,
{Electric.Client.Fetch.Monitor, {request_id, request, client}}
)
|> return_existing()
end
defp return_existing({:ok, pid}), do: {:ok, pid}
defp return_existing({:error, {:already_started, pid}}), do: {:ok, pid}
defp return_existing(error), do: error
defp request_id(%Client{fetch: {fetch_impl, _}}, %Fetch.Request{} = request) do
{
fetch_impl,
URI.to_string(request.endpoint),
request.headers,
Fetch.Request.params(request)
}
end
end