Stream Send

Overview

Publishing events one at a time means each call pays its own round-trip: write the request, wait on the connection, then move to the next event. That's fine for occasional notifications, but it caps throughput when you need to push hundreds or thousands of events per second — log forwarding, sensor telemetry, change-data-capture feeds — where per-call overhead dominates.

client.CreateEventStreamAsync() opens one bidirectional gRPC stream up front. Each subsequent await stream.SendAsync(msg, clientId) writes a message onto that already-open stream instead of negotiating a new call, so a sender loop isn't blocked waiting on a broker round-trip for every event.

Gotchas: because sends don't wait on a per-message round-trip, write failures surface as exceptions from SendAsync rather than as an immediate broker acknowledgement — make sure your loop actually observes them. Events are still fire-and-forget pub/sub underneath: no subscriber means a streamed event is dropped just like a regular one. Always CloseAsync() (or dispose via await using) when you're done; a stream left open holds a gRPC connection on the broker.

Prerequisites

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

Code

// KubeMQ .NET SDK — Events: Stream Publish
//
// This example demonstrates high-throughput event publishing using a bidirectional stream.
// Stream publishing reuses a single gRPC stream for many events, reducing overhead.
//
// 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-stream-send-client",
};

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

await using var stream = await client.CreateEventStreamAsync(
    onError: ex => Console.WriteLine($"Stream error: {ex.Message}"));

for (int i = 0; i < 100; i++)
{
    var msg = new EventMessage
    {
        Channel = "csharp-events.stream-send",
        Body = Encoding.UTF8.GetBytes($"Event #{i}"),
    };
    await stream.SendAsync(msg, options.ClientId!);
}

await stream.CloseAsync();
Console.WriteLine("Sent 100 events via stream.");

How It Works

• CreateEventStreamAsync opens a single bidirectional gRPC stream. All 100 sends reuse it, avoiding the per-call HTTP/2 handshake overhead of SendEventAsync.
• stream.SendAsync(msg, clientId) writes a message to the stream's request channel and returns a Task backed by a response slot — errors surface as exceptions rather than silent drops.
• CloseAsync() sends a half-close, flushing the write side and allowing the server to drain any buffered confirmations before the stream closes.
• Use stream publishing when you need to push many events in quick succession; for infrequent events, SendEventAsync is simpler.

Related

• Pattern overview
• C# SDK Reference
• Basic Pub/Sub
• Cancel Subscription