Stream Send

Overview

Sending one queue message per call works fine for occasional traffic, but each call carries its own round trip. At high volume — event ingestion, sensor telemetry, log shipping — that per-call overhead caps your throughput well below what the connection can support.

SendQueueMessagesUpstreamAsync opens a bidirectional upstream stream and sends an entire batch of messages in a single gRPC call, returning an UpstreamSendResult with a per-message SentAt timestamp and IsError flag so you can confirm each message was individually accepted. Because the batch travels in one stream write instead of being serialized into one large request, this scales better than SendQueueMessagesAsync as payload size and batch count grow.

Gotchas: IsError is set per message, not just on the overall result — a batch can partially succeed, so check every entry in result.Results rather than trusting the top-level status alone. This call is one-shot per batch; a truly persistent multi-batch stream would hold the connection open across calls instead of invoking this per batch. On the receiving side, these are ordinary queue messages — consume with CreateQueueDownstreamReceiverAsync and manual ack like any other queue message.

Prerequisites

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

Code

// KubeMQ .NET SDK — QueuesStream: Stream Send
//
// This example demonstrates sending multiple queue messages via the upstream stream API.
// The upstream stream opens a bidirectional gRPC stream for efficient batch sending.
//
// Prerequisites:
//   - KubeMQ server running on localhost:50000
//   - dotnet run

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

await using var client = new KubeMQClient(new KubeMQClientOptions
{
    ClientId = "csharp-queues-stream-send-client",
});
await client.ConnectAsync();

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

var messages = new[]
{
    new QueueMessage { Channel = "csharp-queues.stream-send", Body = Encoding.UTF8.GetBytes("Message 1") },
    new QueueMessage { Channel = "csharp-queues.stream-send", Body = Encoding.UTF8.GetBytes("Message 2") },
    new QueueMessage { Channel = "csharp-queues.stream-send", Body = Encoding.UTF8.GetBytes("Message 3") },
};

var result = await client.SendQueueMessagesUpstreamAsync(messages);
Console.WriteLine($"Upstream send: IsError={result.IsError}, Results={result.Results.Count}");

foreach (var r in result.Results)
{
    Console.WriteLine($"  {r.MessageId}: SentAt={r.SentAt}, IsError={r.IsError}");
}

Console.WriteLine("Done.");

How It Works

• SendQueueMessagesUpstreamAsync opens a bidirectional upstream stream and sends all messages in a single gRPC call, returning a UpstreamSendResult with per-message results.
• Each r.SentAt timestamp and r.IsError flag lets you verify that every message was individually accepted by the broker.
• This is more efficient than batching via SendQueueMessagesAsync for large payloads because it avoids serializing the entire batch into one request.
• For receiving these messages, use CreateQueueDownstreamReceiverAsync + PollAsync with manual ack as shown in the Stream Receive example.

Related

• Pattern overview
• C# SDK Reference
• Send & Receive
• Ack All