RabbitMQ Transport¶

Verified by tests

RabbitMQTransportTests, RabbitMQChannelPoolTests, RabbitMQConnectionRetryTests, RabbitMQBatchSubscribeTests, RabbitMQTransportBatchPathTests, RabbitMQTransportFailurePathTests, RabbitMQSubscriptionTests, RabbitMQHealthCheckTests, ServiceCollectionExtensionsTests, RabbitMQFifoIntegrationTests — library CI run #31657041675 (2026-08-13)

The RabbitMQ transport provides reliable, distributed messaging using RabbitMQ topic exchanges with automatic dead-letter queue handling, connection pooling, and full AOT compatibility. This enables pub/sub patterns for event-driven architectures with flexible routing and retry semantics.

Why RabbitMQ?¶

RabbitMQ is a battle-tested open-source message broker offering:

Feature	Description	Benefit
Topic Exchanges	Flexible routing patterns	Wildcard subscriptions (`product.*`)
At-Least-Once Delivery	Message acknowledgments	Reliability
Dead Letter Queues	Automatic failure handling	Observability & recovery
Lightweight	Runs on-premise or containers	Developer-friendly
Message TTL	Time-to-live support	Automatic cleanup
Prefetch Control	QoS flow control	Backpressure management

Whizbang Integration: - ✅ AOT-Compatible - Uses JsonContextRegistry for source-generated JSON serialization - ✅ Channel Pooling - Thread-safe operations via semaphore-based pooling - ✅ TestContainers Support - First-class integration testing with Docker - ✅ Dead-Letter Queues - Automatic DLX/DLQ creation and binding - ✅ Pause/Resume - Subscription lifecycle management - ✅ Correlation Tracing - MessageId, CorrelationId, CausationId propagation

Architecture¶

Topic Exchange Pattern¶

flowchart TD
    subgraph Broker["RabbitMQ Broker"]
        subgraph Exchange["Exchange: #quot;products#quot; (topic)"]
            Q1["Queue: #quot;inventory-products-queue#quot;<br/>Binding: #quot;product.*#quot;"]
            Q2["Queue: #quot;analytics-products-queue#quot;<br/>Binding: #quot;product.created#quot;"]
            Q3["Queue: #quot;notifications-queue#quot;<br/>Binding: #quot;#35;#quot; (all messages)"]
        end

        subgraph DLX["Dead Letter Exchange: #quot;products.dlx#quot; (fanout)"]
            DLQ["Dead Letter Queue: #quot;inventory-queue.dlq#quot;"]
        end
    end

    Svc1["Inventory Service"]
    Svc2["Analytics Service"]
    Svc3["Notification Service"]

    Q1 --> Svc1
    Q2 --> Svc2
    Q3 --> Svc3

    class Q1,Q2,Q3,DLQ layer-command
    class Svc1,Svc2,Svc3 layer-core

Channel Pool Architecture¶

RabbitMQ channels are not thread-safe, so Whizbang uses a channel pool for concurrent publishing:

flowchart TD
    subgraph Pool["RabbitMQChannelPool"]
        subgraph Avail["Available Channels (Semaphore)<br/>Max: 10 (configurable)"]
            CH1["CH 1"]
            CH2["CH 2"]
            CH3["CH 3"]
            CHmore["..."]
        end
    end

    Publisher["Publisher (TransportPublishStrategy)<br/><br/>using (var channel =<br/>await pool.RentAsync()) {<br/>// Publish message<br/>// Channel auto-returns on dispose<br/>}"]

    Publisher -->|"Rent"| Pool
    Pool -->|"Return"| Publisher

    class CH1,CH2,CH3,CHmore layer-command
    class Publisher layer-core

Subscriptions get dedicated channels (no pooling) for long-lived operations.

Message Flow¶

Publishing¶

flowchart TD
    Publisher["Publisher (Order Service)"]
    Transport["RabbitMQTransport<br/><br/>- Rent channel from pool<br/>- Declare exchange (idempotent)<br/>- Serialize MessageEnvelope<br/>- Set BasicProperties:<br/>• MessageId<br/>• CorrelationId<br/>• EnvelopeType (for deser)<br/>- BasicPublish(exchange, key)<br/>- Return channel to pool"]
    Exchange["RabbitMQ Exchange: #quot;orders#quot;<br/>Type: topic"]
    Queue["Queue: #quot;fulfillment-orders-queue#quot;<br/>Binding: #quot;order.*#quot;"]

    Publisher -->|"1. PublishAsync(envelope, destination)<br/>Destination.Address: #quot;orders#quot;<br/>Destination.RoutingKey: #quot;order.created#quot;"| Transport
    Transport -->|"2. BasicPublish()"| Exchange
    Exchange -->|"3. Route by pattern<br/>Routing Key: #quot;order.created#quot;"| Queue

    class Publisher layer-core
    class Transport,Exchange,Queue layer-command

Subscribing¶

