Persistent events with replay capabilities. Events are stored by the broker and can be replayed from a sequence number, timestamp, or the beginning.

Types

EventStore

Built via EventStore::Builder. Required field: channel.

auto event_or = kubemq::EventStore::Builder()
    .SetChannel("audit-log")
    .SetBody("user-login")
    .SetMetadata("auth")
    .Build();

Method	Type	Required	Description
`SetChannel(s)`	`string`	Yes	Target channel name
`SetBody(s)`	`string`	No	Message body
`SetMetadata(s)`	`string`	No	String metadata
`SetId(s)`	`string`	No	Event ID (auto-generated)
`SetClientId(s)`	`string`	No	Override client ID
`SetTags(m)`	`map<string,string>`	No	Key-value tags
`AddTag(k, v)`	`string, string`	No	Add a single tag

EventStoreResult

Acknowledgment from SendEventStore.

Field	Type	Description
`id`	`string`	Event identifier
`sent`	`bool`	Whether the event was stored
`error`	`string`	Error message if storage failed

EventStoreReceive

Received in subscription callbacks.

Field	Type	Description
`id`	`string`	Event identifier
`sequence`	`uint64_t`	Sequence number in the store
`timestamp`	`int64_t`	Unix nanoseconds
`channel`	`string`	Channel name
`metadata`	`string`	String metadata
`body`	`string`	Message body
`tags`	`map<string,string>`	Key-value tags

SubscriptionOption

Controls where to start reading from the event store.

Factory Method	Description
`StartFromNewEvents()`	Only new events after subscription
`StartFromFirstEvent()`	Replay from the first stored event
`StartFromLastEvent()`	Replay from the last stored event
`StartFromSequence(n)`	Replay from sequence number `n`
`StartFromTime(tp)`	Replay from a time point
`StartFromTimeDelta(d)`	Replay from `now - d`

Methods

SendEventStore

[[nodiscard]] StatusOr<EventStoreResult> SendEventStore(const EventStore& event);

Send a persistent event. Returns acknowledgment with sequence information.

PublishEventStore

[[nodiscard]] StatusOr<EventStoreResult> PublishEventStore(
    const std::string& channel, const std::string& body,
    const std::string& metadata = "",
    const std::unordered_map<std::string, std::string>& tags = {});

Convenience method to build and send a persistent event in one call.

SendEventStoreStream

[[nodiscard]] StatusOr<std::unique_ptr<EventStoreStreamHandle>> SendEventStoreStream(
    std::function<void(const EventStoreResult&)> on_result,
    std::function<void(const Status&)> on_error);

Open a persistent stream for high-throughput event store publishing. Each sent event receives a confirmation via the on_result callback.

SubscribeToEventsStore

[[nodiscard]] StatusOr<std::unique_ptr<Subscription>> SubscribeToEventsStore(
    const std::string& channel, const std::string& group,
    const SubscriptionOption& start_option,
    std::function<void(const EventStoreReceive&)> on_event,
    std::function<void(const Status&)> on_error);

Subscribe to persistent events with replay. The start_option parameter controls where to begin reading.

Quick Usage

events_store.cc

// Subscribe with replay from sequence 1
auto sub_or = client->SubscribeToEventsStore("ch", "",
    kubemq::SubscriptionOption::StartFromSequence(1),
    [](const kubemq::EventStoreReceive& e) {
        std::cout << "seq=" << e.sequence << " body=" << e.body << "\n";
    },
    [](const kubemq::Status& err) {
        std::cerr << "Error: " << err.message() << "\n";
    });

// Send
auto result = client->SendEventStore(event);
if (result.ok()) {
    std::cout << "Stored: id=" << result->id << "\n";
}

Events Store