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 SubscribeToEvents with an empty group string puts that subscription in broadcast mode instead of load-balanced mode. SendEvent 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 string silently turns broadcast into competing-consumer load-balancing with no error raised. Events are not persisted — a subscriber that isn't connected yet when SendEvent runs misses that event permanently (use Events Store if you need replay). And SendEvent 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 startup delay before publishing in this sample.

Prerequisites

• KubeMQ server running on localhost:50000
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: patterns/fan-out
//
// Demonstrates the fan-out pattern using events.
// A single publisher sends events that are delivered to all subscribers
// on the channel. Each subscriber receives every event independently.
//
// Channel: go-patterns.fan-out
// Client ID: go-patterns-fan-out-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"sync/atomic"
	"time"

	"github.com/kubemq-io/kubemq-go/v2"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000),
		kubemq.WithClientId("go-patterns-fan-out-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-patterns.fan-out"
	var deliveries atomic.Int32

	// Create three independent subscribers (no consumer group = fan-out).
	for i := 1; i <= 3; i++ {
		subscriberID := i
		sub, err := client.SubscribeToEvents(ctx, channel, "",
			kubemq.WithOnEvent(func(event *kubemq.Event) {
				fmt.Printf("Subscriber %d received: body=%s\n", subscriberID, event.Body)
				deliveries.Add(1)
			}),
			kubemq.WithOnError(func(err error) {
				log.Printf("Subscriber %d error: %v", subscriberID, err)
			}),
		)
		if err != nil {
			log.Fatal(err)
		}
		defer sub.Unsubscribe()
	}

	// Publish a single event — all 3 subscribers should receive it.
	err = client.SendEvent(ctx, kubemq.NewEvent().
		SetChannel(channel).
		SetBody([]byte("broadcast message")).
		SetMetadata("fan-out"))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("Event published to all subscribers")

	// Wait for deliveries.
	time.Sleep(2 * time.Second)
	fmt.Printf("Total deliveries: %d (expected 3 for fan-out)\n", deliveries.Load())
}

How It Works

1. The for i := 1; i <= 3; i++ loop creates three separate subscriptions on go-patterns.fan-out, each with an empty group string — no consumer group means broadcast delivery.
2. subscriberID is captured by value in the closure before the loop variable advances, ensuring each callback prints the correct subscriber number.
3. atomic.Int32 provides a race-safe delivery counter that all three callback goroutines increment concurrently.
4. A single SendEvent reaches all three active subscribers; deliveries.Load() after a two-second wait confirms the fan-out multiplier.

Related

• Pattern overview
• Go SDK Reference
• Request-Reply
• Work Queue