flowchart TD
    Subscriber["Subscriber (Fulfillment Service)"]
    Transport["RabbitMQTransport<br/><br/>- Create dedicated channel<br/>- Set QoS prefetch (default: 200)<br/>- Declare exchange<br/>- Declare queue with DLX<br/>- Bind queue to exchange<br/>- Create AsyncEventingBasicConsumer"]
    Handler["Batch Handler<br/><br/>- Check subscription.IsActive<br/>- Collect deliveries into a batch<br/>- Deserialize via EnvelopeType<br/>- Invoke batch handler<br/>- BasicAck each on success<br/>- BasicNack + requeue on failure<br/>- BasicNack → DLQ after max retries"]

    Subscriber -->|"1. SubscribeBatchAsync(batchHandler, destination, batchOptions)<br/>Destination.Address: #quot;orders#quot;<br/>Destination.RoutingKey: #quot;fulfillment-orders-queue#quot;"| Transport
    Transport -->|"2. Receive BasicDeliver event"| Handler

    class Subscriber layer-core
    class Transport layer-command
    class Handler layer-core

Installation¶

Package Reference¶

NuGet Package: Whizbang.Transports.RabbitMQ (when published)

Package Reference

<PackageReference Include="Whizbang.Transports.RabbitMQ" Version="0.1.0" />

Dependencies¶

Dependencies

<ItemGroup>
  <PackageReference Include="RabbitMQ.Client" Version="7.2.0" />
  <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.2" />
  <PackageReference Include="Microsoft.Extensions.Diagnostics.HealthChecks" Version="10.0.2" />
</ItemGroup>

Running the Example Project¶

The ECommerce sample demonstrates cross-service event distribution using Aspire orchestration with switchable transport providers. The same application code runs with either RabbitMQ or Azure Service Bus using compiler directives.

Architecture Overview¶

The sample includes: - OrderService.API: Handles order commands and workflows - InventoryWorker: Manages inventory and publishes stock events - PaymentWorker: Processes payment transactions - ShippingWorker: Handles fulfillment and shipping - NotificationWorker: Sends customer notifications - BFF.API: Aggregates cross-service data via perspectives (GraphQL + REST) - ECommerce.UI: Angular frontend (port 4200)

Cross-Service Event Flow:

OrderService → "orders" topic → [Payment, Shipping, Inventory, Notification, BFF]
InventoryWorker → "products" topic → [BFF, InventoryWorker]
PaymentWorker → "payments" topic → [BFF]
ShippingWorker → "shipping" topic → [BFF]

Transport Provider Selection¶

The ECommerce sample uses compiler directives to switch between transports at build time:

Transport Provider Selection

<!-- Directory.Build.props -->
<DefineConstants Condition="'$(TransportProvider)' == ''">AZURESERVICEBUS</DefineConstants>
<DefineConstants Condition="'$(TransportProvider)' == 'RabbitMQ'">RABBITMQ</DefineConstants>
<DefineConstants Condition="'$(TransportProvider)' == 'AzureServiceBus'">AZURESERVICEBUS</DefineConstants>

Default: Azure Service Bus Switch to RabbitMQ: Add /p:TransportProvider=RabbitMQ to build/run commands

Prerequisites¶

Required Tools: - .NET 10 SDK (or later) - Docker Desktop (for emulators) - Node.js 20+ (for Angular UI) - .NET Aspire Workload: dotnet workload install aspire

Verify Installation: Prerequisites

dotnet --version  # Should be 10.0.1 or later
docker --version  # Should be 20.10 or later
node --version    # Should be 20.0 or later
dotnet workload list | grep aspire  # Should show aspire workload

Running with RabbitMQ¶

1. Build with RabbitMQ Transport: Running with RabbitMQ

cd samples/ECommerce

# Build all projects with RabbitMQ transport
dotnet build /p:TransportProvider=RabbitMQ

2. Start Aspire AppHost: Running with RabbitMQ

cd ECommerce.AppHost

# Run with RabbitMQ (Aspire dashboard at https://localhost:17036)
dotnet run /p:TransportProvider=RabbitMQ

What Aspire Does: - ✅ Starts RabbitMQ container (port 5672, management UI at 15672) - ✅ Starts PostgreSQL container (port 5432, pgAdmin at 5050) - ✅ Creates exchanges (orders, products, payments, shipping, inbox) - ✅ Creates queue bindings with routing patterns - ✅ Initializes 7 microservices with dependency injection - ✅ Starts Angular UI at http://localhost:4200 - ✅ Provides Aspire Dashboard for observability

3. Access Services: Running with RabbitMQ

# Aspire Dashboard
https://localhost:17036

# RabbitMQ Management UI (guest/guest)
http://localhost:15672

# PostgreSQL pgAdmin (admin@admin.com/admin)
http://localhost:5050

# Angular UI
http://localhost:4200

# BFF Swagger UI
http://localhost:5234/swagger

# BFF GraphQL Playground
http://localhost:5234/graphql

4. View RabbitMQ Topology: - Navigate to Exchanges tab → See orders, products, payments, shipping, inbox - Navigate to Queues tab → See all service queues with bindings - Navigate to Connections tab → See all microservice connections

5. Stop All Services: Running with RabbitMQ

# Ctrl+C in terminal running AppHost
# Containers persist by default (ContainerLifetime.Persistent)

# To clean up containers:
docker stop rabbitmq postgres pgadmin
docker rm rabbitmq postgres pgadmin

Running with Azure Service Bus Emulator¶

1. Build with Azure Service Bus Transport: Running with Azure Service Bus Emulator

cd samples/ECommerce

# Build with Azure Service Bus (default)
dotnet build

# Or explicitly:
dotnet build /p:TransportProvider=AzureServiceBus

