Overview

A subscriber that just restarted usually doesn't need the entire event history — it needs to know where things stand right now without paying the cost of replaying everything that happened while it was offline. kubemq::SubscriptionOption::StartFromLastEvent() solves that: it re-anchors a new subscription to the tail of the store, delivering exactly one historical event (the most recently stored one) before switching to live delivery. That's the sweet spot between starting from new (no history at all, so you might miss the current state entirely) and starting from first (the full backlog, which can be slow and mostly irrelevant for a consumer that only cares about "now").

Under the hood, StartFromLastEvent() is passed as a subscription option when subscribing to the events store. The broker looks up the channel's most recent stored event at subscription time, replays that single event to the new subscriber, and then streams every subsequently published event as it arrives — the same live path any other subscription uses.

Gotchas: if the channel is empty when you subscribe, there's no "last" event to deliver — you simply start receiving new events as they're published, with no error raised. StartFromLastEvent gives you one event, not the last N — if you need a short window of recent history, replay from a sequence number instead. And because "last" is resolved at subscribe time, two subscribers starting a few events apart can each get a different one.

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/start_from_last
//
// Demonstrates subscribing to event store with StartFromLastEvent.
// The last stored event is replayed, then new events continue.
//
// Channel: cpp-events-store.start-from-last
// Client ID: cpp-events-store-start-from-last-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 <string>
#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-start-from-last-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.start-from-last";

    // Send some events so there is a "last" event.
    for (int i = 1; i <= 3; i++) {
        auto ev_result = kubemq::EventStore::Builder()
                             .SetChannel(channel)
                             .SetBody("msg-" + std::to_string(i))
                             .Build();
        if (!ev_result.ok()) {
            std::cerr << "[ERROR] Build: " << 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 << "[2] Sent 3 events" << std::endl;

    std::atomic<int> received_count{0};

    // Subscribe with StartFromLastEvent -- starts from the most recent stored event.
    std::cout << "[3] Subscribing with StartFromLastEvent" << std::endl;
    auto sub_result = client->SubscribeToEventsStore(
        channel, "", kubemq::SubscriptionOption::StartFromLastEvent(),
        [&received_count](const kubemq::EventStoreReceive& e) {
            std::cout << "[4] [StartFromLast] seq=" << e.sequence << " body=" << e.body
                      << std::endl;
            received_count.fetch_add(1);
        },
        [](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;

    // Wait for events to arrive.
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "[5] Received " << received_count.load() << " event(s) from last" << std::endl;
    std::cout << "[6] Start from last 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 << "[7] Client closed" << std::endl;

    return 0;
}

How It Works

Sends 3 events to ensure there is a last event in the store.
Subscribes with StartFromLastEvent() to receive the most recent event and then new events.
Only the last stored event is replayed (not the full history).
Useful for getting the current state before processing new events.

Start from Last

Overview

Prerequisites

Code

How It Works

On this page

Start from Last

Overview

Prerequisites

Code

How It Works

Related

On this page