defmodule Pollin.Backend.Memory.CallbackWorker do @moduledoc """ Worker for callbacks """ use GenServer use Pollin.CallbackInterface alias Pollin.Backend.Memory.EndpointWorker @doc false def start_link(id), do: GenServer.start_link(__MODULE__, id, name: id) @doc false def update_endpoint(endpoint), do: GenServer.cast(:"#{endpoint.id}", {:update_endpoint, endpoint}) ## Private API @doc false def init(id) do case EndpointWorker.find("#{id}") do nil -> {:stop, :normal} endpoint -> {:ok, {endpoint, []}} end end @doc false def handle_cast({:update_endpoint, endpoint}, {_endpoint, state}), do: {:noreply, {endpoint, state}} @doc false def handle_cast({:push, cb, secret}, {endpoint, state}) do if endpoint && endpoint.secret == secret do ttl_delete(self, cb.id, endpoint.ttl) cb = put_phase(cb, cb.phase, endpoint.phases) {:noreply, {endpoint, List.insert_at(state, -1, {cb.id, cb})}} else {:noreply, {endpoint, state}} end end @doc false def handle_cast({:reset}, {endpoint, _state}), do: {:noreply, {endpoint, []}} @doc false def handle_cast({:delete, key}, {endpoint, state}) do new_state = List.keydelete(state, key, 0) {:noreply, {endpoint, new_state}} end @doc false def handle_cast({:update_phase, key, phase}, {endpoint, state}) do case List.keyfind(state, key, 0) do {key, cb} -> case Enum.member?(endpoint.phases, phase) do true -> cb = Map.put(cb, :phase, phase) {:noreply, {endpoint, List.keyreplace(state, key, 0, {key, cb})}} false -> {:noreply, {endpoint, state}} end nil -> {:noreply, {endpoint, state}} end end @doc false def handle_call({:pop}, _from, {endpoint, [{_key, cb} | state]}), do: {:reply, cb, {endpoint, state}} @doc false def handle_call({:pop, key}, _from, {endpoint, state}) when is_binary(key) do case List.keytake(state, key, 0) do {{_key, cb}, new_state} -> {:reply, cb, {endpoint, new_state}} nil -> {:reply, nil, {endpoint, state}} end end @doc false def handle_call({:pop, %{phase: phase, offset: offset, limit: limit}}, _from, {endpoint, state}) do items = state |> Enum.group_by(fn {_key, cb} -> cb.phase end) |> Map.get(phase, []) |> Enum.slice(offset, limit) items_without_key = items |> Enum.map(fn {_key, cb} -> cb end) {:reply, items_without_key, {endpoint, state -- items}} end @doc false def handle_call({:pop, %{offset: offset, limit: limit}}, _from, {endpoint, state}) do items = state |> Enum.slice(offset, limit) items_without_key = items |> Enum.map(fn {_key, cb} -> cb end) {:reply, items_without_key, {endpoint, state -- items}} end @doc false def handle_call({:fetch}, _from, {_endpoint, [{_key, cb} | _state]} = {endpoint, state}), do: {:reply, cb, {endpoint, state}} @doc false def handle_call({:fetch, key}, _from, {endpoint, state}) when is_binary(key) do case List.keyfind(state, key, 0) do {_key, cb} -> {:reply, cb, {endpoint, state}} nil -> {:reply, nil, {endpoint, state}} end end @doc false def handle_call({:fetch, %{phase: phase, offset: offset, limit: limit}}, _from, {endpoint, state}) do items = state |> Enum.group_by(fn {_key, cb} -> cb.phase end) |> Map.get(phase, []) |> Enum.slice(offset, limit) |> Enum.map(fn {_key, cb} -> cb end) {:reply, items, {endpoint, state}} end @doc false def handle_call({:fetch, %{offset: offset, limit: limit}}, _from, {endpoint, state}) do items = state |> Enum.slice(offset, limit) |> Enum.map(fn {_key, cb} -> cb end) {:reply, items, {endpoint, state}} end @doc false def handle_call({:fetch_reverse, %{phase: phase, offset: offset, limit: limit}}, _from, {endpoint, state}) do items = state |> Enum.reverse |> Enum.group_by(fn {_key, cb} -> cb.phase end) |> Map.get(phase, []) |> Enum.slice(offset, limit) |> Enum.map(fn {_key, cb} -> cb end) {:reply, items, {endpoint, state}} end @doc false def handle_call({:fetch_reverse, %{offset: offset, limit: limit}}, _from, {endpoint, state}) do items = state |> Enum.reverse |> Enum.slice(offset, limit) |> Enum.map(fn {_key, cb} -> cb end) {:reply, items, {endpoint, state}} end @doc false def handle_call({:count}, _from, {endpoint, state}), do: {:reply, Enum.count(state), {endpoint, state}} @doc false def handle_call({:count, %{phase: phase}}, _from, {endpoint, state}) do count = state |> Enum.group_by(fn {_key, cb} -> cb.phase end) |> Map.get(phase, []) |> Enum.count {:reply, count, {endpoint, state}} end @doc false def handle_call({:dump}, _from, {endpoint, state}) do items = state |> Enum.map(fn {_key, cb} -> cb end) {:reply, items, {endpoint, state}} end @doc false def handle_info({:delete, key}, {endpoint, state}) do new_state = List.keydelete(state, key, 0) {:noreply, {endpoint, new_state}} end defp put_phase(cb, "", phases), do: Map.put(cb, :phase, phases |> List.first) defp put_phase(cb, _phase, _phases), do: cb defp ttl_delete(process, id, ttl) when ttl > 0, do: Process.send_after(process, {:delete, id}, ttl) defp ttl_delete(_process, _id, _ttl), do: :ok end