Cancel Subscription

Overview

A live Events subscription holds a client-side gRPC stream open indefinitely, so a long-running service needs an explicit way to tear one down without closing the whole client connection — for example when a feature flag disables a channel, a worker is draining before shutdown, or a subscription needs to be re-created with different options. Passing a CancellationToken into SubscribeToEventsAsync gives you exactly that: a standard .NET cancellation mechanism instead of a bespoke unsubscribe API.

SubscribeToEventsAsync returns an IAsyncEnumerable<Event> that the await foreach loop iterates over as events arrive. Cancelling the CancellationTokenSource behind the token — whether via a timer, a manual Cancel() call, or an external signal — propagates into the underlying gRPC stream reader, which unwinds the loop and throws OperationCanceledException; catching that exception is the normal, expected way to detect a clean cancellation rather than an error.

Gotchas: cancellation only affects this one subscription — a shared Group on the EventsSubscription keeps delivering to other instances in the group. Events already in flight when the token fires may still be yielded before the loop unwinds. And because Events are fire-and-forget, anything published after cancellation reaches the broker but is simply dropped for this subscriber — there's no queue to catch up from later.

Prerequisites

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

Code

// KubeMQ .NET SDK — Events: Cancel Subscription
//
// This example demonstrates cancelling an event 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.Events;
using System.Text;

var options = new KubeMQClientOptions
{
    Address = "localhost:50000",
    ClientId = "csharp-events-cancel-subscription-client",
};

await using var client = new KubeMQClient(options);
await client.ConnectAsync();

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

var subscription = new EventsSubscription
{
    Channel = "csharp-events.cancel-subscription",
};

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

How It Works

• CancellationTokenSource(TimeSpan.FromSeconds(10)) auto-fires after 10 seconds — the token is passed into SubscribeToEventsAsync, which propagates it into the underlying gRPC stream reader.
• When the token fires, the await foreach exits and throws OperationCanceledException; the catch block converts that into a clean log message.
• The subscription object stores the channel name; pass a Group property to share the subscription across multiple instances for load-balanced delivery.

Related

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