Pub/Sub and Events Store

Overview

This page covers how to use KubeMQ Pub/Sub Events and the persistent Events Store from a service wired by the Aspire integration. For what events are and their delivery semantics, see the core docs — Events for fire-and-forget pub/sub and Events Store for persistent, replayable streams. Below is the Aspire-specific part: how the injected IKubeMQClient exposes those APIs.

Once AddKubeMQClient has registered IKubeMQClient in your service, you work directly with the native KubeMQ.SDK.CSharp Events and Events Store APIs. The Aspire integration does not wrap or replace those APIs — it only handles connection string injection, health checks, and OpenTelemetry tracing/metrics wiring. The AddKubeMQClient extension binds the Aspire-provided connection string to the SDK options and registers the client as a singleton, then layers observability on top:

public static void AddKubeMQClient(
    this IHostApplicationBuilder builder,
    string connectionName,
    Action<KubeMQClientSettings>? configureSettings = null,
    Action<KubeMQClientOptions>? configureOptions = null)
{
    // ... binds connection string, then:
    builder.Services.AddKubeMQ(opts =>
    {
        ApplySettings(opts, settings, host, port);
        configureOptions?.Invoke(opts);
    });

    RegisterTlsWarningAndObservability(builder, settings, connectionName);
}

Everything on this page — publishing, subscribing, consumer groups, persistent replay — is plain KubeMQ.SDK.CSharp usage. The only Aspire-specific lines are builder.AddKubeMQClient("messaging") and resolving the client from DI.

The examples below come from a Worker project (Microsoft.NET.Sdk.Worker, targeting net8.0) that references KubeMQ.Aspire.Client. The relevant SDK namespaces are:

using KubeMQ.Sdk.Client;       // IKubeMQClient
using KubeMQ.Sdk.Events;       // EventMessage, EventsSubscription
using KubeMQ.Sdk.EventsStore;  // EventStoreMessage, EventStoreSubscription, EventStoreStartPosition

Run a broker locally

When you run the Aspire AppHost, it provisions the KubeMQ container for you. If you instead want to run a broker directly for a standalone service, start one in Docker:

docker run -d \  --name kubemq \  -p 50000:50000 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Port 50000 is the gRPC port used by KubeMQ.SDK.CSharp (and therefore by the Aspire-injected client). The gRPC server is always on — the SDK path is a native gRPC client and needs no connector enable flag.

Basic Publish

Resolve IKubeMQClient from the host, call ConnectAsync, then SendEventAsync with an EventMessage. Events are fire-and-forget: the call returns as soon as the broker accepts the message, and every active subscriber on the channel receives a copy (multicast).

var builder = Host.CreateApplicationBuilder(args);
builder.AddServiceDefaults();
builder.AddKubeMQClient("messaging");

var host = builder.Build();
var client = host.Services.GetRequiredService<IKubeMQClient>();
var lifetime = host.Services.GetRequiredService<IHostApplicationLifetime>();

await client.ConnectAsync();
var stoppingToken = lifetime.ApplicationStopping;

for (var i = 1; i <= 5; i++)
{
    await client.SendEventAsync(new EventMessage
    {
        Channel = "events.fanout",
        Body = Encoding.UTF8.GetBytes($"Broadcast #{i}")
    }, stoppingToken);
    await Task.Delay(1000, stoppingToken);
}

The payload travels as raw bytes (Body), so encode and decode it however your application prefers — UTF-8 strings, JSON, or any binary format.

Basic Subscribe

Subscriptions are exposed as an IAsyncEnumerable, so you consume them with await foreach. Pass an EventsSubscription with the Channel and a cancellation token; the loop yields one EventReceiveMessage per delivered event, whose payload is in ev.Body.Span.

var subscription = new EventsSubscription { Channel = "events.fanout" };
await foreach (var ev in client.SubscribeToEventsAsync(subscription, stoppingToken))
{
    logger.LogInformation("Received: {Body}", Encoding.UTF8.GetString(ev.Body.Span));
}

The await foreach runs until the cancellation token is signalled (typically at host shutdown), at which point it throws OperationCanceledException — see Graceful Shutdown.

Fan-Out Semantics

Every subscriber on a channel receives every message published to it. This is the default pub/sub behaviour: there is no load balancing unless you opt into a consumer group. The Fan-Out example starts three subscribers on events.fanout, then a publisher broadcasts five messages — each subscriber sees all five.

// 3 subscribers — all receive every message
for (var s = 1; s <= 3; s++)
{
    var subId = s;
    _ = Task.Run(async () =>
    {
        try
        {
            var subscription = new EventsSubscription { Channel = "events.fanout" };
            await foreach (var ev in client.SubscribeToEventsAsync(subscription, stoppingToken))
            {
                logger.LogInformation("[Sub-{Id}] {Body}", subId,
                    Encoding.UTF8.GetString(ev.Body.Span));
            }
        }
        catch (OperationCanceledException) { }
    }, stoppingToken);
}

Each Task.Run runs an independent subscription loop. Because the channel has no group, the broker fans every Broadcast #i out to all three loops.

Consumer Groups

To load-balance instead of fan out, add a Group to the EventsSubscription. All subscribers that share the same group name form a competing-consumer set: each message goes to exactly one member of the group. This is how you scale out a worker pool — add more instances with the same group and the broker spreads the load across them.

// Worker A — group "workers"
_ = Task.Run(async () =>
{
    try
    {
        var subscription = new EventsSubscription
        {
            Channel = "events.group",
            Group = "workers" // group name for load balancing
        };
        await foreach (var ev in client.SubscribeToEventsAsync(subscription, stoppingToken))
        {
            var body = Encoding.UTF8.GetString(ev.Body.Span);
            logger.LogInformation("[Worker-A] Received: {Body}", body);
        }
    }
    catch (OperationCanceledException) { /* shutting down */ }
}, stoppingToken);

