# Queues (/integrations/aspire/how-to/queues) ## Overview [#overview] This page covers how to send, receive, and settle **KubeMQ Queues** from a service wired by the Aspire integration. For the queue model itself — durability, competing consumers, visibility, and dead-letter handling — see the core [Queues](/learn/queues) docs. Below is the Aspire-specific part: how the injected `IKubeMQClient` exposes the Queues API. Queues provide **durable, point-to-point messaging** with competing consumers. Unlike Pub/Sub, queue messages are persisted on the broker until a consumer pulls and settles them — so producers and consumers do not need to be online at the same time, and each message is delivered to exactly one consumer. When several consumers poll the same channel, the broker distributes the backlog among them, giving you a work queue that scales horizontally. With the .NET Aspire integration you never construct the client by hand. The AppHost provisions a KubeMQ container, and `builder.AddKubeMQClient("messaging")` registers an `IKubeMQClient` in the DI container. You resolve it (or inject it into a controller) and call the SDK's Queues API directly. ```csharp title="Program.cs" var builder = Host.CreateApplicationBuilder(args); builder.AddServiceDefaults(); builder.AddKubeMQClient("messaging"); var host = builder.Build(); var client = host.Services.GetRequiredService(); await client.ConnectAsync(); ``` `"messaging"` is the Aspire connection name — it must match the resource name passed to `AddKubeMQ("messaging")` in your AppHost. The connection string (broker address, port, credentials) is injected automatically by Aspire's service discovery. See [Getting Started](/integrations/aspire/tutorials/getting-started) for the AppHost wiring. The diagram below shows the durable, competing-consumers flow: a producer enqueues messages, and two workers polling the same channel each pull a share of the backlog. ## Sending Messages [#sending-messages] A `QueueMessage` carries a `Channel` and a `Body`. Call `SendQueueMessageAsync` and inspect the result: it exposes `MessageId`, `IsError`, and `Error` so you can confirm the broker accepted the message. ```csharp title="QueuesController.cs" using System.Text; using KubeMQ.Sdk.Client; using KubeMQ.Sdk.Queues; using Microsoft.AspNetCore.Mvc; [ApiController] [Route("api/[controller]")] public sealed class QueuesController : ControllerBase { private readonly IKubeMQClient _client; public QueuesController(IKubeMQClient client) => _client = client; [HttpPost] public async Task SendMessage([FromBody] string body) { var message = new QueueMessage { Channel = "queues.example", Body = Encoding.UTF8.GetBytes(body), }; var result = await _client.SendQueueMessageAsync(message); return Ok(new { result.MessageId, result.IsError, result.Error }); } } ``` ## Receiving with Auto-Ack [#receiving-with-auto-ack] To pull messages, build a `QueuePollRequest`. `MaxMessages` caps how many messages a single poll returns; `WaitTimeoutSeconds` is the long-poll window the broker waits for messages to arrive before returning an empty batch. Setting `AutoAck = true` tells the broker to acknowledge every message the moment it is delivered — the simplest mode, suited to work that cannot fail or that you do not need to retry. Iterate `response.Messages` to process the batch: ```csharp title="QueuesController.cs" [HttpGet] public async Task ReceiveMessages() { var request = new QueuePollRequest { Channel = "queues.example", MaxMessages = 5, WaitTimeoutSeconds = 5, AutoAck = true, }; var response = await _client.ReceiveQueueMessagesAsync(request); var messages = response.Messages.Select(m => new { m.MessageId, Body = Encoding.UTF8.GetString(m.Body.Span), }).ToList(); return Ok(new { Count = messages.Count, Messages = messages }); } ``` ## Work Queue: Competing Consumers [#work-queue-competing-consumers] The defining property of queues is competing consumers. Run multiple workers that poll the **same channel**, and the broker hands each message to exactly one of them — load-balancing the backlog automatically. No worker sees a message another worker already took. The example below enqueues ten tasks, then starts two background workers (`Worker-A` and `Worker-B`) that each long-poll with `MaxMessages = 1`, `WaitTimeoutSeconds = 5`, and `AutoAck = true`. Because they share the `queues.workqueue` channel, the ten tasks are split between them. ```csharp title="Patterns.WorkQueue/Program.cs" using System.Text; using KubeMQ.Sdk.Client; using KubeMQ.Sdk.Queues; var builder = Host.CreateApplicationBuilder(args); builder.AddServiceDefaults(); builder.AddKubeMQClient("messaging"); var host = builder.Build(); var client = host.Services.GetRequiredService(); var logger = host.Services.GetRequiredService>(); var lifetime = host.Services.GetRequiredService(); await client.ConnectAsync(); var stoppingToken = lifetime.ApplicationStopping; // Produce 10 tasks for (var i = 1; i <= 10; i++) { await client.SendQueueMessageAsync(new QueueMessage { Channel = "queues.workqueue", Body = Encoding.UTF8.GetBytes($"Task #{i}") }); } logger.LogInformation("Produced 10 tasks"); // Worker A _ = Task.Run(async () => { try { while (!stoppingToken.IsCancellationRequested) { var response = await client.ReceiveQueueMessagesAsync(new QueuePollRequest { Channel = "queues.workqueue", MaxMessages = 1, WaitTimeoutSeconds = 5, AutoAck = true }, stoppingToken); foreach (var msg in response.Messages) { logger.LogInformation("[Worker-A] {Body}", Encoding.UTF8.GetString(msg.Body.Span)); } } } catch (OperationCanceledException) { } }, stoppingToken); // Worker B _ = Task.Run(async () => { try { while (!stoppingToken.IsCancellationRequested) { var response = await client.ReceiveQueueMessagesAsync(new QueuePollRequest { Channel = "queues.workqueue", MaxMessages = 1, WaitTimeoutSeconds = 5, AutoAck = true }, stoppingToken); foreach (var msg in response.Messages) { logger.LogInformation("[Worker-B] {Body}", Encoding.UTF8.GetString(msg.Body.Span)); } } } catch (OperationCanceledException) { } }, stoppingToken); await host.RunAsync(); ``` Scale throughput by adding more workers — each new consumer on the channel pulls its own share of the backlog. To run identical worker instances as separate processes, register the project multiple times in the AppHost. The repo's `Patterns.WorkQueue`, `Patterns.FanOut`, and `Patterns.RequestReply` examples show the full work-queue, fan-out, and request-reply recipes against the provisioned broker. ## Manual Settlement [#manual-settlement] `AutoAck` is convenient but unforgiving: if your handler throws after the broker has already acknowledged, the message is gone. For at-least-once processing you want **manual settlement** — acknowledge only after the work succeeds, and reject (requeue) on failure. Manual settlement runs through a **downstream receiver**. Create one with `CreateQueueDownstreamReceiverAsync`, then call `PollAsync` with `AutoAck = false`. The receiver is disposable — `await using` ensures it is released when the scope ends. ```csharp title="Queues.AckReject/Program.cs" // Receive with manual settlement via downstream receiver await using var receiver = await client.CreateQueueDownstreamReceiverAsync(); var batch = await receiver.PollAsync(new QueuePollRequest { Channel = "queues.ackreject", MaxMessages = 2, WaitTimeoutSeconds = 10, AutoAck = false }); ``` ### Ack vs. Nack [#ack-vs-nack] The returned batch exposes `HasMessages` (whether anything was returned) and `Messages` (the items). Each message carries its own settlement methods: * `await msg.AckAsync()` — confirm the message; the broker removes it from the queue. * `await msg.NackAsync()` — reject the message; the broker requeues it for redelivery to a consumer. The example acknowledges messages it can handle and rejects the rest so they are redelivered: ```csharp title="Queues.AckReject/Program.cs" if (batch.HasMessages) { foreach (var msg in batch.Messages) { var body = Encoding.UTF8.GetString(msg.Body.Span); if (body.Contains("ACK")) { await msg.AckAsync(); logger.LogInformation("ACK: {Body}", body); } else { await msg.NackAsync(); logger.LogInformation("NACK (rejected): {Body}", body); } } } ``` A rejected message is requeued and will be redelivered. Without a redelivery limit, a message that always fails will loop forever — pair `NackAsync` with a [dead-letter queue](#advanced-queue-features) so poison messages are diverted after a set number of attempts. ## Advanced Queue Features [#advanced-queue-features] The Aspire examples suite ships a runnable project for each advanced queue capability. They all follow the same shape — `AddKubeMQClient("messaging")`, `ConnectAsync`, then the SDK call — and target the standard (poll-based) Queues API: | Example | Capability | | ------------------------- | --------------------------------------------------------------------------------------------------------- | | `Queues.Peek` | Read messages without consuming them (`is_peek`) — inspect the head of the queue while leaving it intact. | | `Queues.DeadLetterQueue` | Divert messages to a dead-letter channel after they exceed a redelivery limit. | | `Queues.DelayedMessages` | Enqueue a message that becomes visible to consumers only after a delay. | | `Queues.ExpirationPolicy` | Drop messages that are not consumed within a time-to-live window. | | `Queues.Batch` | Send and receive messages in batches for higher throughput. | | `Queues.PollMode` | Tune long-poll behavior — `MaxMessages` and `WaitTimeoutSeconds`. | | `Queues.SendReceive` | The minimal send-then-receive round trip. | | `Queues.AckReject` | Manual ack / nack settlement (shown above). | These projects are registered in the AppHost's `Queues` region, so you can launch them through Aspire alongside the broker. ### Streaming Queue Receivers [#streaming-queue-receivers] For advanced settlement flows the suite also includes a separate **QueuesStream** category built on a streaming receiver rather than discrete polls. It exposes bulk-settlement and redelivery operations: | Example | Capability | | ------------------------------- | ------------------------------------------------------- | | `QueuesStream.StreamReceive` | Continuously receive over a streaming channel. | | `QueuesStream.AckAll` | Acknowledge every message in the current batch at once. | | `QueuesStream.NackAll` | Reject every message in the batch at once. | | `QueuesStream.RequeueAll` | Requeue the whole batch for redelivery. | | `QueuesStream.DeadLetterPolicy` | Apply a dead-letter policy on the streaming receiver. | These live in the AppHost's `QueuesStream` region. Reach for them when you need long-lived, high-throughput consumers or batch-wide settlement; for most request/response and worker workloads the poll-based API above is simpler. ## Body Handling [#body-handling] Message bodies are **byte arrays** (`byte[]`), so KubeMQ stays payload-agnostic. Encode and decode at the edges of your code: * On send: `Encoding.UTF8.GetBytes(body)` * On receive: `Encoding.UTF8.GetString(m.Body.Span)` — the received `Body` is a `ReadOnlyMemory`, so use its `.Span`. For structured payloads, serialize to JSON (or your format of choice) before encoding to UTF-8, and deserialize after decoding. ```csharp title="encode-decode.cs" // Send var message = new QueueMessage { Channel = "queues.example", Body = Encoding.UTF8.GetBytes(body), }; // Receive var text = Encoding.UTF8.GetString(m.Body.Span); ``` ## Related [#related]