2. Start Aspire AppHost: Running with Azure Service Bus Emulator

cd ECommerce.AppHost

# Run with Azure Service Bus (Aspire dashboard at https://localhost:17036)
dotnet run

What Aspire Does: - ✅ Starts Azure Service Bus Emulator (port 5672) - ✅ Creates topics (orders, products, payments, shipping, inbox) - ✅ Creates subscriptions with correlation filters - ✅ Starts PostgreSQL container (port 5432, pgAdmin at 5050) - ✅ Initializes 7 microservices with dependency injection - ✅ Starts Angular UI at http://localhost:4200 - ✅ Provides Aspire Dashboard for observability

3. Access Services (same as RabbitMQ except no Management UI): Running with Azure Service Bus Emulator

# Aspire Dashboard
https://localhost:17036

# PostgreSQL pgAdmin (admin@admin.com/admin)
http://localhost:5050

# Angular UI
http://localhost:4200

# BFF Swagger UI
http://localhost:5234/swagger

# BFF GraphQL Playground
http://localhost:5234/graphql

Note: Azure Service Bus Emulator has no management UI. Use Aspire Dashboard to monitor service health.

Running Integration Tests¶

The ECommerce sample includes dedicated integration test projects for each transport: - ECommerce.Integration.Tests → Azure Service Bus (48 lifecycle tests) - ECommerce.RabbitMQ.Integration.Tests → RabbitMQ (48 lifecycle tests)

Why Separate Test Projects?: - Different fixtures for emulator management (shared vs per-test containers) - Different topic/exchange isolation strategies - Different cleanup and drain logic

RabbitMQ Integration Tests¶

1. Start RabbitMQ Container: RabbitMQ Integration Tests

docker run -d \
  --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  rabbitmq:4-management

2. Run Tests: RabbitMQ Integration Tests (2)

cd samples/ECommerce/tests/ECommerce.RabbitMQ.Integration.Tests

# Run all 48 lifecycle tests (sequential execution)
dotnet test

# Run specific test
dotnet run -- --treenode-filter "/*/*/*/RestockInventory_FromZeroStock_IncreasesCorrectlyAsync"

Test Duration: ~1.5-2 minutes (48 tests, ~1-2s each)

What Tests Validate: - Cross-service event publication (InventoryWorker → RabbitMQ → BFF) - Topic exchange routing with test-specific exchanges - Dead letter queue handling (automatic DLX/DLQ creation) - Perspective materialization from cross-service events - Channel pool thread-safety under concurrent load - Subscription pause/resume lifecycle

Azure Service Bus Integration Tests¶

1. Start Service Bus Emulator: Azure Service Bus Integration Tests

docker run -d \
  --name servicebus-emulator \
  -p 5672:5672 \
  -v $(pwd)/samples/ECommerce/tests/ECommerce.Integration.Tests/Config-Named.json:/ServiceBus_Emulator/ConfigFiles/Config.json \
  mcr.microsoft.com/azure-messaging/servicebus-emulator:latest

2. Run Tests: Azure Service Bus Integration Tests (2)

cd samples/ECommerce/tests/ECommerce.Integration.Tests

# Run all 48 lifecycle tests (sequential execution)
dotnet test

# Run specific test
dotnet run -- --treenode-filter "/*/*/*/SeedProducts_CreatesProducts_AndPerspectivesCompleteAsync"

Test Duration: ~2-2.5 minutes (emulator startup adds 45-60 seconds on first run)

Key Differences Between Transports¶

Aspect	RabbitMQ	Azure Service Bus
AppHost Setup	`AddRabbitMQ()` with exchanges/queues	`AddAzureServiceBus()` with topics/subscriptions
Topic Creation	Dynamic (created on first publish)	Static (pre-defined via Aspire or Config.json)
Management UI	✅ Built-in (port 15672)	❌ No UI (use Aspire Dashboard)
Routing	Wildcard patterns (`product.*`, `#`)	Subscription filters (CorrelationId, MessageId)
DLQ Handling	Automatic DLX/DLQ creation	Requires explicit subscription configuration
Startup Time	~10 seconds	~45-60 seconds (emulator initialization)
Test Isolation	Test-specific exchanges (`inventory-{testId}`)	Shared topics (`topic-00`, `topic-01`)

Troubleshooting¶

AppHost won't start¶

AppHost won't start

# Ensure Aspire workload is installed
dotnet workload install aspire

# Check Docker is running
docker ps

# Rebuild all projects
cd samples/ECommerce
dotnet clean
dotnet build /p:TransportProvider=RabbitMQ  # or AzureServiceBus

RabbitMQ container fails to start¶

RabbitMQ container fails to start

# Check port conflicts
lsof -i :5672
lsof -i :15672

# Remove existing container
docker stop rabbitmq
docker rm rabbitmq

# Start fresh
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4-management

Azure Service Bus Emulator not ready¶

Azure Service Bus Emulator not ready

# Check emulator logs
docker logs servicebus-emulator

# Wait for "Emulator is ready" message (can take 60 seconds)

Integration tests timeout¶

Integration tests timeout

# Ensure emulator is running
docker ps | grep rabbitmq  # or servicebus-emulator

# Check emulator logs
docker logs rabbitmq

# Restart emulator if needed
docker restart rabbitmq

Wrong transport used at runtime¶

Wrong transport used at runtime

