Packages
electric_client
0.2.4-pre-1
0.10.3
0.10.2
0.10.1
0.10.1-beta-1
0.10.0
0.9.5-beta-1
0.9.4
0.9.4-beta-1
0.9.3
0.9.2
0.9.1
0.9.0
0.8.3
0.8.3-beta-1
0.8.2
0.8.1
0.8.0
0.8.0-beta-1
0.7.3
0.7.2
0.7.1
0.7.0
0.6.5
0.6.5-beta-5
0.6.5-beta-4
0.6.5-beta-3
0.6.5-beta-2
0.6.5-beta-1
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.0
0.5.0-beta-1
0.4.1
0.4.0
0.3.2
0.3.1
0.3.0
0.3.0-beta.4
0.3.0-beta.3
0.3.0-beta.2
0.2.6-pre-1
retired
0.2.6-beta.1
0.2.6-beta.0
0.2.5
0.2.4
0.2.4-pre-8
0.2.4-pre-7
0.2.4-pre-6
0.2.4-pre-5
0.2.4-pre-4
0.2.4-pre-3
0.2.4-pre-2
0.2.4-pre-1
0.2.3
0.2.3-rc-1
0.2.2
0.2.2-rc-1
0.2.1
0.2.1-rc-3
0.2.1-rc-2
0.2.1-rc-1
0.2.0
0.1.2
0.1.1
0.1.0
0.1.0-dev-9
0.1.0-dev-8
0.1.0-dev-7
0.1.0-dev-6
0.1.0-dev-5
0.1.0-dev-4
0.1.0-dev-3
0.1.0-dev-2
0.1.0-dev-17
0.1.0-dev-16
0.1.0-dev-15
0.1.0-dev-14
0.1.0-dev-13
0.1.0-dev-12
0.1.0-dev-11
0.1.0-dev-10
0.1.0-dev
Elixir client for ElectricSQL
Current section
Files
Jump to
Current section
Files
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)
# register this pid before making the request to avoid race conditions for
# very fast responses
{:ok, monitor_pid} = start_monitor(request_id)
try do
ref = Fetch.Monitor.register(monitor_pid, self())
{:ok, _request_pid} = start_request(request_id, request, client, monitor_pid)
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_request(request_id, request, client, monitor_pid) do
DynamicSupervisor.start_child(
Electric.Client.RequestSupervisor,
{Fetch.Request, {request_id, request, client, monitor_pid}}
)
|> return_existing()
end
defp start_monitor(request_id) do
DynamicSupervisor.start_child(
Electric.Client.RequestSupervisor,
{Electric.Client.Fetch.Monitor, request_id}
)
|> 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{shape_handle: nil} = request) do
%{endpoint: endpoint, shape: shape_definition} = request
{fetch_impl, URI.to_string(endpoint), shape_definition}
end
defp request_id(%Client{fetch: {fetch_impl, _}}, %Fetch.Request{} = request) do
%{endpoint: endpoint, offset: offset, live: live, shape_handle: shape_handle} = request
{fetch_impl, URI.to_string(endpoint), shape_handle, Client.Offset.to_tuple(offset), live}
end
end