Overview

This tutorial builds the "hello world" of KubeMQ messaging: a publisher and a subscriber talking over the Events pattern. Events are fire-and-forget — the broker fans a message out to every subscriber currently listening on the channel and moves on. There's no persistence, no acknowledgment, and no replay, which makes this the pattern to reach for when you need low-latency, high-throughput broadcast (metrics ticks, live status updates, cache-invalidation signals) and can tolerate losing a message if nobody is listening at the moment it's sent.

You'll wire up SubscribeToEventsAsync and drive it with await foreach over an EventsSubscription, give the subscription a moment to register with the server, then call SendEventAsync to publish an EventMessage. Every connected subscriber on the channel gets its own copy, as opposed to a consumer group where only one member would receive it. Gotchas: if the subscriber isn't fully established before you publish, the event is simply gone — there's no queue catching it, which is why the sample waits on a Task.Delay before sending; and because delivery isn't acknowledged, a crashed or disconnected subscriber never knows it missed anything.

Prerequisites

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

Code

Program.cs

// KubeMQ .NET SDK — Events: Basic Pub/Sub
//
// This example demonstrates publishing and subscribing to events on a channel.
// Events are fire-and-forget with no delivery guarantee.
//
// Prerequisites:
//   - KubeMQ server running on localhost:50000
//   - dotnet run
//
// See also:
//   - Events.WildcardSubscription for wildcard channel patterns
//   - EventsStore.PersistentPubSub for persistent events with replay

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

// TODO: Replace with your KubeMQ server address
await using var client = new KubeMQClient(new KubeMQClientOptions
{
    ClientId = "csharp-events-basic-pubsub-client",
});
await client.ConnectAsync();

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

// Start subscribing in the background
var cts = new CancellationTokenSource();
var subscribeTask = Task.Run(async () =>
{
    await foreach (var msg in client.SubscribeToEventsAsync(
        new EventsSubscription { Channel = "csharp-events.basic-pubsub" }, cts.Token))
    {
        var body = Encoding.UTF8.GetString(msg.Body.Span);
        Console.WriteLine($"Received event: {body}");
    }
});

// Allow time for subscription to establish
await Task.Delay(1000);

// Publish 5 events
for (var i = 1; i <= 5; i++)
{
    await client.SendEventAsync(new EventMessage
    {
        Channel = "csharp-events.basic-pubsub",
        Body = Encoding.UTF8.GetBytes($"Event #{i}"),
        Tags = new Dictionary<string, string> { ["source"] = "example" }
    });
    Console.WriteLine($"Published event #{i}");
}

// Wait for messages to arrive, then shut down
await Task.Delay(2000);
cts.Cancel();

Console.WriteLine("Done.");

// Expected output:
// Connected to KubeMQ server
// Published event #1
// Published event #2
// Published event #3
// Published event #4
// Published event #5
// Received event: Event #1
// Received event: Event #2
// Received event: Event #3
// Received event: Event #4
// Received event: Event #5
// Done.

How It Works

SubscribeToEventsAsync returns an IAsyncEnumerable<EventReceived> — the await foreach loop drives the gRPC subscription stream. The CancellationTokenSource controls teardown.
SendEventAsync serialises the message body as a ReadOnlyMemory<byte> and pushes it over a pooled gRPC channel to the broker.
Events are fire-and-forget: the broker delivers to all current subscribers but does not persist messages. If no subscriber is online, the event is dropped.
The 1-second Task.Delay before publishing gives the subscription stream time to handshake with the broker before the first event arrives.

Basic Pub/Sub

Overview

Prerequisites

Code

How It Works

Related

On this page