Fan-Out

Overview

Fan-out is the default delivery behavior of KubeMQ Events pub/sub: when subscribers don't join a consumer group, every subscriber gets its own independent copy of each published event. Reach for it whenever several unrelated services need to react to the same occurrence — an order placed, a config change, an audit event — without the publisher knowing or caring who's listening, and without one subscriber's slowness affecting another's delivery.

The mechanism is simply omission: calling KubeMQ.Client.subscribe_to_events without a group option puts that subscription in broadcast mode instead of load-balanced mode. KubeMQ.Client.send_event doesn't change at all — the publisher sends once, and the broker independently pushes a copy to every active subscriber on the channel.

Gotchas: fan-out is opt-out by default, so a typo'd or accidentally shared group value silently turns broadcast into competing-consumer load-balancing with no error raised. Events are not persisted — a subscriber that hasn't called subscribe_to_events yet when send_event runs misses that event permanently (use Events Store if you need replay). And send_event returns as soon as the broker accepts it, not after subscribers process it, so a publisher can outrun subscription setup on a cold start — hence the short Process.sleep before publishing in this sample.

Prerequisites

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

Code

channel = "elixir-patterns.fan-out"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-fan-out")
parent = self()

services = ["email-service", "analytics-service", "audit-service"]

subs =
  Enum.map(services, fn service ->
    {:ok, sub} =
      KubeMQ.Client.subscribe_to_events(client, channel,
        on_event: fn event ->
          IO.puts("[#{service}] Processing: #{event.body}")
          send(parent, {:processed, service})
        end
      )

    {service, sub}
  end)

IO.puts("#{length(services)} services subscribed (fan-out)")
Process.sleep(500)

event = KubeMQ.Event.new(
  channel: channel,
  body: "New user registered: user@example.com",
  tags: %{"event_type" => "user.registered"}
)

:ok = KubeMQ.Client.send_event(client, event)
IO.puts("\nPublished event. Waiting for all services to process...")

received = for _ <- 1..3 do
  receive do
    {:processed, service} -> service
  after
    5_000 -> "timeout"
  end
end

IO.puts("Processed by: #{Enum.join(received, ", ")}")

Enum.each(subs, fn {_, sub} -> KubeMQ.Subscription.cancel(sub) end)
KubeMQ.Client.close(client)

How It Works

• Three independent services subscribe without a consumer group
• Each service receives a copy of every published event
• This pattern is ideal for triggering multiple side effects from a single event (email, analytics, audit)

Related

• Multiple Subscribers