Overview

Which to use

This page is the pattern-level walkthrough of fan-out (three subscribers, a runnable end-to-end example). For the SDK-scoped how-to that also covers the alternative consumer-group delivery model, see Multiple Subscribers.

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 registered 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 sleep before publishing in this sample.

Prerequisites

KubeMQ server running on localhost:50000
C++ SDK installed (vcpkg or CMake FetchContent)
C++17 compiler (GCC 9+, Clang 9+, MSVC 2019+)

Code

main.cc

// 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
// because no consumer group is used.
//
// Channel: cpp-patterns.fan-out
// Client ID: cpp-patterns-fan-out-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).

#include <kubemq/kubemq.h>

#include <atomic>
#include <chrono>
#include <iostream>
#include <thread>
#include <vector>

int main() {
    std::cout << "[1] Connecting to localhost:50000" << std::endl;

    kubemq::ClientOptions options;
    options.set_address("localhost", 50000);
    options.set_client_id("cpp-patterns-fan-out-client");

    auto client_result = kubemq::Client::Create(options);
    if (!client_result.ok()) {
        std::cerr << "[ERROR] Failed to create client: " << client_result.status().message()
                  << std::endl;
        return 1;
    }
    auto& client = *client_result;

    std::string channel = "cpp-patterns.fan-out";
    std::atomic<int> deliveries{0};

    // Create 3 independent subscribers (no consumer group = fan-out).
    std::cout << "[2] Creating 3 fan-out subscribers" << std::endl;
    std::vector<std::unique_ptr<kubemq::Subscription>> subs;
    for (int i = 1; i <= 3; i++) {
        int subscriber_id = i;
        auto sub_result = client->SubscribeToEvents(
            channel, "",
            [subscriber_id, &deliveries](const kubemq::EventReceive& event) {
                std::cout << "[4] Subscriber " << subscriber_id << " received: body=" << event.body
                          << std::endl;
                deliveries.fetch_add(1);
            },
            [subscriber_id](const kubemq::Status& err) {
                std::cerr << "[ERROR] Subscriber " << subscriber_id << " error: " << err.message()
                          << std::endl;
            });
        if (!sub_result.ok()) {
            std::cerr << "[ERROR] SubscribeToEvents (subscriber " << i
                      << "): " << sub_result.status().message() << std::endl;
            return 1;
        }
        subs.push_back(std::move(*sub_result));
    }

    // Allow subscriptions to fully establish.
    std::this_thread::sleep_for(std::chrono::seconds(1));

    // Publish a single event -- all 3 subscribers should receive it.
    std::cout << "[3] Publishing event to all subscribers" << std::endl;
    auto event_result = kubemq::Event::Builder()
                            .SetChannel(channel)
                            .SetBody("broadcast message")
                            .SetMetadata("fan-out")
                            .Build();
    if (!event_result.ok()) {
        std::cerr << "[ERROR] Build event: " << event_result.status().message() << std::endl;
        return 1;
    }
    auto send_status = client->SendEvent(*event_result);
    if (!send_status.ok()) {
        std::cerr << "[ERROR] SendEvent: " << send_status.message() << std::endl;
        return 1;
    }

    // Wait for all deliveries.
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "[5] Total deliveries: " << deliveries.load() << " (expected 3 for fan-out)"
              << std::endl;

    // Cancel all subscriptions explicitly.
    // Note: The Subscription destructor also calls Cancel(), but explicit
    // cleanup is shown here for clarity and to match Go's defer pattern.
    for (auto& sub : subs) {
        sub->Cancel();
    }
    std::cout << "[6] All subscriptions cancelled" << std::endl;

    auto close_status = client->Close();
    if (!close_status.ok()) {
        std::cerr << "[ERROR] Close failed: " << close_status.message() << std::endl;
        return 1;
    }
    std::cout << "[7] Client closed" << std::endl;

    return 0;
}

How It Works

Creates 3 independent subscribers on the same channel without a consumer group.
Publishes a single event that is delivered to all 3 subscribers.
Uses std::atomic<int> to count total deliveries across callback threads.
Total deliveries should equal the number of subscribers (3 for fan-out).

Fan-Out

Overview

Prerequisites

Code

How It Works

On this page

Fan-Out

Overview

Prerequisites

Code

How It Works

Related

On this page