defmodule Oban.Notifiers.Phoenix do @external_resource readme = Path.join([__DIR__, "../../../README.md"]) @moduledoc readme |> File.read!() |> String.split("") |> Enum.fetch!(1) @behaviour Oban.Notifier use GenServer alias Oban.Notifier alias Phoenix.PubSub defstruct [:conf, :pubsub] @doc false def child_spec(opts), do: super(opts) @impl Notifier def start_link(opts) do {name, opts} = Keyword.pop(opts, :name, __MODULE__) GenServer.start_link(__MODULE__, struct!(__MODULE__, opts), name: name) end @impl Notifier def listen(server, channels) do with {:ok, %{pubsub: pubsub}} <- get_state(server) do for channel <- channels, do: PubSub.subscribe(pubsub, to_string(channel), metadata: __MODULE__) :ok end end @impl Notifier def unlisten(server, channels) do with {:ok, %{pubsub: pubsub}} <- get_state(server) do for channel <- channels, do: PubSub.unsubscribe(pubsub, to_string(channel)) :ok end end @impl Notifier def notify(server, channel, payload) do with {:ok, %{conf: conf, pubsub: pubsub}} <- get_state(server) do PubSub.broadcast( pubsub, to_string(channel), {conf.name, channel, payload}, __MODULE__ ) :ok end end @impl GenServer def init(state) do put_state(state) {:ok, state} end @doc false def dispatch(entries, :none, {name, channel, payload}) do pids = Enum.map(entries, &elem(&1, 0)) conf = Oban.config(name) for message <- payload, do: Notifier.relay(conf, pids, channel, message) end # Backward compatibility with 0.2.1 def dispatch(entries, :none, {name, _node, channel, payload}) do dispatch(entries, :none, {name, channel, payload}) end # Backward compatibility with < 0.2.0 def dispatch(entries, _from, {conf, channel, payload}) do dispatch(entries, :none, {conf.name, conf.node, channel, payload}) end defp put_state(state) do Registry.update_value(Oban.Registry, {state.conf.name, Oban.Notifier}, fn _ -> state end) end defp get_state(server) do [name] = Registry.keys(Oban.Registry, server) case Oban.Registry.lookup(name) do {_pid, state} -> {:ok, state} nil -> {:error, RuntimeError.exception("no notifier running as #{inspect(name)}")} end end end