# Verify build used correct transport
cd samples/ECommerce
dotnet clean
dotnet build /p:TransportProvider=RabbitMQ  # or AzureServiceBus

# Check generated symbols in .whizbang/cache folder
grep -r "RABBITMQ\|AZURESERVICEBUS" ECommerce.InventoryWorker/.whizbang/cache

Configuration¶

Basic Setup¶

Basic Setup

using Whizbang.Transports.RabbitMQ;

var builder = WebApplication.CreateBuilder(args);

// Register RabbitMQ transport
builder.Services.AddRabbitMQTransport(
    connectionString: "amqp://guest:guest@localhost:5672/",
    configureOptions: options => {
        options.MaxChannels = 20;                      // Channel pool size
        options.MaxDeliveryAttempts = 5;               // Retry limit before DLQ
        options.PrefetchCount = 10;                    // QoS prefetch count
        options.AutoDeclareDeadLetterExchange = true;  // Auto-create DLX/DLQ
    }
);

// Add health checks
builder.Services.AddRabbitMQHealthChecks();

var app = builder.Build();

// Health check endpoint
app.MapHealthChecks("/health");

app.Run();

Configuration Options¶

Property	Default	Description
`MaxChannels`	10	Maximum pooled channels for publishing
`MaxDeliveryAttempts`	10	Retry limit before dead-lettering
`DefaultQueueName`	`null`	Fallback queue name if not specified
`PrefetchCount`	200	QoS prefetch count per consumer (high default sized for batch receive; match to `TransportBatchOptions.BatchSize`)
`AutoDeclareDeadLetterExchange`	`true`	Auto-create DLX and DLQ
`EnableSingleActiveConsumer`	`false`	Enable Single Active Consumer for FIFO ordering

Single Active Consumer (FIFO Ordering)¶

RabbitMQ guarantees per-publisher per-channel ordering, but with multiple consumers, messages can be processed out of order. Single Active Consumer (SAC) ensures only one consumer is active at a time per queue, preserving FIFO ordering.

When EnableSingleActiveConsumer is true:

Queues are declared with x-single-active-consumer: true
Only one consumer receives messages at a time — guaranteeing FIFO ordering
If the active consumer disconnects, RabbitMQ promotes another consumer automatically
The transport claims TransportCapabilities.Ordered only when SAC is enabled

Enable FIFO Ordering

builder.Services.AddRabbitMQTransport(
    connectionString: "amqp://guest:guest@localhost:5672",
    configureOptions: options => {
        options.EnableSingleActiveConsumer = true;  // Enable FIFO ordering
    }
);

:::note SAC limits throughput to one consumer per queue. For scaling with FIFO guarantees, consider using consistent hash exchange to pin streams to specific queues, each with SAC enabled. :::

Domain Topic Auto-Provisioning¶

New

When you declare domain ownership via OwnDomains(), Whizbang automatically provisions topic exchanges at worker startup.

The RabbitMQInfrastructureProvisioner is automatically registered and creates topic exchanges for owned domains:

Domain Topic Auto-Provisioning

services.AddWhizbang()
    .WithRouting(routing => {
        routing.OwnDomains("myapp.users", "myapp.orders");
    })
    .AddTransportConsumer();

// At startup, these exchanges are auto-created:
// - myapp.users (type: topic, durable: true)
// - myapp.orders (type: topic, durable: true)

Key behaviors: - Exchange names are lowercased for consistency - Exchange declaration is idempotent (safe if already exists) - Provisioning happens before subscriptions are created - Multiple service instances can provision concurrently (race-safe)

📖 See Domain Topic Provisioning for full details.

Connection Retry Options¶

The transport includes built-in connection retry with exponential backoff for handling transient connection failures:

Property	Default	Description
`InitialRetryAttempts`	5	Initial retry attempts with warning logs
`InitialRetryDelay`	1 second	Delay before first retry
`MaxRetryDelay`	120 seconds	Maximum delay (caps exponential backoff)
`BackoffMultiplier`	2.0	Multiplier for exponential backoff
`RetryIndefinitely`	`true`	Continue retrying after initial attempts

Example Configuration: Connection Retry Options

builder.Services.AddRabbitMQTransport(
    connectionString: "amqp://guest:guest@localhost:5672/",
    configureOptions: options => {
        // Connection retry settings
        options.InitialRetryAttempts = 10;              // More warnings for slow containers
        options.InitialRetryDelay = TimeSpan.FromSeconds(2);
        options.MaxRetryDelay = TimeSpan.FromMinutes(2);
        options.BackoffMultiplier = 1.5;
        options.RetryIndefinitely = true;               // Keep trying until success
    }
);

Retry Behavior (with defaults): 1. Initial attempt → fails 2. Wait 1s → retry 1 (logged as warning) 3. Wait 2s → retry 2 (logged as warning) 4. Wait 4s → retry 3 (logged as warning) 5. Wait 8s → retry 4 (logged as warning) 6. Wait 16s → retry 5 (logged as warning) 7. Continue retrying indefinitely at intervals up to 120s (logged every 10 attempts)

Key Behaviors: - Initial Phase: First 5 attempts log warnings for each failure - Indefinite Phase: After initial attempts, continues retrying (logged less frequently) - Capped Backoff: Delay never exceeds MaxRetryDelay (default 120s) - Graceful Shutdown: Responds to cancellation token for clean shutdown

