diff --git a/lib/excache/clients.ex b/lib/excache/clients.ex new file mode 100644 index 0000000..3c60dcb --- /dev/null +++ b/lib/excache/clients.ex @@ -0,0 +1,164 @@ +defmodule Excache.Clients do + @moduledoc false + + use GenServer + + @name __MODULE__ + @supervisor Excache.Server.ConnectionSupervisor + @pubsubRegistry Excache.PubSub.Registry + @defaultPoll 500 + @selfKillDelay 50 + + def start_link(_arg), do: GenServer.start_link(__MODULE__, nil, name: @name) + + def snapshot, do: GenServer.call(@name, :snapshot) + + def find(pid), do: GenServer.call(@name, {:find, pid}) + + def setName(pid, name), do: GenServer.call(@name, {:setName, pid, name}) + + def kill(matcher, caller), do: GenServer.call(@name, {:kill, matcher, caller}) + + def subscriptionCount(pid), do: length(Registry.keys(@pubsubRegistry, pid)) + + def subscribedChannels do + @pubsubRegistry + |> Registry.select([{{:"$1", :_, :_}, [], [:"$1"]}]) + |> Enum.uniq() + |> length() + end + + def blocked?(pid) do + case Process.info(pid, :monitored_by) do + {:monitored_by, watchers} -> Process.whereis(Excache.Blocking) in watchers + nil -> false + end + end + + @impl true + def init(_arg) do + schedule() + + {:ok, + %{ + nextId: 1, + clients: %{}, + monitors: %{}, + connectionsReceived: 0, + closedNetIn: 0, + closedNetOut: 0, + runId: newRunId(), + startedAt: System.system_time(:millisecond) + }} + end + + @impl true + def handle_call(:snapshot, _from, state) do + state = sync(state) + {:reply, buildSnapshot(state), state} + end + + @impl true + def handle_call({:find, pid}, _from, state) do + state = sync(state) + {:reply, Map.get(state.clients, pid), state} + end + + @impl true + def handle_call({:setName, pid, name}, _from, state) do + state = sync(state) + + case Map.fetch(state.clients, pid) do + {:ok, client} -> + renamed = Map.put(state.clients, pid, %{client | name: name}) + {:reply, :ok, %{state | clients: renamed}} + + :error -> + {:reply, :error, state} + end + end + + @impl true + def handle_call({:kill, matcher, caller}, _from, state) do + state = sync(state) + victims = state.clients |> Map.values() |> Enum.filter(matcher) + {itself, others} = Enum.split_with(victims, &(&1.pid == caller)) + + Enum.each(others, fn client -> terminate(client.pid) end) + + Enum.each(itself, fn client -> + Process.send_after(self(), {:terminate, client.pid}, @selfKillDelay) + end) + + {:reply, length(victims), state} + end + + @impl true + def handle_info(:poll, state) do + schedule() + {:noreply, sync(state)} + end + + @impl true + def handle_info({:terminate, pid}, state) do + terminate(pid) + {:noreply, state} + end + + @impl true + def handle_info({:DOWN, monitor, :process, pid, _reason}, state) do + {_pid, monitors} = Map.pop(state.monitors, monitor) + {client, clients} = Map.pop(state.clients, pid) + {:noreply, accountClosed(%{state | monitors: monitors, clients: clients}, client)} + end + + defp schedule do + interval = Application.get_env(:excache, :clientsPollInterval, @defaultPoll) + Process.send_after(self(), :poll, interval) + end + + defp sync(state) do + Enum.reduce(livePids(), state, fn pid, acc -> + case Map.fetch(acc.clients, pid) do + {:ok, client} -> %{acc | clients: Map.put(acc.clients, pid, refresh(client))} + :error -> register(acc, pid) + end + end) + end + + defp livePids do + for {_id, pid, _type, _modules} <- DynamicSupervisor.which_children(@supervisor), + is_pid(pid), + do: pid + end + + defp register(state, pid) do + with socket when socket != nil <- socketOf(pid), + {netIn, netOut} <- netStats(socket) do + now = System.system_time(:millisecond) + + client = %{ + id: state.nextId, + pid: pid, + socket: socket, + addr: addressOf(socket, &:inet.peername/1), + laddr: addressOf(socket, &:inet.sockname/1), + name: "", + connectedAt: now, + lastActiveAt: now, + netIn: netIn, + netOut: netOut + } + + %{ + state + | nextId: state.nextId + 1, + clients: Map.put(state.clients, pid, client), + monitors: Map.put(state.monitors, Process.monitor(pid), pid), + connectionsReceived: state.connectionsReceived + 1 + } + else + _unavailable -> state + end + end +