Overview

This tutorial builds a publisher and subscriber on a KubeMQ Events Store channel — reach for this pattern when a subscriber can't guarantee it's listening the instant a message is published. Plain events are fire-and-forget: publish with no one subscribed and the message is gone. Events Store persists every event to a durable, ordered log, so a subscriber connecting seconds or a full restart later still catches up — useful for anything needing a complete history, like an audit trail or event-sourced state.

The two calls involved: SendEventStore publishes and returns a result confirming storage with an id and sent flag plus a broker-assigned sequence number, and SubscribeToEventsStore takes a required SubscriptionOption telling the broker where to start — new events only (StartFromNewEvents(), used here), from the first stored event, or a given sequence or time. Production subscribers usually resume from a saved checkpoint instead of starting fresh.

Gotchas: starting from new events means anything published earlier is silently skipped — this sample papers over that race with a fixed sleep_for instead of a ready signal, fine for a demo but not production. Replaying from the first event on every restart replays the whole log, which gets costly on a busy channel. Persistence isn't consumer coordination: each independent subscriber gets its own full replay unless grouped with a consumer group.

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: events_store/persistent_pubsub
//
// Demonstrates basic persistent event store publish/subscribe.
// Events are stored and can be replayed. This example sends an event
// and subscribes with StartFromNewEvents to receive only new events.
//
// Channel: cpp-events-store.persistent-pubsub
// Client ID: cpp-events-store-persistent-pubsub-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).

#include <kubemq/kubemq.h>

#include <chrono>
#include <iostream>
#include <thread>

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

    kubemq::ClientOptions options;
    options.set_address("localhost", 50000);
    options.set_client_id("cpp-events-store-persistent-pubsub-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-events-store.persistent-pubsub";

    // Subscribe to new events on the store channel.
    std::cout << "[2] Subscribing to channel " << channel << std::endl;
    auto sub_result = client->SubscribeToEventsStore(
        channel, "", kubemq::SubscriptionOption::StartFromNewEvents(),
        [](const kubemq::EventStoreReceive& e) {
            std::cout << "[5] Received: seq=" << e.sequence << " body=" << e.body << std::endl;
        },
        [](const kubemq::Status& err) {
            std::cerr << "[ERROR] Subscription error: " << err.message() << std::endl;
        });
    if (!sub_result.ok()) {
        std::cerr << "[ERROR] SubscribeToEventsStore: " << sub_result.status().message()
                  << std::endl;
        return 1;
    }
    auto& sub = *sub_result;

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

    // Send a persistent event.
    auto ev_result = kubemq::EventStore::Builder()
                         .SetChannel(channel)
                         .SetBody("persistent hello")
                         .SetMetadata("greeting")
                         .Build();
    if (!ev_result.ok()) {
        std::cerr << "[ERROR] Build event store: " << ev_result.status().message() << std::endl;
        return 1;
    }
    auto send_result = client->SendEventStore(*ev_result);
    if (!send_result.ok()) {
        std::cerr << "[ERROR] SendEventStore: " << send_result.status().message() << std::endl;
        return 1;
    }
    std::cout << "[3] Event stored: id=" << send_result->id << " sent=" << send_result->sent
              << std::endl;

    // Wait for the event to be received.
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "[4] Persistent pub/sub demo complete" << std::endl;

    // Cancel the subscription explicitly.
    // Note: The Subscription destructor also calls Cancel(), but explicit
    // cleanup is shown here for clarity and to match Go's defer pattern.
    sub->Cancel();

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

    return 0;
}

How It Works

Subscribes with SubscribeToEventsStore() using StartFromNewEvents() to receive only new events.
Builds a persistent event with EventStore::Builder() and sends with SendEventStore().
The result includes an id and sent boolean confirming storage.
Unlike regular events, stored events persist and can be replayed by future subscribers.

Persistent Pub/Sub

Overview

Prerequisites

Code

How It Works

On this page

Persistent Pub/Sub

Overview

Prerequisites

Code

How It Works

Related

On this page