The events channel · dive 01

Subscribe and handle

One subscription, on one connection — and the bus pushes every event two ways.

A listener subscribes once to emq:{q}:events on a protocol: 3 connector, and from then on the bus pushes every event to it. Each push is delivered two ways: to every registered subscriber pid as {:emq_event, name, payload}, and to an optional handler module's handle_event/3. Step the subscription's lifecycle on the right, then read the code.

The one-time subscribe The two deliveries Fan-out inspector Pattern & implementation References

A subscriber's lifecycle · step the moves

The one-time subscribe

EchoMQ.Events is a GenServer that owns one subscription. On init/1 it takes a connector — either a caller-supplied :conn (which must be protocol: 3 and push_to the listener) or a :connector options list it starts linked, with protocol: 3 and push_to: self() set for you — and issues exactly one subscribe to emq:{q}:events. That single subscribe is the whole wire cost; every later subscriber rides it.

Why protocol: 3? Pub/sub pushes are out-of-band server messages. RESP3 carries them on a separate push channel the connector routes to the push_to process as {:emq_push, frame}, so a published message never collides with a command reply on the same socket.

# echo_mq — EchoMQ.Events.init/1 (the one-time subscribe)
# The listener owns ONE subscription to emq:{q}:events. The connector MUST be
# RESP3 (protocol: 3) and MUST push_to this process — so pub/sub pushes arrive
# as {:emq_push, …} out-of-band, never colliding with a command reply.
def init(opts) do
  Process.flag(:trap_exit, true)
  queue = Keyword.fetch!(opts, :queue)

  conn =
    case Keyword.fetch(opts, :conn) do
      {:ok, c} ->
        c                                    # a caller-supplied RESP3 + push_to connector
      :error ->
        conn_opts =
          opts
          |> Keyword.fetch!(:connector)
          |> Keyword.put_new(:protocol, 3)   # RESP3 — push frames on their own channel
          |> Keyword.put(:push_to, self())   # route every push to this listener
        {:ok, c} = Connector.start_link(conn_opts)
        c
    end

  chan = channel(queue)                       # emq:{q}:events

  case Connector.subscribe(conn, chan) do     # the SINGLE SUBSCRIBE — every subscriber rides it
    :ok ->
      {:ok, %{conn: conn, queue: queue, channel: chan,
              handler: Keyword.get(opts, :handler),
              handler_state: Keyword.get(opts, :handler_state),
              subscribers: MapSet.new()}}
    {:error, reason} ->
      {:stop, {:subscribe_failed, reason}}
  end
end

The feed stays live across a reconnect because of work done one layer down. The connector keeps the set of channels it is subscribed to in a MapSet; when the socket drops and reconnects, it re-issues SUBSCRIBE for every channel in that set. The listener never re-subscribes — it does not even learn the socket bounced. That resubscribe MapSet is what makes "subscribe once" safe in a long-lived process.

One subscription is the whole cost. Open it on a RESP3 push connection, and the bus does the rest — including bringing the feed back, by itself, after a reconnect.

The two deliveries

A published message arrives at the listener as {:emq_push, ["message", channel, payload]}. The listener matches its own channel, reads the event name, and dispatches both ways in one pass: a send/2 to every registered subscriber pid, and a call into the handler module if one is configured.

# echo_mq — EchoMQ.Events.handle_info/2 (the dispatch)
# A published message arrives as {:emq_push, ["message", channel, payload]}.
# (The connector's "subscribe"/"unsubscribe" confirmation frames are ignored.)
# Each message dispatches TWO ways off the one subscription:
def handle_info({:emq_push, ["message", chan, payload]}, %{channel: chan} = s)
    when is_binary(payload) do
  event = event_name(payload)                              # the name, by scan
  # 1) every subscriber pid gets {:emq_event, name, raw_payload}
  Enum.each(s.subscribers, fn pid -> send(pid, {:emq_event, event, payload}) end)
  # 2) the optional handler module folds the event into its state
  {:noreply, dispatch_handler(s, event, payload)}
end

def handle_info({:emq_push, _frame}, s), do: {:noreply, s}  # ignore non-message frames

Subscriber pids. subscribe(server, pid \ self()) registers a pid (idempotent, defaults to the caller); unsubscribe/2 removes it. A registered pid receives {:emq_event, event_name, raw_payload} for every event, and pattern-matches the ones it cares about in its own handle_info. This is the loose-coupling shape: a LiveView, a metrics collector, a test process — each subscribes independently.