Use Cases: - Container Startup: RabbitMQ container may take 10-30 seconds to become ready - Network Glitches: Temporary network issues during service startup - Cluster Failover: RabbitMQ cluster switching to different node - Infrastructure Outage: Service survives extended outages and reconnects automatically

Fail Fast (disable indefinite retry): Connection Retry Options (2)

options.RetryIndefinitely = false;  // Throws after InitialRetryAttempts

Runtime Reconnection¶

RabbitMQ transport uses the RabbitMQ client's built-in Automatic Recovery feature for runtime reconnection. When a connection is lost during operation:

Automatic Detection: Connection shutdown is detected immediately
Automatic Recovery: Client attempts to reconnect automatically
Channel Recovery: All channels and consumers are automatically re-established
Topology Recovery: Exchanges, queues, and bindings are automatically re-declared

Connection State Monitoring: The transport logs connection state changes for observability: - ConnectionShutdown → Warning with reason code and message - RecoverySucceeded → Information that connection recovered - ConnectionRecoveryError → Error with exception details - ConnectionBlocked → Warning when broker blocks the connection (resource alarm) - ConnectionUnblocked → Information when normal operation resumes

Configuration: Runtime Reconnection

// NetworkRecoveryInterval is set to match InitialRetryDelay
options.InitialRetryDelay = TimeSpan.FromSeconds(5);  // Recovery interval = 5s

No Manual Reconnection Needed: The RabbitMQ client handles all reconnection automatically. Your application code continues to work transparently after recovery

Connection String Format¶

amqp://username:password@hostname:port/virtualhost
amqps://username:password@hostname:port/virtualhost  # TLS

Examples: - Local development: amqp://guest:guest@localhost:5672/ - Production: amqps://prod-user:secret@rabbitmq.example.com:5671/production - Docker: amqp://guest:guest@rabbitmq:5672/

Usage¶

Publishing Messages¶

Publishing Messages

public class ProductService {
    private readonly ITransport _transport;
    private readonly ILogger<ProductService> _logger;

    public ProductService(ITransport transport, ILogger<ProductService> logger) {
        _transport = transport;
        _logger = logger;
    }

    public async Task CreateProductAsync(CreateProductCommand command) {
        // Create message envelope
        var envelope = new MessageEnvelope<ProductCreatedEvent> {
            MessageId = MessageId.New(),
            Payload = new ProductCreatedEvent {
                ProductId = command.ProductId,
                Name = command.Name,
                Price = command.Price,
                CreatedAt = DateTime.UtcNow
            }
        };

        // Publish to exchange with routing key
        var destination = new TransportDestination(
            Address: "products",                    // Exchange name
            RoutingKey: "product.created",          // Routing key
            Metadata: new Dictionary<string, JsonElement> {
                ["Priority"] = JsonSerializer.SerializeToElement(5)
            }
        );

        await _transport.PublishAsync(envelope, destination);

        _logger.LogInformation("Published ProductCreatedEvent for {ProductId}", command.ProductId);
    }
}

Subscribing to Messages¶

Subscribing to Messages

public class InventoryWorker : BackgroundService {
    private readonly ITransport _transport;
    private readonly ILogger<InventoryWorker> _logger;
    private ISubscription? _subscription;

    public InventoryWorker(ITransport transport, ILogger<InventoryWorker> logger) {
        _transport = transport;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
        var destination = new TransportDestination(
            Address: "products",                           // Exchange name
            RoutingKey: "inventory-products-queue",        // Queue name
            Metadata: new Dictionary<string, JsonElement> {
                ["RoutingPattern"] = JsonSerializer.SerializeToElement("product.*")
            }
        );

        // ITransport exposes batch receive; the handler is invoked once per batch.
        _subscription = await _transport.SubscribeBatchAsync(
            batchHandler: async (messages, ct) => {
                foreach (var message in messages) {
                    _logger.LogInformation("Received message: {MessageId}", message.Envelope.MessageId);

                    if (message.Envelope.Payload is ProductCreatedEvent evt) {
                        await HandleProductCreatedAsync(evt, ct);
                    }
                }
            },
            destination,
            new TransportBatchOptions(),  // BatchSize 200, SlideMs 20, MaxWaitMs 1000
            stoppingToken
        );

        _logger.LogInformation("Subscribed to products exchange");
    }

    private async Task HandleProductCreatedAsync(ProductCreatedEvent evt, CancellationToken ct) {
        // Update inventory levels
        _logger.LogInformation("Handling ProductCreatedEvent for {ProductId}", evt.ProductId);
        // ... business logic
    }

    public override async Task StopAsync(CancellationToken ct) {
        if (_subscription != null) {
            await _subscription.DisposeAsync();
        }
        await base.StopAsync(ct);
    }
}

Custom Routing Patterns¶

Custom Routing Patterns

// Subscribe to specific event types
var destination = new TransportDestination(
    Address: "products",
    RoutingKey: "analytics-queue",
    Metadata: new Dictionary<string, JsonElement> {
        ["RoutingPattern"] = JsonSerializer.SerializeToElement("product.created")
    }
);

// Subscribe to all product events
var destination = new TransportDestination(
    Address: "products",
    RoutingKey: "audit-queue",
    Metadata: new Dictionary<string, JsonElement> {
        ["RoutingPattern"] = JsonSerializer.SerializeToElement("product.*")
    }
);

