Transports

Dispatch supports multiple message transports for distributed messaging. You can use a single transport or route messages to different transports based on rules.

Before You Start

.NET 10.0

Install the core package plus your transport:

dotnet add package Excalibur.Dispatch
dotnet add package Excalibur.Dispatch.Transport.Kafka  # or RabbitMQ, AzureServiceBus, etc.

Familiarity with handlers and pipeline concepts

Not sure which transport to use?

Start with the Choosing a Transport guide for a decision matrix and trade-off comparison across all five production transports.

Supported Transports

Transport	Package	Use Case
In-Memory	Built-in	Testing, development
Cron Timer	Built-in	Scheduled jobs, background tasks
Kafka	`Excalibur.Dispatch.Transport.Kafka`	High-throughput event streaming
RabbitMQ	`Excalibur.Dispatch.Transport.RabbitMQ`	Traditional messaging, routing patterns
Azure Service Bus	`Excalibur.Dispatch.Transport.AzureServiceBus`	Azure-native messaging
AWS SQS	`Excalibur.Dispatch.Transport.AwsSqs`	AWS-native messaging
Google Pub/Sub	`Excalibur.Dispatch.Transport.GooglePubSub`	GCP-native messaging
gRPC	`Excalibur.Dispatch.Transport.Grpc`	Service-to-service messaging over gRPC — unary send/receive, server-streaming subscribe, TLS required by default
Apache Pulsar	`Excalibur.Dispatch.Transport.Pulsar`	Pulsar messaging — transport primitives only (keyed sender/receiver; not yet pipeline-integrated)
MQTT	`Excalibur.Dispatch.Transport.Mqtt`	IoT / device messaging — transport primitives only (keyed sender/receiver; QoS + shared subscriptions)
IBM MQ	`Excalibur.Dispatch.Transport.IbmMq`	Enterprise queue messaging — transport primitives only (keyed sender/receiver; unit-of-work per message)

Quick Start

Single Transport

Configure transports through the AddDispatch() builder using Use{Transport}() methods:

// Install the transport package
// dotnet add package Excalibur.Dispatch.Transport.RabbitMQ

services.AddDispatch(dispatch =>
{
    dispatch.AddHandlersFromAssembly(typeof(Program).Assembly);

    // Configure transport through the builder (recommended)
    dispatch.UseRabbitMQ(rmq => rmq.HostName("localhost"));
});

All five transports follow the same Use prefix pattern:

services.AddDispatch(dispatch =>
{
    dispatch.UseKafka(kafka => kafka.BootstrapServers("localhost:9092"));
    dispatch.UseRabbitMQ(rmq => rmq.HostName("localhost"));
    dispatch.UseAzureServiceBus(asb => asb.ConnectionString("..."));
    dispatch.UseAwsSqs(sqs => sqs.Region("us-east-1"));
    dispatch.UseGooglePubSub(pubsub => pubsub.ProjectId("my-project"));
});

Multi-Transport Routing

Route different message types to different transports:

services.AddDispatch(dispatch =>
{
    dispatch.AddHandlersFromAssembly(typeof(Program).Assembly);

    // Register transports through the builder
    dispatch.UseKafka(kafka => kafka.BootstrapServers("localhost:9092"));
    dispatch.UseRabbitMQ(rmq => rmq.HostName("localhost"));

    dispatch.UseRouting(routing =>
    {
        routing.Transport
            // High-volume events to Kafka
            .Route<OrderCreatedEvent>().To("kafka")
            // Payment events to RabbitMQ
            .Route<PaymentProcessedEvent>().To("rabbitmq")
            // Default transport
            .Default("rabbitmq");
    });
});

Transport Interfaces

Dispatch provides minimal transport interfaces inspired by Microsoft.Extensions.AI (IChatClient), Azure.Messaging.ServiceBus (ServiceBusClient), and HttpClientFactory (DelegatingHandler).

ITransportSender / ITransportReceiver

Each transport implements two minimal interfaces — 3 methods each plus GetService() for raw SDK access:

public interface ITransportSender : IAsyncDisposable
{
    string Destination { get; }
    Task<SendResult> SendAsync(TransportMessage message, CancellationToken cancellationToken);
    Task<BatchSendResult> SendBatchAsync(IReadOnlyList<TransportMessage> messages, CancellationToken cancellationToken);
    Task FlushAsync(CancellationToken cancellationToken);
    object? GetService(Type serviceType) => null;
}