// Worker B — same group "workers"
_ = Task.Run(async () =>
{
    try
    {
        var subscription = new EventsSubscription
        {
            Channel = "events.group",
            Group = "workers"
        };
        await foreach (var ev in client.SubscribeToEventsAsync(subscription, stoppingToken))
        {
            var body = Encoding.UTF8.GetString(ev.Body.Span);
            logger.LogInformation("[Worker-B] Received: {Body}", body);
        }
    }
    catch (OperationCanceledException) { /* shutting down */ }
}, stoppingToken);

When the publisher sends ten Task #i events to events.group, the broker distributes them across Worker-A and Worker-B — each task is handled once, not twice. Subscribers on the same channel but in different groups (or with no group at all) still receive their own copy of every message.

Events Store (Persistent Events)

Plain events are not retained: if nobody is subscribed when an event is published, it is gone. The Events Store persists events on the channel so they can be replayed later. Publishing uses SendEventStoreAsync with an EventStoreMessage:

for (var i = 1; i <= 5; i++)
{
    await client.SendEventStoreAsync(new EventStoreMessage
    {
        Channel = "store.first",
        Body = Encoding.UTF8.GetBytes($"Historical event #{i}")
    }, stoppingToken);
}
logger.LogInformation("Published 5 events to store.first");

Persisted events keep an ordered, broker-assigned sequence number, which is what makes replay possible.

Replay

To consume persisted history, subscribe with SubscribeToEventsStoreAsync and an EventStoreSubscription. The StartPosition controls where the replay begins; EventStoreStartPosition.StartFromFirst replays the entire history from the first stored event. Each delivered message exposes its ev.Sequence alongside ev.Body.

var subscription = new EventStoreSubscription
{
    Channel = "store.first",
    StartPosition = EventStoreStartPosition.StartFromFirst
};
await foreach (var ev in client.SubscribeToEventsStoreAsync(subscription, stoppingToken))
{
    var body = Encoding.UTF8.GetString(ev.Body.Span);
    logger.LogInformation("[Replay] Seq={Sequence}: {Body}", ev.Sequence, body);
}

A fresh subscriber on store.first replays all five historical events in order, logging Seq=1 through Seq=5, and then continues to receive any new events published afterward.

Other Replay Positions

StartFromFirst is one of several start positions exercised across the examples suite (see the EventsStore region of the AppHost). The full set is:

For example, the time-based and sequence-based examples set an extra field on the subscription:

// Start at a specific point in time
var byTime = new EventStoreSubscription
{
    Channel = "store.time",
    StartPosition = EventStoreStartPosition.StartAtTime,
    StartTime = startTime
};

// Replay from a specific sequence number forward
var bySequence = new EventStoreSubscription
{
    Channel = "store.sequence",
    StartPosition = EventStoreStartPosition.StartAtSequence,
    StartSequence = 5 // begin at sequence 5
};

The AppHost registers a dedicated example project for each position — StartFromFirst, StartFromLast, StartNewOnly, StartAtTime, ReplayFromSequence, and ReplayFromTime — so you can run any one of them against the provisioned broker.

Web vs Worker Hosting

How you get the client depends on the host type:

[ApiController]
[Route("api/[controller]")]
public sealed class EventsController : ControllerBase
{
    private readonly IKubeMQClient _client;

    public EventsController(IKubeMQClient client) => _client = client;

    [HttpPost]
    public async Task<IActionResult> PublishEvent([FromBody] string body)
    {
        var message = new EventMessage
        {
            Channel = "events.example",
            Body = Encoding.UTF8.GetBytes(body),
            Tags = new Dictionary<string, string> { ["source"] = "aspire-sample" },
        };

        await _client.SendEventAsync(message);
        return Ok(new { Status = "published" });
    }
}

var builder = Host.CreateApplicationBuilder(args);
builder.AddServiceDefaults();
builder.AddKubeMQClient("messaging");

var host = builder.Build();
var client = host.Services.GetRequiredService<IKubeMQClient>();
var lifetime = host.Services.GetRequiredService<IHostApplicationLifetime>();

await client.ConnectAsync();
var stoppingToken = lifetime.ApplicationStopping;

var subscription = new EventsSubscription { Channel = "events.fanout" };
await foreach (var ev in client.SubscribeToEventsAsync(subscription, stoppingToken))
{
    // handle ev.Body.Span
}

await host.RunAsync();

Graceful Shutdown

Long-lived subscriptions should be tied to the host lifecycle. Resolve IHostApplicationLifetime and pass its ApplicationStopping token into every SubscribeToEventsAsync / SubscribeToEventsStoreAsync call. When the host begins shutting down, the token is cancelled, the await foreach loop terminates, and you swallow the resulting OperationCanceledException so shutdown stays clean:

var lifetime = host.Services.GetRequiredService<IHostApplicationLifetime>();
var stoppingToken = lifetime.ApplicationStopping;

_ = Task.Run(async () =>
{
    try
    {
        var subscription = new EventsSubscription { Channel = "events.group", Group = "workers" };
        await foreach (var ev in client.SubscribeToEventsAsync(subscription, stoppingToken))
        {
            logger.LogInformation("Received: {Body}", Encoding.UTF8.GetString(ev.Body.Span));
        }
    }
    catch (OperationCanceledException) { /* shutting down */ }
}, stoppingToken);

Related