# echo_mq — EchoMQ.Events.subscribe/2 + unsubscribe/2 (the pid registry)
# Register a pid to receive every event as {:emq_event, name, payload}.
# Idempotent (a MapSet); defaults to the caller. unsubscribe/2 is the inverse.
def subscribe(server, pid \ self()), do: GenServer.call(server, {:subscribe, pid})
def unsubscribe(server, pid \ self()), do: GenServer.call(server, {:unsubscribe, pid})

# the owner side: add/remove the pid in the subscribers MapSet
def handle_call({:subscribe, pid}, _from, s),
  do: {:reply, :ok, %{s | subscribers: MapSet.put(s.subscribers, pid)}}
def handle_call({:unsubscribe, pid}, _from, s),
  do: {:reply, :ok, %{s | subscribers: MapSet.delete(s.subscribers, pid)}}

The handler module. An optional :handler implementing handle_event/3 is invoked per event with the handler's state and answers {:ok, new_state}. use EchoMQ.Events derives a no-op default that keeps the state, so a real handler overrides only the events of interest. This is the stateful-reducer shape: fold the event stream into a running state without writing a receive loop.

# echo_mq — the handle_event/3 behaviour + the no-op default from `use EchoMQ.Events`
@callback handle_event(event :: atom(), payload :: binary(), state :: term()) ::
            {:ok, state :: term()}

# `use EchoMQ.Events` derives this no-op — override only the events you care about:
defmacro __using__(_opts) do
  quote do
    @behaviour EchoMQ.Events
    @impl EchoMQ.Events
    def handle_event(_event, _payload, state), do: {:ok, state}   # keep state by default
    defoverridable handle_event: 3
  end
end

Two delivery shapes, one subscription: a send/2 to each pid for loose coupling, and a handle_event/3 reducer for state. Neither costs a second connection.

The fan-out, inspected

Set the registered subscribers and whether a handler is configured, then push a fixed event through the dispatch and read exactly what is sent: one send/2 per subscriber pid and, if present, one handle_event/3 call. This is the real handle_info/2 path over a fixed message — pure, no network.

Dispatch a pushed message · count the deliveries

registered subscriber pids 3

The dispatch is one Enum.each over the subscribers plus one optional handler call. Per-subscriber cost is a single message send — the subscription is shared, so adding a subscriber adds a send, not a connection.

Pattern & implementation

the pattern (Redis Patterns Applied)

Publish/subscribe lets a producer broadcast to many consumers without knowing them; a consumer reacts to events instead of polling. R5 · Streams & Events teaches the pub/sub side.

the implementation (echo_mq)

EchoMQ.Events holds one RESP3 subscription to emq:{q}:events and dispatches each push two ways — to subscriber pids and to a handle_event/3 handler — with the connector's resubscribe MapSet keeping the feed live across a reconnect.

The pattern says: broadcast, and let consumers react. The implementation holds one subscription and fans each message to pids and a reducer — no second connection, no polling.

Recap — subscribe once, deliver twice

EchoMQ.Events subscribes once to emq:{q}:events on a protocol: 3 + push_to connector. Each pushed message is delivered to every registered subscriber pid as {:emq_event, name, payload} and to an optional handle_event/3 handler. The connector's resubscribe MapSet re-issues the SUBSCRIBE across a reconnect, so the feed stays live. The next dive reads where the events come from: the host-side publish after a verdict.

Next in the module: Publish after the verdict — why the publish lives in the host, the hand-built cjson payload, and the substring-scan name read.

References

Where the one-time subscription and the two-way delivery are grounded, and the command beneath them.

Sources

Valkey — SUBSCRIBE — the one-time subscription the listener holds on the events channel.
Valkey — Introduction to Streams — the bus surfaces this pillar builds on; here, the pub/sub half.
Valkey — Cluster specification — the {q} hashtag co-locates the events channel with the queue on one slot.

Related in this course

The events channel — the module this dive belongs to.
Publish after the verdict — where the events the listener receives come from.
Fire-and-forget — what happens when no one is subscribed.
The Protocol — the keyspace emq:{q}:events is built from.
redis-patterns · Streams & Events — the pub/sub pattern.