public interface ITransportReceiver : IAsyncDisposable
{
    string Source { get; }
    Task<IReadOnlyList<TransportReceivedMessage>> ReceiveAsync(int maxMessages, CancellationToken cancellationToken);
    Task AcknowledgeAsync(TransportReceivedMessage message, CancellationToken cancellationToken);
    Task RejectAsync(TransportReceivedMessage message, string? reason, bool requeue, CancellationToken cancellationToken);
    object? GetService(Type serviceType) => null;
}

Settlement: acknowledge and reject

AcknowledgeAsync and RejectAsync are how you tell the broker what became of a message. Returning normally means the broker accepted the settlement; it is not a best-effort call whose outcome you have to infer from a later redelivery.

When a settlement does not happen, the transport raises TransportSettlementException. Read RedeliveryExpectation on it to decide what to do — Expected means the broker still owns the message and your handler's work may run again, so handlers must be idempotent. Treat Unspecified the same way: it means the transport could not determine the outcome, and assuming the work returns is the safe direction.

requeue states the outcome you require rather than a hint: true asks the broker to deliver the message again, false asks it not to.

This contract is newly stated and not every transport meets it yet

IBM MQ reports settlement failures as TransportSettlementException today. Several other receivers still let their broker client's own exception escape, and a settlement that times out surfaces as an OperationCanceledException. Until each transport is converted, portable code that must not fault should catch InvalidOperationException — the base of TransportSettlementException — and treat an unrecognised exception from a settlement call as an outcome it cannot confirm: assume the message may be redelivered.

MQTT and IBM MQ do not yet honour requeue: false; both currently cause redelivery. If you dead-letter a poison message on those transports, make the dead-letter write idempotent — see dead-letter queues.

Batch results and selective retry

SendBatchAsync returns one result per input, in the order the inputs were given: Results[i] is the outcome of messages[i], and Results.Count always equals the number of messages you passed. That holds however the provider actually sent them — including when it splits the batch across several calls to the broker, or cannot fit every message into one native batch and sends the remainder individually. Results are never grouped into successes followed by failures.

This is what makes a partial failure recoverable. Resend only the inputs whose result failed:

var result = await sender.SendBatchAsync(messages, cancellationToken);

if (!result.IsCompleteSuccess)
{
    var retry = new List<TransportMessage>();
    for (var i = 0; i < messages.Count; i++)
    {
        if (!result.Results[i].IsSuccess && result.Results[i].Error?.IsRetryable == true)
        {
            retry.Add(messages[i]);
        }
    }

    await sender.SendBatchAsync(retry, cancellationToken);
}

Without index alignment the only safe recovery would be resending the whole batch, which on an at-least-once transport duplicates every message that was already delivered.

note

SendResult.MessageId is not yet a portable way to confirm which input a result belongs to — providers differ over whether they report the broker's identifier or the message's own Id. Use the index, as above, rather than matching on MessageId.

ITransportSubscriber (Push-Based)

For transports with native push semantics (Kafka consumer groups, RabbitMQ BasicConsume, Azure Event Hubs, Google Pub/Sub streaming pull), ITransportSubscriber provides a push-based alternative to the pull-based ITransportReceiver:

public interface ITransportSubscriber : IAsyncDisposable
{
    string Source { get; }
    Task SubscribeAsync(
        Func<TransportReceivedMessage, CancellationToken, Task<MessageAction>> handler,
        CancellationToken cancellationToken);
    object? GetService(Type serviceType) => null;
}

public enum MessageAction { Acknowledge, Reject, Requeue }

The handler callback returns a MessageAction telling the transport what to do with the message. DelegatingTransportSubscriber provides the decorator base class.

Interface	Pattern	Use When
`ITransportReceiver`	Pull	You control the polling loop, batch receive
`ITransportSubscriber`	Push	Transport drives delivery, handler reacts

All 5 transports implement ITransportSubscriber:

Transport	Push Model	SDK API
Azure Service Bus	Native push	`ServiceBusProcessor` events
RabbitMQ	Native push	`AsyncEventingBasicConsumer` + `BasicConsumeAsync`
Google Pub/Sub	Native push	`SubscriberClient.StartAsync()` streaming pull
Kafka	Continuous poll	`IConsumer<string,byte[]>.Consume()` in loop
AWS SQS	Long poll	`ReceiveMessageAsync` with 20s wait

