# Pub/Sub and Events Store (/integrations/aspire/how-to/events)



## Overview [#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](/learn/events) for fire-and-forget pub/sub and [Events Store](/learn/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](https://www.nuget.org/packages/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:

```csharp title="KubeMQClientExtensions.cs (excerpt)"
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.

<Callout type="info">
  The `"messaging"` argument is the connection name. It must match the resource name in your AppHost (`builder.AddKubeMQ("messaging")`). See [Getting Started](/integrations/aspire/tutorials/getting-started) for the AppHost wiring.
</Callout>

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:

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

## Run a broker locally [#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:

<RunKubeMQ ports="[50000]" />

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 [#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).

```csharp title="Patterns.FanOut/Program.cs (publisher)"
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 [#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`.

```csharp title="Patterns.FanOut/Program.cs (subscriber)"
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](#graceful-shutdown).

## Fan-Out Semantics [#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.

```csharp title="Patterns.FanOut/Program.cs"
// 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.

<Mermaid
  chart="flowchart LR
    P[Publisher] -->|events.fanout| K[(KubeMQ)]
    K --> S1[Sub-1]
    K --> S2[Sub-2]
    K --> S3[Sub-3]"
/>

## Consumer Groups [#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.

```csharp title="PubSub.ConsumerGroup/Program.cs"
// 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) [#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`:

```csharp title="EventsStore.StartFromFirst/Program.cs (publish)"
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 [#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`.

```csharp title="EventsStore.StartFromFirst/Program.cs (replay)"
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 [#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:

<TypeTable
  type="{
  StartFromFirst: {
    type: 'EventStoreStartPosition',
    description: 'Replay every stored event from the beginning of the channel.',
  },
  StartFromLast: {
    type: 'EventStoreStartPosition',
    description: 'Start at the last stored event, then receive new ones going forward.',
  },
  StartFromNew: {
    type: 'EventStoreStartPosition',
    description: 'Skip all history; receive only events published after subscribing.',
  },
  StartAtSequence: {
    type: 'EventStoreStartPosition',
    description: 'Replay from a specific sequence number forward.',
  },
  StartAtTime: {
    type: 'EventStoreStartPosition',
    description: 'Replay from a specific point in time, set via StartTime.',
  },
}"
/>

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

```csharp title="EventsStore start positions (variations)"
// 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 [#web-vs-worker-hosting]

How you get the client depends on the host type:

<Tabs groupId="aspire-host" items="['Web API (controller)', 'Background worker']">
  <Tab value="Web API (controller)">
    In an ASP.NET Core service, inject `IKubeMQClient` straight into the controller via the constructor — the Aspire-registered singleton is resolved by DI. The client is already connected by the time a request arrives, so you publish directly:

    ```csharp title="EventsController.cs"
    [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" });
        }
    }
    ```
  </Tab>

  <Tab value="Background worker">
    In a Worker (or any console host), resolve the client from the built host and call `ConnectAsync` before subscribing. This is the pattern used by all the pub/sub and events store examples:

    ```csharp title="Worker Program.cs"
    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();
    ```
  </Tab>
</Tabs>

## Graceful Shutdown [#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:

```csharp
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);
```

<Callout type="info">
  Passing `stoppingToken` to `Task.Run` as well as to the subscribe call ensures both the wrapping task and the subscription stream observe the same cancellation signal, so the worker drains cleanly when Aspire stops the resource.
</Callout>

## Related [#related]

<Cards>
  <Card title="Events" href="/learn/events" description="The core KubeMQ Pub/Sub model — fan-out, channels, groups, and delivery semantics." />

  <Card title="Queues" href="/integrations/aspire/how-to/queues" description="Durable point-to-point messaging with competing consumers, ack/reject, and dead-letter handling." />

  <Card title="Commands & Queries" href="/integrations/aspire/how-to/commands-queries" description="Synchronous request-response messaging over the Aspire-injected client." />

  <Card title="Client API" href="/integrations/aspire/reference/client-api" description="Connection settings, health checks, and OpenTelemetry configuration for the Aspire client." />
</Cards>