Packages
electric_client
0.1.0-dev
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/monitor.ex
defmodule Electric.Client.Fetch.Monitor do
@moduledoc false
# Companion process that registers processes listening for the result of a
# given client request.
#
# Separates the list of subscribers from the actual request process so that
# if the request process crashes the list of subscribers is retained and also
# so that registering subscribers can happen while the request process is
# blocked performing it's actual HTTP request.
use GenServer
require Logger
def name(request_id) do
{:via, Registry, {Electric.Client.Registry, {__MODULE__, request_id}}}
end
def child_spec(request_id) do
%{
id: {__MODULE__, request_id},
start: {__MODULE__, :start_link, [request_id]},
restart: :transient,
type: :worker
}
end
def start_link(request_id) do
GenServer.start_link(__MODULE__, request_id, name: name(request_id))
end
def register(monitor_pid, listener_pid) do
GenServer.call(monitor_pid, {:register, listener_pid})
end
def wait(ref) do
receive do
{:response, ^ref, response} -> response
end
end
def reply(pid, response) when is_pid(pid) do
GenServer.call(pid, {:reply, response})
end
@impl true
def init(request_id) do
state = %{
request_id: request_id,
subscribers: []
}
{:ok, state}
end
@impl true
def handle_call({:register, listener_pid}, _from, state) do
ref = Process.monitor(listener_pid)
Logger.debug(fn ->
[message: "Registering listener pid #{inspect(listener_pid)}", request_id: state.request_id]
end)
state = Map.update!(state, :subscribers, &[{listener_pid, ref} | &1])
{:reply, ref, state}
end
def handle_call({:reply, response}, _from, state) do
case response do
%{status: status} ->
Logger.debug(
fn ->
"Returning response #{status}"
end,
request_id: state.request_id
)
{:error, %{status: _} = response} ->
Logger.warning(
fn ->
"Request failed: #{inspect(response)}"
end,
request_id: state.request_id
)
{:error, reason} ->
Logger.error(
fn ->
"Request failed: #{inspect(reason)}"
end,
request_id: state.request_id
)
end
for {pid, ref} <- state.subscribers do
send(pid, {:response, ref, response})
end
{:stop, :normal, :ok, state}
end
@impl true
def handle_info({:DOWN, ref, :process, pid, reason}, state) do
Logger.debug(fn ->
[
message:
"#{inspect(pid)} exited with reason #{inspect(reason)}. Removing from subscribers",
request_id: state.request_id
]
end)
state =
Map.update!(state, :subscribers, fn subscribers ->
Enum.reject(subscribers, &(&1 == {pid, ref}))
end)
{:noreply, state}
end
end