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 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.

Program.cs

var builder = Host.CreateApplicationBuilder(args);
builder.AddServiceDefaults();
builder.AddKubeMQClient("messaging");

var host = builder.Build();
var client = host.Services.GetRequiredService<IKubeMQClient>();

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 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

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.

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<IActionResult> 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

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:

QueuesController.cs

[HttpGet]
public async Task<IActionResult> 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

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.

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<IKubeMQClient>();
var logger = host.Services.GetRequiredService<ILogger<Program>>();
var lifetime = host.Services.GetRequiredService<IHostApplicationLifetime>();

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

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.

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

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:

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 so poison messages are diverted after a set number of attempts.

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

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

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<byte>, so use its .Span.

For structured payloads, serialize to JSON (or your format of choice) before encoding to UTF-8, and deserialize after decoding.

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);

Queues