// Subscribe to all messages
var destination = new TransportDestination(
    Address: "products",
    RoutingKey: "logger-queue",
    Metadata: new Dictionary<string, JsonElement> {
        ["RoutingPattern"] = JsonSerializer.SerializeToElement("#")
    }
);

Pause and Resume Subscriptions¶

Pause and Resume Subscriptions

public class OrderProcessor {
    private ISubscription? _subscription;

    public async Task PauseProcessingAsync() {
        if (_subscription != null) {
            await _subscription.PauseAsync();
            // Messages will be nack'd with requeue while paused
        }
    }

    public async Task ResumeProcessingAsync() {
        if (_subscription != null) {
            await _subscription.ResumeAsync();
            // Message processing continues
        }
    }
}

Dead Letter Queues¶

Automatic DLX/DLQ Setup¶

When AutoDeclareDeadLetterExchange = true (default), the transport automatically creates:

Dead Letter Exchange ({exchange}.dlx): Fanout exchange for failed messages
Dead Letter Queue ({queue}.dlq): Queue storing permanently failed messages
DLX Binding: Main queue declares x-dead-letter-exchange argument

Message Retry Flow¶

flowchart TD
    Main["Main Queue #quot;orders-queue#quot;<br/><br/>Delivery attempt 1 → Nack<br/>Delivery attempt 2 → Nack<br/>...<br/>Attempt 10 (max) → Nack<br/><br/>x-dead-letter-exchange set"]
    DLX["Dead Letter Exchange<br/>#quot;orders.dlx#quot;"]
    DLQ["Dead Letter Queue #quot;orders-queue.dlq#quot;<br/><br/>Permanently failed messages stored"]

    Main -->|"Message moved to DLX"| DLX
    DLX -->|"Fanout routing"| DLQ

    class Main layer-command
    class DLX,DLQ layer-event

Inspecting Failed Messages¶

Inspecting Failed Messages

# List messages in DLQ
rabbitmqadmin get queue=orders-queue.dlq count=10

# Republish message from DLQ (manual intervention)
rabbitmqadmin get queue=orders-queue.dlq requeue=true

Health Checks¶

ASP.NET Core Health Checks¶

ASP.NET Core Health Checks

using Whizbang.Transports.RabbitMQ;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddRabbitMQTransport("amqp://localhost:5672/");
builder.Services.AddRabbitMQHealthChecks();

var app = builder.Build();

app.MapHealthChecks("/health");

app.Run();

Health Check Response: ASP.NET Core Health Checks (2)

{
  "status": "Healthy",
  "results": {
    "rabbitmq": {
      "status": "Healthy",
      "description": "RabbitMQ transport is healthy",
      "data": {}
    }
  }
}

Custom Readiness Checks¶

Custom Readiness Checks

public class RabbitMQReadinessCheck : ITransportReadinessCheck {
    private readonly IConnection _connection;

    public RabbitMQReadinessCheck(IConnection connection) {
        _connection = connection;
    }

    public Task<bool> IsReadyAsync(CancellationToken cancellationToken = default) {
        return Task.FromResult(_connection.IsOpen);
    }
}

Testing¶

Unit Testing with Test Doubles¶

Unit Testing with Test Doubles

using Whizbang.Transports.RabbitMQ;
using Whizbang.Transports.RabbitMQ.Tests;

public class ProductServiceTests {
    [Test]
    public async Task CreateProductAsync_PublishesEvent() {
        // Arrange - fake IConnection/IChannel doubles (see tests/Whizbang.Transports.RabbitMQ.Tests/TestDoubles.cs)
        var fakeChannel = new FakeChannel();
        var fakeConnection = new FakeConnection(() => Task.FromResult<IChannel>(fakeChannel));

        var options = new RabbitMQOptions();
        var transport = new RabbitMQTransport(
            fakeConnection,
            JsonContextRegistry.CreateCombinedOptions(),
            new RabbitMQChannelPool(fakeConnection, options.MaxChannels),
            options,
            NullLogger<RabbitMQTransport>.Instance
        );

        var service = new ProductService(transport, NullLogger<ProductService>.Instance);

        // Act
        await service.CreateProductAsync(new CreateProductCommand {
            ProductId = ProductId.From(Guid.NewGuid()),
            Name = "Test Product",
            Price = 10.00m
        });

        // Assert
        await Assert.That(fakeChannel.BasicPublishAsyncCalled).IsTrue();
        await Assert.That(fakeChannel.PublishedMessages[0].Exchange).IsEqualTo("products");
        await Assert.That(fakeChannel.PublishedMessages[0].RoutingKey).IsEqualTo("product.created");
    }
}

Integration Testing with TestContainers¶

Integration Testing with TestContainers

using Testcontainers.RabbitMQ;

[NotInParallel]  // RabbitMQ container isolation
public class RabbitMQIntegrationTests {
    private RabbitMqContainer? _container;
    private ITransport? _transport;

    [Before(Test)]
    public async Task SetupAsync() {
        // Start RabbitMQ container
        _container = new RabbitMqBuilder()
            .WithImage("rabbitmq:3.13-management-alpine")
            .WithPortBinding(5672, 5672)
            .WithPortBinding(15672, 15672)
            .Build();

        await _container.StartAsync();

        // Create transport
        var services = new ServiceCollection();
        services.AddRabbitMQTransport(_container.GetConnectionString());
        var provider = services.BuildServiceProvider();

        _transport = provider.GetRequiredService<ITransport>();
    }

