Pub/Sub with Registry and send/2

Before the code: messaging and replication, in plain English

Picture a group chat. One person publishes a message. The chat app knows who is in the room and forwards the message to each of them. That is publish and subscribe, and it is the simplest piece of Redis that is not a key-value store. Replication is the same idea pointed at a backup machine: the leader sends every write to a follower, who quietly replays them. Same plumbing, different audience. On the BEAM this is genuinely the cleanest module in the workshop: Registry gives you a phone book of subscribers and send/2 delivers the message.

Fan-out and follow-the-leader

A publisher sends once. Registry returns every subscriber pid. send/2 delivers the message.

Quiz: Quiz

Loading practice…

Registry is OTP's built-in pid-keyed lookup table. keys: :duplicate supports many pids per name. Registry.register binds THIS pid to a name; Registry.lookup returns every pid for a name.

07-pubsub/server.exs
elixir
defmodule PubSub do
  def start_link, do: Registry.start_link(keys: :duplicate, name: __MODULE__)
  def subscribe(channel), do: Registry.register(__MODULE__, channel, nil)
  def publish(channel, payload) do
    pids = Registry.lookup(__MODULE__, channel) |> Enum.map(fn {pid, _} -> pid end)
    Enum.each(pids, fn pid -> send(pid, {:pubsub_msg, channel, payload}) end)
    length(pids)
  end
end

The entire PubSub module. Three functions, all delegating to Registry and send/2.

This is the cleanest pub/sub module of any sibling course. Other languages need a connection list, locks, write pumps, channels, mpsc bridges. Elixir uses Registry + send/2. Registry auto-cleans up when a subscriber pid dies.

07-pubsub/server.exs
elixir
defp serve(socket, buf, subs) do
  receive do
    {:pubsub_msg, channel, payload} ->
      :gen_tcp.send(socket, encode_message(channel, payload))
      serve(socket, buf, subs)
  after 0 ->
    case :gen_tcp.recv(socket, 0, 50) do
      {:ok, data} -> ...
      {:error, :timeout} -> serve(socket, buf, subs)
      {:error, _} -> :gen_tcp.close(socket)
    end
  end
end

The connection process drains its mailbox between recv attempts.

Quiz: Quiz

Loading practice…