MessageAction settlement maps to transport-native operations:

MessageAction	Azure SB	RabbitMQ	Kafka	AWS SQS	Google Pub/Sub
`Acknowledge`	Complete	BasicAck	Commit offset	Delete	Ack
`Reject`	Dead-letter	Nack (no requeue)	Commit (DLQ via decorator)	Delete (DLQ via redrive)	Nack
`Requeue`	Abandon	Nack (requeue)	Seek back	Visibility timeout = 0	Nack

GetService exposes the underlying subscriber client:

Transport	Subscriber Returns
Azure Service Bus	`ServiceBusProcessor`
RabbitMQ	`IChannel`
Kafka	`IConsumer<string, byte[]>`
AWS SQS	`IAmazonSQS`
Google Pub/Sub	`SubscriberClient`

TransportMessage (Slim)

TransportMessage is a slim message type (9 properties). Transport-specific hints flow via the Properties dictionary with well-known keys:

var message = new TransportMessage
{
    Body = Encoding.UTF8.GetBytes(payload),
    ContentType = "application/json",
    MessageType = "OrderCreated",
    CorrelationId = correlationId,
};

// Transport hints via Properties dictionary
message.Properties[TransportTelemetryConstants.PropertyKeys.OrderingKey] = orderId;
message.Properties[TransportTelemetryConstants.PropertyKeys.PartitionKey] = customerId;

Decorator Pattern

Cross-cutting concerns — telemetry, ordering, deduplication, scheduling, dead-letter routing — are added by composing a sender or receiver through a builder. You compose them by calling the Use… methods below; the decorator types themselves are an implementation detail and are not part of the public surface.

var sender = new TransportSenderBuilder(nativeSender)
    .UseTelemetry("Kafka", meter, activitySource)
    .UseOrdering(msg => msg.Subject)
    .Build();

var receiver = new TransportReceiverBuilder(nativeReceiver)
    .UseTelemetry("Kafka", meter, activitySource)
    .UseDeadLetterQueue("Kafka", (message, reason, ct) => RouteToDlqAsync(message, reason, ct))
    .Build();

Call	Direction	What it adds
`UseTelemetry(transportName, meter, activitySource)`	Send	OpenTelemetry metrics and traces around each send
`UseTelemetry(transportName, meter, activitySource)`	Receive	OpenTelemetry metrics and traces around each receive
`UseOrdering(keySelector)`	Send	Sets the ordering key on each message from the selector
`UseDeduplication(idSelector)`	Send	Sets the deduplication id on each message from the selector
`UseScheduling(timeSelector)`	Send	Sets a scheduled delivery time on each message from the selector
`UseDeadLetterQueue(transportName, handler, meter?)`	Receive	Routes a failed message to your dead-letter handler

Each returns the builder, so they compose in any order; Build() returns an ITransportSender or ITransportReceiver you can register or pass on. If you need a decorator of your own, derive from DelegatingTransportSender or DelegatingTransportReceiver — both are public — and add it with Use(...).

CloudEvents

The two directions are not symmetric.

Inbound decoding is applied for you. Each transport's own registration wraps its receiver with CloudEvent decoding for that transport's binding, so there is no call to make and no registration that can be missing. The decoded event is attached to the received message's provider data; the message body is not replaced, so code that reads the raw body is unaffected and acknowledgement still addresses the original message.

