# Queues (Send) (/integrations/masstransit/how-to/queues)



## Overview [#overview]

MassTransit `Send` maps to **KubeMQ Queues** — durable, point-to-point messaging. Where `Publish` fans a message out to every active subscriber, `Send` directs a message to a single endpoint where **exactly one consumer** receives it. Messages are **persisted in the queue until consumed**, so the consumer need not be running at the moment you send: the broker holds the message and delivers it when a consumer is available.

→ For what a Queue is and KubeMQ's queue semantics, see [Queues](/learn/queues). This page documents the MassTransit transport's API surface only.

Delivery is acknowledgment-based. After a consumer's `Consume` method completes successfully, the transport **ack**s the message and KubeMQ removes it from the queue. If the consumer throws, the transport **nack**s the message, and faulted messages route to the `{queue}_error` channel (skipped messages route to `{queue}_skipped`).

<Mermaid
  chart="`
graph LR
P[&#x22;Sender<br/>endpoint.Send(msg)&#x22;]
Q[&#x22;KubeMQ Queue<br/>order-processing&#x22;]
C[&#x22;IConsumer&lt;SubmitOrder&gt;<br/>(exactly one)&#x22;]
E[&#x22;order-processing_error&#x22;]

P -- &#x22;gRPC :50000&#x22; --> Q
Q -- &#x22;poll + deliver&#x22; --> C
C -- &#x22;ack on success&#x22; --> Q
C -. &#x22;nack on failure&#x22; .-> E

class P,C external
class Q,E broker
`"
/>

*A sent message lands in a durable KubeMQ Queue and is delivered to exactly one consumer; faults route to the `_error` channel.*

<Callout type="info">
  `Send` is the right choice when each message must be processed once and must survive a consumer being offline. For fan-out delivery to all subscribers, use [Events](/integrations/masstransit/how-to/events) instead.
</Callout>

## API surface [#api-surface]

| Member                                                                       | Where             | Purpose                                                                        |
| ---------------------------------------------------------------------------- | ----------------- | ------------------------------------------------------------------------------ |
| `bus.GetSendEndpoint(Uri)` / `IKubeMQRider.GetSendEndpoint(Uri, ct)`         | bus / rider       | Resolve a send endpoint for a `queue:` or `kubemq://host:port/channel` address |
| `ISendEndpoint.Send<T>(message, ctx => …)`                                   | send endpoint     | Send a message; the context pipe sets `Delay`, `TimeToLive`, and headers       |
| `SendContext.Delay`                                                          | send context      | Delayed delivery → `QueueMessage.DelaySeconds`                                 |
| `SendContext.TimeToLive`                                                     | send context      | Per-message TTL → `QueueMessage.ExpirationSeconds`                             |
| `IKubeMQEndpointTransportConfigurator.ExpirationSeconds`                     | `ConfigureKubeMQ` | Per-endpoint TTL for every message the endpoint produces                       |
| `IKubeMQBusFactoryConfigurator.UsePriorityQueues(…)`                         | bus factory       | Enable weighted `_high`/`_normal`/`_low` priority channels                     |
| `IKubeMQReceiveEndpointConfigurator.PollTimeoutSeconds` / `.MaxPollMessages` | receive endpoint  | Tune long-poll wait and batch size per endpoint                                |

Send endpoints use the `kubemq://host:port/channel-name` URI scheme. The short `queue:channel-name` form resolves to the same queue channel on the configured host. Full naming and address rules live in the [reference](/integrations/masstransit/reference/configuration).

## Usage [#usage]

### Basic send and receive [#basic-send-and-receive]

To send to a queue, obtain a send endpoint for the `queue:` address and call `Send`. The matching consumer implements `IConsumer<T>`. Unlike Events, a queue endpoint does not call `UseVolatileEvents()` — the default endpoint behavior is a KubeMQ Queue consumer.

```csharp title="Program.cs"
using MassTransit;
using MassTransit.KubeMQTransport;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();
    x.UsingKubeMQ((context, cfg) =>
    {
        cfg.Host("localhost", 50000);
        cfg.ReceiveEndpoint("send-basic-queue", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    });
});

builder.Services.AddHostedService<BasicSendSender>();

var host = builder.Build();
await host.RunAsync();

// --- Message type ---
public record OrderMessage(Guid OrderId, string ProductName, decimal Amount);

// --- Consumer ---
public class OrderConsumer(ILogger<OrderConsumer> logger) : IConsumer<OrderMessage>
{
    public Task Consume(ConsumeContext<OrderMessage> context)
    {
        logger.LogInformation(
            "Received order: OrderId={OrderId}, Product={Product}, Amount={Amount:C}",
            context.Message.OrderId,
            context.Message.ProductName,
            context.Message.Amount);
        return Task.CompletedTask;
    }
}

// --- Background sender ---
public class BasicSendSender(
    ILogger<BasicSendSender> logger,
    IHostApplicationLifetime lifetime) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await Task.Delay(2000, stoppingToken);

        var rider = KubeMQRiderAccessor.Current
            ?? throw new InvalidOperationException("KubeMQ rider not started.");

        var endpoint = await rider.GetSendEndpoint(
            new Uri("kubemq://localhost:50000/send-basic-queue"), stoppingToken);

        var order = new OrderMessage(Guid.NewGuid(), "Widget", 29.99m);
        await endpoint.Send(order, stoppingToken);
        logger.LogInformation("Message sent successfully.");

        await Task.Delay(5000, stoppingToken);
        lifetime.StopApplication();
    }
}
```

<Callout type="info">
  The short `queue:channel-name` form (as in `bus.GetSendEndpoint(new Uri("queue:order-processing"))`) resolves to the same queue channel as the fully-qualified `kubemq://host:port/channel-name`. Inside a hosted service, `KubeMQRiderAccessor.Current` gives direct access to the rider's `GetSendEndpoint`.
</Callout>

### Delayed delivery [#delayed-delivery]

Set `ctx.Delay` on the send context to schedule a message for later. The transport maps `Delay` to `QueueMessage.DelaySeconds`: the message is enqueued immediately but stays invisible to consumers until the delay elapses.

```csharp title="Delayed send (maps to QueueMessage.DelaySeconds)"
var endpoint = await rider.GetSendEndpoint(
    new Uri("kubemq://localhost:50000/send-delayed-queue"), stoppingToken);

var reminder = new ReminderMessage(
    Guid.NewGuid(), "Follow up with client", DateTimeOffset.UtcNow.AddSeconds(5));

// Set the delay on the send context — transport maps this to QueueMessage.DelaySeconds
await endpoint.Send(reminder, context =>
{
    context.Delay = TimeSpan.FromSeconds(5);
}, stoppingToken);
```

<Callout type="warn">
  Delayed delivery is a **queue-only** feature. It is not supported for `Publish` (Events or EventsStore) — attempting to delay a published event throws `KubeMQTransportConfigurationException`. Use `Send` whenever you need scheduled delivery.
</Callout>

### Message expiration (TTL) [#message-expiration-ttl]

A queue message can carry a time-to-live. If it is not consumed within the TTL, KubeMQ discards it rather than delivering a stale message. Set it per message with `ctx.TimeToLive`, or per endpoint with `ExpirationSeconds` in `ConfigureKubeMQ`. Both map to `QueueMessage.ExpirationSeconds`.

<Tabs items="[&#x22;Per-message TTL&#x22;, &#x22;Per-endpoint TTL&#x22;]">
  <Tab value="Per-message TTL">
    ```csharp title="Per-message TTL"
    var endpoint = await rider.GetSendEndpoint(
        new Uri("kubemq://localhost:50000/send-expiration-queue"), stoppingToken);

    var coupon = new CouponMessage(Guid.NewGuid(), "SAVE20", 20.0m);

    await endpoint.Send(coupon, context =>
    {
        context.TimeToLive = TimeSpan.FromSeconds(30);
    }, stoppingToken);
    ```
  </Tab>

  <Tab value="Per-endpoint TTL">
    ```csharp title="Per-endpoint expiration"
    cfg.ReceiveEndpoint("expiration-demo-queue", e =>
    {
        // Messages not consumed within 1 hour are discarded by KubeMQ
        e.ConfigureKubeMQ(k =>
        {
            k.ExpirationSeconds = 3600;
        });
    });
    ```
  </Tab>
</Tabs>

An expired message is dropped, not moved to `{queue}_error` — TTL is distinct from dead-letter routing.

### Custom headers [#custom-headers]

Custom headers set on the send context travel with the message and are readable on the consumer. The transport maps each user header to a KubeMQ tag with the `MT-Header-{name}` prefix (so `tenant-id` becomes `MT-Header-tenant-id`); MassTransit envelope headers map to `MT-*` tags. The full mapping table lives in the [reference](/integrations/masstransit/reference/configuration).

```csharp title="Setting and reading custom headers"
// Send
await endpoint.Send(invoice, context =>
{
    context.Headers.Set("tenant-id", "acme-corp");
    context.Headers.Set("x-correlation-hint", "batch-2024-q1");
}, stoppingToken);

// Receive
public class HeaderConsumer(ILogger<HeaderConsumer> logger) : IConsumer<InvoiceMessage>
{
    public Task Consume(ConsumeContext<InvoiceMessage> context)
    {
        var tenantId = context.Headers.Get<string>("tenant-id") ?? "(not set)";
        logger.LogInformation("Received invoice; tenant-id={TenantId}", tenantId);
        return Task.CompletedTask;
    }
}
```

You set and read plain MassTransit header names; the transport handles the `MT-Header-` tag prefix on the wire.

### Batch consuming [#batch-consuming]

A consumer can receive multiple messages per `Consume` call by implementing `IConsumer<Batch<T>>`. Configure batching on the consumer with `BatchOptions` — `SetMessageLimit` caps how many messages accumulate before delivery, `SetTimeLimit` flushes a partial batch after a timeout. The number the transport pulls per poll is governed by the endpoint's `MaxPollMessages` (default `32`).

```csharp title="Batch consumer"
public record BatchOrderMessage(Guid OrderId, string ProductName, decimal Amount, int Sequence);

public class OrderBatchConsumer(ILogger<OrderBatchConsumer> logger)
    : IConsumer<Batch<BatchOrderMessage>>
{
    public Task Consume(ConsumeContext<Batch<BatchOrderMessage>> context)
    {
        logger.LogInformation("Received batch of {Count} messages", context.Message.Length);

        foreach (var msg in context.Message)
        {
            logger.LogInformation(
                "  Batch item: Seq={Sequence}, OrderId={OrderId}",
                msg.Message.Sequence, msg.Message.OrderId);
        }

        return Task.CompletedTask;
    }
}

// Configures batch size and time limits
public class OrderBatchConsumerDefinition : ConsumerDefinition<OrderBatchConsumer>
{
    protected override void ConfigureConsumer(
        IReceiveEndpointConfigurator endpointConfigurator,
        IConsumerConfigurator<OrderBatchConsumer> consumerConfigurator,
        IRegistrationContext context)
    {
        consumerConfigurator.Options<BatchOptions>(options => options
            .SetMessageLimit(10)
            .SetTimeLimit(TimeSpan.FromSeconds(5)));
    }
}
```

<Callout type="info">
  `MaxPollMessages` controls how many messages the transport fetches from KubeMQ per poll; `BatchOptions.SetMessageLimit` controls how many MassTransit delivers to one `Consume` call. Tune `MaxPollMessages` to be at least as large as your batch message limit so a full batch can be assembled from a single poll.
</Callout>

### Competing consumers [#competing-consumers]

Multiple receive endpoints that share the same channel name form a **consumer group**. KubeMQ load-balances queue messages across the group, so each message is delivered to exactly one consumer — the competing-consumers pattern, and how you scale queue throughput horizontally.

```csharp title="Competing consumers in one process"
x.UsingKubeMQ((context, cfg) =>
{
    cfg.Host("localhost", 50000);

    // Two endpoints on the same channel compete for messages.
    // KubeMQ delivers each message to exactly one member.
    cfg.ReceiveEndpoint("competing-consumers-queue", e =>
    {
        e.ConfigureKubeMQ(t => { });
    });

    cfg.ReceiveEndpoint("competing-consumers-queue", e =>
    {
        e.ConfigureKubeMQ(t => { });
    });
});
```

The same principle scales across processes: deploy multiple instances that each declare a receive endpoint with the same channel name, and KubeMQ distributes the queue's messages across them — at-least-once delivery with automatic load balancing, no partitioning or sharding configuration required.

### Priority queues [#priority-queues]

Priority queues let high-priority work jump ahead of lower-priority work. Enable them with `cfg.UsePriorityQueues()`. The transport creates three channels per endpoint — `{queue}_high`, `{queue}_normal`, `{queue}_low` — and polls them with **weighted round-robin** (default weights `High=3`, `Normal=2`, `Low=1`). You send to a level by addressing the suffixed channel.

```csharp title="Priority queues"
x.UsingKubeMQ((context, cfg) =>
{
    cfg.Host("localhost", 50000);

    // Default weights: high=3, normal=2, low=1
    cfg.UsePriorityQueues();

    cfg.ReceiveEndpoint("pq-basic-orders", e =>
    {
        e.ConfigureKubeMQ(t => { });
    });
});

// Send to a specific priority level by addressing the suffixed channel
var highEndpoint = await rider.GetSendEndpoint(
    new Uri("kubemq://localhost:50000/pq-basic-orders_high"), stoppingToken);
await highEndpoint.Send(new PriorityOrder(Guid.NewGuid(), "HIGH", "HighItem", 100.00m), stoppingToken);
```

To change the polling ratios, pass a configurator and call `SetWeights(highWeight, normalWeight, lowWeight)` on `IPriorityQueueConfigurator`:

```csharp title="Custom priority weights (5:3:1)"
// high-priority channel is polled 5x per cycle vs 1x for low
cfg.UsePriorityQueues(p => p.SetWeights(highWeight: 5, normalWeight: 3, lowWeight: 1));
```

<Callout type="info">
  Priority is enforced by weighted polling, not strict ordering. A higher weight means a channel is sampled more frequently, so high-priority messages are processed first under load — but low-priority messages still make progress and are never starved.
</Callout>

### Tuning poll behavior [#tuning-poll-behavior]

Queue receive uses long-polling. Two settings control it, set globally on `KubeMQTransportOptions` or per endpoint.

| Setting              | Default | Range  | Effect                                                           |
| -------------------- | ------- | ------ | ---------------------------------------------------------------- |
| `PollTimeoutSeconds` | `5`     | 1–3600 | How long a single poll waits for messages before returning empty |
| `MaxPollMessages`    | `32`    | 1–1024 | Maximum messages fetched per poll batch                          |

```csharp title="Per-endpoint poll tuning"
cfg.ReceiveEndpoint("order-processing", e =>
{
    e.PollTimeoutSeconds = 10;   // wait up to 10s per poll
    e.MaxPollMessages = 64;      // fetch up to 64 messages per poll
});
```

A longer `PollTimeoutSeconds` reduces idle gRPC traffic on quiet queues; a larger `MaxPollMessages` increases throughput on busy queues and supports larger batch consumers, at the cost of more in-flight messages per poll.

## Errors and next steps [#errors-and-next-steps]

* Faulted/skipped message routing, native DLQ via `UseNativeDlq`, and the retry pipeline → [Error Handling & DLQ](/integrations/masstransit/how-to/error-handling-dlq).
* The full transport-options table and exception hierarchy → [reference](/integrations/masstransit/reference/configuration).

<Cards>
  <Card title="Events (Publish)" href="/integrations/masstransit/how-to/events" description="Fire-and-forget fan-out delivery to all active subscribers." />

  <Card title="Commands & Queries" href="/integrations/masstransit/how-to/commands-queries" description="Native request-reply over KubeMQ CQ — no temporary reply queues." />

  <Card title="Configuration" href="/integrations/masstransit/how-to/configuration" description="Host, TLS, auth, poll tuning, and endpoint options." />
</Cards>