Stream Receive

Overview

A downstream receiver is the persistent-connection way to pull queue messages: instead of opening and tearing down a request for every batch, you open one gRPC stream and reuse it across many poll cycles. That matters for any consumer that runs continuously — a worker loop, a background processor — where reconnecting per batch would add latency and churn on both the client and the broker.

The receiver is created once with CreateQueueDownstreamReceiverAsync, then each call to PollAsync fetches a batch with AutoAck = false so nothing is removed from the queue until you explicitly settle it. Acknowledging the batch as a whole with batch.AckAllAsync() permanently removes every message in one round-trip, while leaving messages unacknowledged returns them for redelivery once the visibility timeout expires.

Gotchas: an unclosed receiver holds server-side state — the await using pattern here half-closes the stream on disposal, so don't skip it; a crash between receiving and acknowledging redelivers the whole batch, so processing must be idempotent; and AckAllAsync is all-or-nothing — use per-message AckAsync/NackAsync instead when only some messages in a batch succeed.

Prerequisites

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

Code

// KubeMQ .NET SDK — QueuesStream: Stream Receive
//
// This example demonstrates receiving messages from a queue using the downstream
// receiver API with manual acknowledgment. Messages are polled and then
// acknowledged as a batch using batch.AckAllAsync().
//
// Prerequisites:
//   - KubeMQ server running on localhost:50000
//   - Send some messages to "csharp-queues.stream-receive" first
//   - 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-receive-client",
});
await client.ConnectAsync();

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

// Send some test messages first
for (var i = 1; i <= 3; i++)
{
    await client.SendQueueMessageAsync(new QueueMessage
    {
        Channel = "csharp-queues.stream-receive",
        Body = Encoding.UTF8.GetBytes($"Stream message #{i}"),
    });
}

Console.WriteLine("Sent 3 messages");

await using var receiver = await client.CreateQueueDownstreamReceiverAsync();

var batch = await receiver.PollAsync(new QueuePollRequest
{
    Channel = "csharp-queues.stream-receive",
    MaxMessages = 10,
    WaitTimeoutSeconds = 5,
    AutoAck = false,
});

Console.WriteLine($"Received {batch.Messages.Count} messages");

foreach (var msg in batch.Messages)
{
    Console.WriteLine($"  {msg.MessageId}: {Encoding.UTF8.GetString(msg.Body.Span)}");
}

if (batch.HasMessages)
{
    await batch.AckAllAsync();
    Console.WriteLine("All messages acknowledged.");
}

Console.WriteLine("Done.");

How It Works

• CreateQueueDownstreamReceiverAsync opens a reusable downstream gRPC stream. The await using pattern ensures the stream is half-closed and resources are released on exit.
• PollAsync with AutoAck = false locks the returned messages for this receiver. batch.AckAllAsync() acknowledges the entire batch in one round-trip to the broker.
• Use AckAllAsync when all messages in a batch are processed atomically; use per-message AckAsync / NackAsync for partial success scenarios.
• The receiver is separate from the client — you can create multiple receivers (e.g., per-thread or per-goroutine) without creating multiple KubeMQClient instances.

Related

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