    [Test]
    public async Task PublishAndSubscribe_MessageReceivedAsync() {
        // Arrange
        var receivedEvent = new TaskCompletionSource<ProductCreatedEvent>();

        var destination = new TransportDestination(
            Address: "test-products",
            RoutingKey: "test-queue"
        );

        // Subscribe (batch receive - handler invoked once per collected batch)
        await _transport!.SubscribeBatchAsync(
            batchHandler: (messages, ct) => {
                foreach (var message in messages) {
                    if (message.Envelope.Payload is ProductCreatedEvent evt) {
                        receivedEvent.SetResult(evt);
                    }
                }
                return Task.CompletedTask;
            },
            destination,
            new TransportBatchOptions(),
            CancellationToken.None
        );

        // Act - Publish
        var expected = new ProductCreatedEvent {
            ProductId = ProductId.From(Guid.NewGuid()),
            Name = "Integration Test Product",
            Price = 99.99m
        };

        var envelope = new MessageEnvelope<ProductCreatedEvent> {
            MessageId = MessageId.New(),
            Payload = expected
        };

        await _transport.PublishAsync(envelope, destination);

        // Assert
        var received = await receivedEvent.Task.WaitAsync(TimeSpan.FromSeconds(5));
        await Assert.That(received.ProductId).IsEqualTo(expected.ProductId);
        await Assert.That(received.Name).IsEqualTo(expected.Name);
    }

    [After(Test)]
    public async Task TeardownAsync() {
        if (_transport != null) {
            await _transport.DisposeAsync();
        }
        if (_container != null) {
            await _container.StopAsync();
        }
    }
}

Best Practices¶

1. Channel Pool Sizing¶

Guideline: Set MaxChannels based on concurrent publishing threads.

Channel Pool Sizing

// Low throughput (< 10 msg/sec)
options.MaxChannels = 10;  // Default

// Medium throughput (10-100 msg/sec)
options.MaxChannels = 20;

// High throughput (> 100 msg/sec)
options.MaxChannels = 50;

Why: Channels are lightweight, but excessive pooling wastes resources. Profile your workload.

2. Prefetch Count Tuning¶

Guideline: Set PrefetchCount based on message processing time.

Prefetch Count Tuning

// Batch-receive workloads (default: 200, matched to TransportBatchOptions.BatchSize)
options.PrefetchCount = 200;

// Fast processing (< 100ms per message)
options.PrefetchCount = 20;

// Medium processing (100ms - 1s)
options.PrefetchCount = 10;

// Slow processing (> 1s per message)
options.PrefetchCount = 1;

Why: Higher prefetch improves throughput but increases memory usage and delays redelivery on failure. The default (200) is sized for the transport consumer's batch receive path.

3. Retry Limits¶

Guideline: Set MaxDeliveryAttempts based on failure characteristics.

Retry Limits

// Transient failures (network glitches)
options.MaxDeliveryAttempts = 3;

// Intermittent failures (external API timeouts)
options.MaxDeliveryAttempts = 5;

// Persistent failures (message format errors)
options.MaxDeliveryAttempts = 1;  // Fail fast to DLQ

Why: Excessive retries delay DLQ routing and waste resources.

4. Exchange and Queue Naming¶

Convention: Use hierarchical names for topic routing.

Exchange and Queue Naming

// Good - hierarchical routing
Address: "ecommerce.products"
RoutingKey: "product.created"

// Good - tenant isolation
Address: "tenant-123.orders"
RoutingKey: "order.*"

// Avoid - flat namespace
Address: "products"
RoutingKey: "created"

5. Dead Letter Queue Monitoring¶

Setup alerting for DLQ depth:

Dead Letter Queue Monitoring

# Check DLQ message count
rabbitmqadmin list queues name messages | grep dlq

# Set CloudWatch/Prometheus alert
# If dlq_messages > threshold, investigate

Why: Messages in DLQ indicate persistent failures requiring manual intervention.

Capabilities¶

The RabbitMQ transport supports the following TransportCapabilities:

Capability	Supported	Notes
PublishSubscribe	✅ Yes	Topic exchanges with wildcard routing
Reliable	✅ Yes	At-least-once delivery with retries
BulkPublish	✅ Yes	Bulk publish over pooled channels
Ordered	⚠️ Conditional	Claimed only when `EnableSingleActiveConsumer` is true
RequestResponse	❌ No	Not implemented in v1.0.0
ExactlyOnce	❌ No	Use inbox/outbox pattern (Whizbang.Core)

Ordering Considerations: - Single Active Consumer enabled: FIFO ordering, transport claims TransportCapabilities.Ordered - Multiple consumers (SAC disabled): No ordering guarantee - Use partitioning or Azure Service Bus for strict ordering at scale

Troubleshooting¶

Connection Refused¶

Symptom: BrokerUnreachableException: None of the specified endpoints were reachable

Causes: 1. RabbitMQ server not running 2. Incorrect connection string 3. Firewall blocking port 5672

Solution: Connection Refused

# Check RabbitMQ is running
docker ps | grep rabbitmq

# Verify connection
telnet localhost 5672