Outbound formatting is not a sender decorator. Envelope↔CloudEvent mapping happens in the CloudEvents bridge you enable with AddCloudEvents(...), through a single canonical emit path, with the per-transport binding (Kafka's structured ce_ binding, for example) supplied by that transport's CloudEvent adapter.

GetService() — Raw SDK Access

Access the underlying transport SDK client for advanced scenarios:

// Kafka: get the native IProducer
var producer = sender.GetService(typeof(IProducer<string, byte[]>))
    as IProducer<string, byte[]>;

// Azure Service Bus: get the native ServiceBusSender
var sbSender = sender.GetService(typeof(ServiceBusSender))
    as ServiceBusSender;

Transport	Sender Returns	Receiver Returns	Subscriber Returns
Kafka	`IProducer<string, byte[]>`	`IConsumer<string, byte[]>`	`IConsumer<string, byte[]>`
RabbitMQ	`IChannel`	`IChannel`	`IChannel`
Azure Service Bus	`ServiceBusSender`	`ServiceBusReceiver`	`ServiceBusProcessor`
AWS SQS	`IAmazonSQS`	`IAmazonSQS`	`IAmazonSQS`
Google Pub/Sub	`PublisherServiceApiClient`	`SubscriberServiceApiClient`	`SubscriberClient`

Transport Selection Guide

Requirement	Recommended Transport
High throughput (>100k msg/sec)	Kafka
Complex routing patterns	RabbitMQ
Azure-native integration	Azure Service Bus
AWS-native integration	AWS SQS
GCP-native integration	Google Pub/Sub
Scheduled jobs / cron tasks	Cron Timer
Local development/testing	In-Memory

Common Configuration

Connection Resilience

Configure resilience per transport via the options classes:

services.AddPluggableSerialization(); // Transports don't seat a default serializer
services.AddRabbitMQTransport(rmq =>
{
    rmq.HostName("rabbitmq")
       .AutomaticRecovery(enabled: true, networkRecoveryInterval: TimeSpan.FromSeconds(10));
});

Health Checks

services.AddHealthChecks()
    .AddTransportHealthChecks();

app.MapHealthChecks("/health");

Observability

All transports emit OpenTelemetry traces and metrics once telemetry is composed onto the sender and receiver (UseTelemetry(...), above):

services.AddOpenTelemetry()
    .WithTracing(builder =>
    {
        builder.AddSource("Excalibur.Dispatch");
        // Transport-specific traces: Excalibur.Dispatch.Transport.{Name}
        builder.AddSource("Excalibur.Dispatch.Transport.Kafka");
        builder.AddSource("Excalibur.Dispatch.Transport.RabbitMQ");
    })
    .WithMetrics(builder =>
    {
        builder.AddDispatchMetrics();
    });

Standard transport metric names follow dispatch.transport.* convention:

Metric	Type	Description
`dispatch.transport.messages.sent`	Counter	Messages sent successfully
`dispatch.transport.messages.send_failed`	Counter	Send failures
`dispatch.transport.messages.received`	Counter	Messages received
`dispatch.transport.messages.acknowledged`	Counter	Messages acknowledged
`dispatch.transport.messages.rejected`	Counter	Messages rejected
`dispatch.transport.send.duration`	Histogram	Send operation duration (ms)
`dispatch.transport.receive.duration`	Histogram	Receive operation duration (ms)
`dispatch.transport.batch.size`	Histogram	Batch operation message count

Message Serialization

By default, messages are serialized using JSON (System.Text.Json). You can configure different serializers:

// Use MemoryPack for .NET-only maximum performance
services.AddMemoryPackSerializer();

// Or MessagePack for compact binary format
services.AddMessagePackSerializer();

Serializer	Package	Best For
System.Text.Json (default)	Built-in	General purpose, cross-language, debugging
MemoryPack	`Excalibur.Dispatch.Serialization.MemoryPack`	.NET-only, maximum performance
MessagePack	`Excalibur.Dispatch.Serialization.MessagePack`	Cross-language, compact
Protobuf	`Excalibur.Dispatch.Serialization.Protobuf`	Schema-based, cross-language

Dead Letter Queue Support

Each transport can implement IDeadLetterQueueManager from Excalibur.Dispatch.Transport.Abstractions for transport-native dead letter handling:

Transport	DLQ Support	Mechanism	Registration
Google Pub/Sub	Yes	Subscription-based	Built-in
AWS SQS	Yes	Queue-based (native redrive)	Built-in
Kafka	Yes	Topic-based (`{topic}.dead-letter`)	`services.AddKafkaDeadLetterQueue()`
Azure Service Bus	Yes	Native `$DeadLetterQueue` subqueue	`services.AddServiceBusDeadLetterQueue()`
RabbitMQ	Yes	Dead letter exchange (DLX)	`services.AddRabbitMqDeadLetterQueue()`

IDeadLetterQueueManager Interface

All transport DLQ implementations share the same base interface:

public interface IDeadLetterQueueManager
{
    Task<string> MoveToDeadLetterAsync(
        TransportMessage message, string reason,
        Exception? exception,
        CancellationToken cancellationToken);

    Task<IReadOnlyList<DeadLetterMessage>> GetDeadLetterMessagesAsync(
        int maxMessages,
        CancellationToken cancellationToken);

    Task<ReprocessResult> ReprocessDeadLetterMessagesAsync(
        IEnumerable<DeadLetterMessage> messages,
        ReprocessOptions options,
        CancellationToken cancellationToken);

    Task<DeadLetterStatistics> GetStatisticsAsync(
        CancellationToken cancellationToken);

    Task<int> PurgeAllTenantsDeadLetterQueueAsync(
        CancellationToken cancellationToken);
}

Kafka DLQ Example

services.AddPluggableSerialization(); // Transports don't seat a default serializer
services.AddKafkaTransport("events", kafka => { /* ... */ });

services.AddKafkaDeadLetterQueue(dlq =>
{
    dlq.TopicSuffix = ".dead-letter";       // Default
    dlq.ConsumerGroupId = "dlq-processor";  // Default
    dlq.MaxDeliveryAttempts = 5;            // Default
    dlq.MessageRetentionPeriod = TimeSpan.FromDays(14);
    dlq.AutoCreateTopics = true;
});

AWS SQS DLQ

AWS SQS uses its own native redrive policy: attach a dead-letter queue to the source queue and SQS moves a message there automatically once it exceeds the maximum receive count. Configure it on the queue builder:

queue.DeadLetterQueue(dlq => dlq
    .QueueArn("arn:aws:sqs:us-east-1:123456789012:orders-dlq")
    .MaxReceiveCount(5));

Azure Service Bus DLQ

Azure Service Bus uses the native $DeadLetterQueue subqueue. The manager accesses it via ServiceBusReceiverOptions { SubQueue = SubQueue.DeadLetter } — no admin API package required.

services.AddServiceBusDeadLetterQueue(dlq =>
{
    dlq.EntityPath = "orders";           // Queue or subscription path
    dlq.MaxBatchSize = 100;              // Batch size for purge/retrieve (default: 100)
    dlq.ReceiveWaitTime = TimeSpan.FromSeconds(5);  // Wait time (default: 5s)
    dlq.StatisticsPeekCount = 1000;      // Max messages to peek for stats (default: 1000)
    dlq.IncludeStackTrace = true;        // Include stack traces (default: true)
});

RabbitMQ DLQ

RabbitMQ uses dead letter exchanges (DLX). Messages are published to the DLX exchange with dlq_reason and dlq_original_source headers. Peek semantics use BasicGet(autoAck: false) + Nack(requeue: true).

services.AddRabbitMqDeadLetterQueue(dlq =>
{
    dlq.Exchange = "dead-letters";         // DLX exchange name (default: "dead-letters")
    dlq.QueueName = "dead-letter-queue";   // DLQ queue name (default: "dead-letter-queue")
    dlq.RoutingKey = "#";                  // Routing key (default: "#")
    dlq.IncludeStackTrace = true;          // Include stack traces (default: true)
    dlq.MaxBatchSize = 100;               // Batch size for stats (default: 100)
});

For more details, see Dead Letter Handling.

Poison Message Handling

services.AddDispatch(dispatch =>
{
    dispatch.UsePoisonMessageHandling(options =>
    {
        options.MaxRetryAttempts = 5;
        options.EnableAlerting = true;
        options.AlertThreshold = 10;
    });
});

In This Section

In-Memory — Built-in transport for testing
Cron Timer — Scheduled jobs and recurring tasks
Kafka — Apache Kafka transport
RabbitMQ — RabbitMQ transport
Azure Service Bus — Azure Service Bus transport
AWS SQS — AWS SQS transport
Google Pub/Sub — Google Pub/Sub transport
Multi-Transport Routing — Route to multiple transports

Before You Start​

Supported Transports​

Quick Start​

Single Transport​

Multi-Transport Routing​

Transport Interfaces​

ITransportSender / ITransportReceiver​

Settlement: acknowledge and reject​

Batch results and selective retry​

ITransportSubscriber (Push-Based)​

TransportMessage (Slim)​

Decorator Pattern​

GetService() — Raw SDK Access​

Transport Selection Guide​

Common Configuration​

Connection Resilience​

Health Checks​

Observability​

Message Serialization​

Dead Letter Queue Support​

IDeadLetterQueueManager Interface​

Kafka DLQ Example​

AWS SQS DLQ​

Azure Service Bus DLQ​

RabbitMQ DLQ​

Poison Message Handling​

In This Section​

See Also​