Persistent Pub/Sub

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: SendEventStoreAsync publishes and returns a result confirming storage plus a broker-assigned sequence number, and SubscribeToEventsStoreAsync takes a required EventStoreStartPosition telling the broker where to start — new events only, from the first stored event (StartFromFirst, used here), or a given sequence or time. Production subscribers usually resume from a saved checkpoint instead of replaying from the beginning.

Gotchas: replaying from the first event on every restart replays the whole log, which gets costly on a busy channel — track the last Sequence you processed instead. Starting from new-only has the opposite risk: anything published earlier is silently skipped, so don't rely on a fixed Task.Delay like this sample's to paper over that race in production. 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 (dotnet add package KubeMQ.SDK.CSharp)

Code

// KubeMQ .NET SDK — Events Store: Persistent Pub/Sub
//
// This example demonstrates publishing and subscribing with server-side persistence.
// Events Store messages are stored by the server and can be replayed.
//
// Prerequisites:
//   - KubeMQ server running on localhost:50000
//   - dotnet run

using KubeMQ.Sdk.Client;
using KubeMQ.Sdk.EventsStore;
using System.Text;

await using var client = new KubeMQClient(new KubeMQClientOptions
{
    ClientId = "csharp-eventsstore-persistent-pubsub-client",
});
await client.ConnectAsync();

Console.WriteLine("Connected to KubeMQ server");

// Publish events first — they are persisted
for (var i = 1; i <= 5; i++)
{
    await client.SendEventStoreAsync(new EventStoreMessage
    {
        Channel = "csharp-eventsstore.persistent-pubsub",
        Body = Encoding.UTF8.GetBytes($"Persistent Event #{i}")
    });
    Console.WriteLine($"Published persistent event #{i}");
}

// Subscribe from the beginning — replays all stored events
var cts = new CancellationTokenSource();
var subscribeTask = Task.Run(async () =>
{
    await foreach (var msg in client.SubscribeToEventsStoreAsync(
        new EventStoreSubscription
        {
            Channel = "csharp-eventsstore.persistent-pubsub",
            StartPosition = EventStoreStartPosition.StartFromFirst
        }, cts.Token))
    {
        Console.WriteLine($"[Seq={msg.Sequence}] {Encoding.UTF8.GetString(msg.Body.Span)}");
    }
});

await Task.Delay(3000);
cts.Cancel();

Console.WriteLine("Done.");

// Expected output:
// Connected to KubeMQ server
// Published persistent event #1
// Published persistent event #2
// Published persistent event #3
// Published persistent event #4
// Published persistent event #5
// [Seq=<seq>] Persistent Event #1
// [Seq=<seq>] Persistent Event #2
// [Seq=<seq>] Persistent Event #3
// [Seq=<seq>] Persistent Event #4
// [Seq=<seq>] Persistent Event #5
// Done.

How It Works

• SendEventStoreAsync publishes a message and the broker persists it to the channel's durable log before acknowledging.
• SubscribeToEventsStoreAsync with StartFromFirst replays every stored message from the beginning of the log, then continues streaming new events. No messages are lost even if the subscriber was offline during publishing.
• EventStoreStartPosition controls the replay window: StartFromFirst, StartFromLast, StartFromNew, StartAtSequence, and StartAtTimeDelta are all available.
• The 3-second Task.Delay gives the replay time to complete before the cancellation token fires and the subscription loop exits.

Related

• Pattern overview
• C# SDK Reference
• Cancel Subscription
• Consumer Group