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: an EventsSubscription built without a group puts that subscription in broadcast mode instead of load-balanced mode. 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 sleep before publishing in this sample.

Prerequisites

• KubeMQ server running on localhost:50000
• Ruby SDK installed (gem install kubemq)

Code

require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-patterns.fan-out'

begin
  client = KubeMQ::PubSubClient.new(address: address, client_id: 'fan-out-example')
  puts "Connected to #{address}"

  cancel = KubeMQ::CancellationToken.new

  %w[Service-A Service-B Service-C].each do |name|
    sub = KubeMQ::PubSub::EventsSubscription.new(channel: channel)
    client.subscribe_to_events(sub, cancellation_token: cancel, on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
      puts "#{name} received: #{event.metadata}"
    end
  end
  sleep 1

  msg = KubeMQ::PubSub::EventMessage.new(channel: channel, metadata: 'system-update', body: 'update-data')
  client.send_event(msg)

  sleep 2
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  cancel&.cancel
  client&.close
  puts 'Done'
end

How It Works

• Three subscribers without a group — all receive each event (fan-out delivery).
• Ruby's block-based subscriptions make it natural to iterate over service names and create subscribers dynamically.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Ruby SDK Reference
• Multiple Subscribers
• Consumer Group