Current section

Files

Jump to
x3m_system lib aggregate_registry.ex
Raw

lib/aggregate_registry.ex

defmodule X3m.System.AggregateRegistry do
@moduledoc """
Keeps track of registered aggregate pids.
"""
require Logger
use GenServer
def name(aggregate_mod),
do: Module.concat(aggregate_mod, PidRegistry)
## Client API
@doc """
Starts process registry
"""
def start_link(name),
do: GenServer.start_link(__MODULE__, name, name: name)
@doc """
Returns `true` if `key` is already registered in `server`, `false` otherwise.
"""
def has_key?(server, key), do: GenServer.call(server, {:has_key?, key})
@doc """
Looks up the process pid for `id` stored in `server`.
Returns `{:ok, pid}` if the one exists, `:error` otherwise.
"""
def get(server, key),
do: server |> pid_table |> _get(key)
@doc """
Ensures there is a `pid` associated with `key` in `server`.
Returns :ok once when process is successfully registered
"""
def register(server, key, pid),
do: GenServer.call(server, {:register, key, pid})
## Server Callbacks
def init(name) do
name
|> pid_table
|> :ets.new([:named_table, read_concurrency: true])
name
|> ref_table
|> :ets.new([:named_table, read_concurrency: true])
{:ok, %{pid_table: pid_table(name), ref_table: ref_table(name)}}
end
## Private functions
defp _get(table, key) do
case :ets.lookup(table, key) do
[{^key, val}] -> {:ok, val}
[] -> :error
end
end
defp pid_table(name), do: Module.concat(name, Pids)
defp ref_table(name), do: Module.concat(name, Refs)
def handle_call({:has_key?, key}, _from, state) do
{:reply, Map.has_key?(state.processes, key), state}
end
def handle_call({:get, key}, _from, state) do
{:reply, Map.fetch(state.processes, key), state}
end
def handle_call({:register, key, pid}, _from, state) do
result =
case _get(state.pid_table, key) do
{:ok, pid} ->
{:error, :key_already_registered, pid}
:error ->
ref = Process.monitor(pid)
true = :ets.insert(state.pid_table, {key, pid})
true = :ets.insert(state.ref_table, {ref, key})
:ok
end
{:reply, result, state}
end
def handle_info({:DOWN, ref, :process, pid, reason}, state) do
case _get(state.ref_table, ref) do
{:ok, key} ->
Logger.warning("Process #{inspect(pid)} went down: #{inspect(reason)}")
:ets.delete(state.ref_table, ref)
:ets.delete(state.pid_table, key)
end
{:noreply, state}
end
def handle_info(_msg, state), do: {:noreply, state}
end