# Check RabbitMQ logs
docker logs <rabbitmq-container>

Channel Pool Exhaustion¶

Symptom: PublishAsync() hangs or times out

Cause: All channels rented, none returned (likely exception in using block)

Solution: Channel Pool Exhaustion

// Ensure channel returns on exception
try {
    using (var channel = await pool.RentAsync()) {
        await transport.PublishAsync(envelope, destination);
    }  // Channel auto-returns here
} catch (Exception ex) {
    _logger.LogError(ex, "Publish failed");
    throw;
}

Messages Not Routed¶

Symptom: Messages published but not received

Causes: 1. Exchange/queue binding mismatch 2. Incorrect routing key pattern 3. Queue not declared

Diagnosis: Messages Not Routed

# Check exchange exists
rabbitmqadmin list exchanges name type

# Check queue bindings
rabbitmqadmin list bindings source destination

# Inspect queue
rabbitmqadmin list queues name messages

Dead Letter Loop¶

Symptom: Messages cycling between queue and DLQ

Cause: Handler always fails, message requeued from DLQ

Solution: Disable DLQ auto-requeue or fix handler logic.

Performance Considerations¶

Throughput Characteristics¶

Baseline (local RabbitMQ, default settings): - Publish: ~2,000-5,000 msg/sec - Subscribe: ~1,000-3,000 msg/sec (depends on handler)

Tuning for High Throughput: 1. Increase MaxChannels (50-100) 2. Increase PrefetchCount (20-50) 3. Use multiple consumers 4. Disable unnecessary plugins (management UI)

Latency Characteristics¶

Typical Latencies (local RabbitMQ): - Publish: 1-5 ms - End-to-End: 10-50 ms (depends on handler)

Reducing Latency: 1. Reduce PrefetchCount (1-5) 2. Use faster serialization (System.Text.Json with source generation) 3. Optimize handler logic 4. Co-locate publisher and consumer

Transports Overview - Compare transport implementations
Azure Service Bus Transport - Alternative transport with strict ordering
In-Memory Transport - Testing and development
Custom Transports - Implement your own transport
Lifecycle Hooks - Pre/post processing for messages
Observability - OpenTelemetry tracing

Code References¶

Core Implementation¶

RabbitMQTransport.cs - Main transport implementation
RabbitMQChannelPool.cs - Thread-safe channel pooling
RabbitMQOptions.cs - Configuration options
RabbitMQSubscription.cs - Subscription lifecycle

Tests¶

RabbitMQTransportTests.cs - Unit tests
RabbitMQChannelPoolTests.cs - Channel pool tests
Integration Tests - TestContainers-based end-to-end tests

Summary¶

The RabbitMQ transport provides a robust, performant foundation for distributed event-driven architectures with:

✅ Reliable Delivery - At-least-once guarantees with automatic retries ✅ Flexible Routing - Topic exchanges with wildcard patterns ✅ AOT-Compatible - Source-generated JSON serialization ✅ Thread-Safe - Channel pooling for concurrent publishing ✅ Dead Letter Queues - Automatic failure handling and observability ✅ Production-Ready - Health checks, pause/resume, TestContainers integration

Next Steps: 1. Install and configure the RabbitMQ transport 2. Review usage examples for publishing and subscribing 3. Explore integration tests for real-world patterns 4. Read best practices for production deployments

RabbitMQ Transport¶

Why RabbitMQ?¶

Architecture¶

Topic Exchange Pattern¶

Channel Pool Architecture¶

Message Flow¶

Publishing¶

Subscribing¶

Installation¶

Package Reference¶

Dependencies¶

Running the Example Project¶

Architecture Overview¶

Transport Provider Selection¶

Prerequisites¶

Running with RabbitMQ¶

Running with Azure Service Bus Emulator¶

Running Integration Tests¶

RabbitMQ Integration Tests¶

Azure Service Bus Integration Tests¶

Key Differences Between Transports¶

Troubleshooting¶

AppHost won't start¶

RabbitMQ container fails to start¶

Azure Service Bus Emulator not ready¶

Integration tests timeout¶

Wrong transport used at runtime¶

Configuration¶

Basic Setup¶

Configuration Options¶

Single Active Consumer (FIFO Ordering)¶

Domain Topic Auto-Provisioning¶

Connection Retry Options¶

Runtime Reconnection¶

Connection String Format¶

Usage¶

Publishing Messages¶

Subscribing to Messages¶

Custom Routing Patterns¶

Pause and Resume Subscriptions¶

Dead Letter Queues¶

Automatic DLX/DLQ Setup¶

Message Retry Flow¶

Inspecting Failed Messages¶

Health Checks¶

ASP.NET Core Health Checks¶

Custom Readiness Checks¶

Testing¶

Unit Testing with Test Doubles¶

Integration Testing with TestContainers¶

Best Practices¶

1. Channel Pool Sizing¶

2. Prefetch Count Tuning¶

3. Retry Limits¶

4. Exchange and Queue Naming¶

5. Dead Letter Queue Monitoring¶

Capabilities¶

Troubleshooting¶

Connection Refused¶

Channel Pool Exhaustion¶

Messages Not Routed¶

Dead Letter Loop¶

Performance Considerations¶

Throughput Characteristics¶

Latency Characteristics¶

Related Topics¶

Code References¶

Core Implementation¶

Tests¶

Summary¶