Multiple Subscribers

Overview

Fan-out delivery lets several independent consumers each get their own copy of every event published on a channel — the pattern behind broadcasting a notification to every connected service or feeding the same stream to a cache invalidator and a metrics collector at once. Reach for it whenever multiple, unrelated pieces of code all need to react to the same event, rather than compete for it.

It works by calling subscribe_to_events more than once for the same channel while leaving the :group option out. Each call opens its own subscription, and the broker treats every subscriber with no group as broadcast: publishing one event delivers it to every open subscription — the opposite of a consumer group, where subscribers sharing a :group value split events among themselves for load balancing.

Gotchas: Events pub/sub has no durability — a subscriber that hasn't finished subscribing yet, or that disconnects, simply misses events published in that window; there's no redelivery. Passing a :group option to one subscriber on the same channel silently turns broadcast into load-balancing for it. Each on_event callback runs in the client process and just messages the caller, so slow handling in one doesn't block the others.

Prerequisites

• KubeMQ server running on localhost:50000
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-events.multiple-subscribers"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-events-multi")
parent = self()

subs =
  for i <- 1..3 do
    {:ok, sub} =
      KubeMQ.Client.subscribe_to_events(client, channel,
        on_event: fn event ->
          IO.puts("[Subscriber #{i}] #{event.body}")
          send(parent, {:sub, i})
        end
      )

    sub
  end

IO.puts("3 subscribers ready on '#{channel}'")
Process.sleep(500)

event = KubeMQ.Event.new(channel: channel, body: "Broadcast message")
:ok = KubeMQ.Client.send_event(client, event)
IO.puts("Event sent. Waiting for all subscribers...")

for _ <- 1..3 do
  receive do
    {:sub, id} -> IO.puts("  Confirmed from subscriber #{id}")
  after
    5_000 -> IO.puts("  Timeout")
  end
end

Enum.each(subs, &KubeMQ.Subscription.cancel/1)
KubeMQ.Client.close(client)
IO.puts("Done.")

How It Works

• Without a group option, each subscriber independently receives every event
• This is the fan-out delivery pattern — one event, multiple consumers
• List comprehension creates 3 subscribers and collects their subscription handles
• Enum.each/2 cancels all subscriptions at cleanup

Related

• Consumer Group
• Fan-Out Pattern