# Stream Receive (/sdks/csharp/how-to/queues/stream-receive)



## Overview [#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 [#prerequisites]

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

## Code [#code]

```csharp title="Program.cs"
// 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 [#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 [#related]

* [Pattern overview](/learn/queues/getting-started)
* [C# SDK Reference](/sdks/csharp/reference)
* [Send & Receive](/sdks/csharp/tutorials/send-receive)
* [Ack All](/sdks/csharp/how-to/queues/ack-all)