defmodule Oban.Notifier do
@moduledoc """
The `Notifier` coordinates listening for and publishing notifications for events in predefined
channels.
Every Oban supervision tree contains a notifier process, registered as `Oban.Notifier`, which
must be an implementation of the `Oban.Notifier` behaviour. The default implementation is uses
the `LISTEN/NOTIFY` operations built into Postgres.
All incoming notifications are relayed through the notifier to other processes.
## Channels
Internally, Oban uses a variety of predefined channels with distinct responsibilities:
* `insert` — as jobs are inserted into the database an event is published on the `insert`
channel. Processes such as queue producers use this as a signal to dispatch new jobs.
* `leader` — messages regarding node leadership exchanged between peers
* `signal` — instructions to take action, such as scale a queue or kill a running job, are sent
through the `signal` channel
* `gossip` — arbitrary communication for coordination between nodes
* `stager` — messages regarding job staging, e.g. notifying queues that jobs are ready for execution
## Examples
Broadcasting after a job is completed:
defmodule MyApp.Worker do
use Oban.Worker
@impl Oban.Worker
def perform(job) do
:ok = MyApp.do_work(job.args)
Oban.Notifier.notify(Oban, :my_app_jobs, %{complete: job.id})
:ok
end
end
Listening for job complete events from another process:
def insert_and_listen(args) do
:ok = Oban.Notifier.listen([:my_app_jobs])
{:ok, %{id: job_id} = job} =
args
|> MyApp.Worker.new()
|> Oban.insert()
receive do
{:notification, :my_app_jobs, %{"complete" => ^job_id}} ->
IO.puts("Other job complete!")
after
30_000 ->
IO.puts("Other job didn't finish in 30 seconds!")
end
end
"""
alias Oban.{Config, Registry}
@type server :: GenServer.server()
@type option :: {:name, module()} | {:conf, Config.t()}
@type channel :: atom()
@doc """
Starts a notifier instance.
"""
@callback start_link([option]) :: GenServer.on_start()
@doc """
Register the current process to receive messages from one or more channels.
"""
@callback listen(server(), channels :: list(channel())) :: :ok
@doc """
Unregister current process from channels.
"""
@callback unlisten(server(), channels :: list(channel())) :: :ok
@doc """
Broadcast a notification to all subscribers of a channel.
"""
@callback notify(server(), channel :: channel(), payload :: [map()]) :: :ok
@doc false
def child_spec(opts) do
conf = Keyword.fetch!(opts, :conf)
opts = Keyword.put_new(opts, :name, conf.notifier)
%{id: opts[:name], start: {conf.notifier, :start_link, [opts]}}
end
@doc """
Register the current process to receive relayed messages for the provided channels.
All messages are received as `JSON` and decoded _before_ they are relayed to registered
processes. Each registered process receives a three element notification tuple in the following
format:
{:notification, channel :: channel(), decoded :: map()}
## Example
Register to listen for all `:gossip` channel messages:
Oban.Notifier.listen(:gossip)
Listen for messages on multiple channels:
Oban.Notifier.listen([:gossip, :insert, :leader, :signal, :stager])
Listen for messages when using a custom Oban name:
Oban.Notifier.listen(MyApp.MyOban, [:gossip, :signal])
"""
@spec listen(server(), channel() | [channel()]) :: :ok
def listen(name \\ Oban, channels)
def listen(name, channel) when is_atom(channel) do
listen(name, [channel])
end
def listen(name, channels) when is_list(channels) do
unless Enum.all?(channels, &is_atom/1) do
raise ArgumentError, "expected channels to be a list of atoms, got: #{inspect(channels)}"
end
conf = Oban.config(name)
name
|> Registry.whereis(Oban.Notifier)
|> conf.notifier.listen(channels)
end
@doc """
Unregister the current process from receiving relayed messages on provided channels.
## Example
Stop listening for messages on the `:gossip` channel:
Oban.Notifier.unlisten([:gossip])
Stop listening for messages when using a custom Oban name:
Oban.Notifier.unlisten(MyApp.MyOban, [:gossip])
"""
@spec unlisten(server(), [channel]) :: :ok
def unlisten(name \\ Oban, channels) when is_list(channels) do
conf = Oban.config(name)
name
|> Registry.whereis(Oban.Notifier)
|> conf.notifier.unlisten(channels)
end
@doc """
Broadcast a notification to listeners on all nodes.
Notifications are scoped to the configured `prefix`. For example, if there are instances running
with the `public` and `private` prefixes, a notification published in the `public` prefix won't
be picked up by processes listening with the `private` prefix.
Using notify/3 with a config is soft deprecated. Use a server as the first argument instead
## Example
Broadcast a gossip message:
Oban.Notifier.notify(:gossip, %{message: "hi!"})
"""
@spec notify(Config.t() | server(), channel :: channel(), payload :: map() | [map()]) :: :ok
def notify(conf_or_name \\ Oban, channel, payload)
def notify(%Config{} = conf, channel, payload) when is_atom(channel) do
with_span(conf, channel, payload, fn ->
conf.name
|> Registry.whereis(Oban.Notifier)
|> conf.notifier.notify(channel, normalize_payload(payload))
end)
end
def notify(name, channel, payload) when is_atom(channel) do
name
|> Oban.config()
|> notify(channel, payload)
end
defp with_span(conf, channel, payload, fun) do
tele_meta = %{conf: conf, channel: channel, payload: payload}
:telemetry.span([:oban, :notifier, :notify], tele_meta, fn ->
{fun.(), tele_meta}
end)
end
defp normalize_payload(payload) do
payload
|> List.wrap()
|> Enum.map(&encode/1)
end
@doc false
@spec relay(Config.t(), [pid()], atom(), binary()) :: :ok
def relay(_conf, [], _channel, _payload), do: :ok
def relay(conf, listeners, channel, payload) when is_atom(channel) and is_binary(payload) do
decoded = decode(payload)
if in_scope?(decoded, conf) do
for pid <- listeners, do: send(pid, {:notification, channel, decoded})
end
:ok
end
defp encode(payload) do
payload
|> to_encodable()
|> Jason.encode!()
|> :zlib.gzip()
|> Base.encode64()
end
defp decode(payload) do
case Base.decode64(payload) do
{:ok, decoded} ->
decoded
|> :zlib.gunzip()
|> Jason.decode!()
# Messages emitted by the insert trigger aren't compressed.
:error ->
Jason.decode!(payload)
end
end
defp to_encodable(%_{} = struct), do: struct
defp to_encodable(map) when is_map(map) do
for {key, val} <- map, into: %{}, do: {key, to_encodable(val)}
end
defp to_encodable(list) when is_list(list) do
for element <- list, do: to_encodable(element)
end
defp to_encodable(tuple) when is_tuple(tuple) do
tuple
|> Tuple.to_list()
|> to_encodable()
end
defp to_encodable(term), do: term
defp in_scope?(%{"ident" => "any"}, _conf), do: true
defp in_scope?(%{"ident" => ident}, conf), do: Config.match_ident?(conf, ident)
defp in_scope?(_payload, _conf), do: true
end
