Overview

Replaying from a timestamp lets a consumer recover a window of history without knowing exact sequence numbers — you reach for it after a deploy, an outage, or any gap where you know roughly when you went dark but not where you left off in the stream. It turns an Events Store channel into a rewindable log: resubscribe with a point in time and the broker replays every event stored at or after it, then hands off to live delivery.

The subscription's option is set to kubemq::SubscriptionOption::StartFromTime(since) with a std::chrono::system_clock::time_point value — the broker compares this against the storage timestamp it assigned to each event, not any timestamp embedded in the payload. Because it's wall-clock based, the window is approximate rather than exact: pass a time far enough back to be safe.

Gotchas: clock skew between your subscriber's clock and the server's matters — favor a generous buffer over a precise cutoff. Storage timestamps reflect when the broker persisted the event, not when the producer created it, so under load the two can drift. And unlike sequence-based replay, a time-based start position has no way to guarantee "no gaps, no duplicates" across a network hiccup — use StartFromSequence instead if you need exact resumption.

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/replay_from_time
//
// Demonstrates subscribing to event store with StartFromTime.
// Events are replayed starting from a specific point in time.
//
// Channel: cpp-events-store.replay-from-time
// Client ID: cpp-events-store-replay-from-time-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>

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-replay-from-time-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.replay-from-time";
    std::atomic<int> received_count{0};

    // Subscribe starting from 1 hour ago -- replays events stored in the last hour.
    auto since = std::chrono::system_clock::now() - std::chrono::hours(1);
    std::cout << "[2] Subscribing with StartFromTime(now - 1 hour)" << std::endl;
    auto sub_result = client->SubscribeToEventsStore(
        channel, "", kubemq::SubscriptionOption::StartFromTime(since),
        [&received_count](const kubemq::EventStoreReceive& e) {
            std::cout << "[4] [StartFromTime] 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;

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

    // Send a new event that should be received.
    std::cout << "[3] Sending event within time window" << std::endl;
    auto ev_result = kubemq::EventStore::Builder()
                         .SetChannel(channel)
                         .SetBody("event within time window")
                         .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;
    }

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

Subscribes with StartFromTime(since) using a time point 1 hour in the past.
All events stored within the last hour are replayed.
Uses std::chrono::system_clock::now() for the time calculation.
Useful for reprocessing events from a specific moment.

Replay from Time

Overview

Prerequisites

Code

How It Works

On this page

Replay from Time

Overview

Prerequisites

Code

How It Works

Related

On this page