Cancel Subscription

Overview

Every events store subscription opens a long-lived stream to the broker — an IAsyncEnumerable reader that keeps pulling delivered events until you tell it to stop. Cancelling the CancellationToken passed to SubscribeToEventsStoreAsync is how you release that stream deliberately: shutting down a worker, rotating consumers, or tearing down a request without leaking connections or leaving a dangling stream on the server.

Internally, the token is threaded through to the underlying gRPC stream reader; when it fires, the await foreach loop throws OperationCanceledException, which is the expected signal to unwind cleanly rather than a failure.

Gotchas: cancelling only stops this subscription — the channel keeps storing every event published afterward, so nothing is lost, and a fresh subscription with StartFromFirst (or a specific sequence) picks up exactly where this one left off. You must catch OperationCanceledException yourself; letting it propagate unhandled looks like a crash even though it's a normal shutdown path.

Prerequisites

• KubeMQ server running on localhost:50000
• C# SDK installed (dotnet add package KubeMQ.SDK.CSharp)

Code

// KubeMQ .NET SDK — Events Store: Cancel Subscription
//
// This example demonstrates cancelling an event store subscription after a timeout.
// Uses CancellationTokenSource to auto-cancel after 10 seconds.
//
// 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
{
    Address = "localhost:50000",
    ClientId = "csharp-eventsstore-cancel-subscription-client",
});
await client.ConnectAsync();

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

var subscription = new EventStoreSubscription
{
    Channel = "csharp-eventsstore.cancel-subscription",
    StartPosition = EventStoreStartPosition.StartFromNew,
};

Console.WriteLine("Subscribing for 10 seconds...");
try
{
    await foreach (var evt in client.SubscribeToEventsStoreAsync(subscription, cts.Token))
    {
        Console.WriteLine($"[Seq {evt.Sequence}] {Encoding.UTF8.GetString(evt.Body.Span)}");
    }
}
catch (OperationCanceledException)
{
    Console.WriteLine("Subscription cancelled.");
}

How It Works

• StartFromNew means the subscription only receives events published after it is established — no replay of history. Use StartFromFirst to replay all stored events.
• The CancellationTokenSource(TimeSpan.FromSeconds(10)) fires automatically after 10 seconds. The token is threaded through SubscribeToEventsStoreAsync, which propagates it to the underlying gRPC stream reader.
• OperationCanceledException is the expected signal: the catch block converts it into a clean shutdown message rather than a stack trace.
• To cancel programmatically at any time, call cts.Cancel() from another thread or on a condition.

Related

• Pattern overview
• C# SDK Reference
• Persistent Pub/Sub
• Consumer Group