NetEvolve.Pulse.RabbitMQ 0.84.106

NetEvolve.Pulse.RabbitMQ

RabbitMQ transport for the Pulse outbox pattern. Publishes outbox messages directly to RabbitMQ exchanges using the official .NET client, enabling reliable event delivery without routing through Dapr or other intermediaries.

Features

• Direct Publishing: Send messages to RabbitMQ exchanges without additional infrastructure
• Flexible Routing: Automatic routing key resolution based on event types via ITopicNameResolver
• Health Checks: Verify connection and channel state for readiness probing
• Delivery Guarantees: Publisher confirms, persistent delivery mode and mandatory routing on every publish
• Batch Support: Batches publish on one channel and await their confirmations together
• Connection Management: Singleton connection with lazy channel initialization

Installation

NuGet Package Manager

Install-Package NetEvolve.Pulse.RabbitMQ

.NET CLI

dotnet add package NetEvolve.Pulse.RabbitMQ

PackageReference

<PackageReference Include="NetEvolve.Pulse.RabbitMQ" Version="x.x.x" />

Quick Start

1. Add the RabbitMQ client package

dotnet add package RabbitMQ.Client

2. Register services

using Microsoft.Extensions.DependencyInjection;
using NetEvolve.Pulse;
using RabbitMQ.Client;

var services = new ServiceCollection();

// Register RabbitMQ connection before UseRabbitMqTransport
services.AddSingleton<IConnection>(sp =>
{
    var factory = new ConnectionFactory
    {
        HostName = "localhost",
        Port = 5672,
        VirtualHost = "/",
        UserName = "guest",
        Password = "guest"
    };
    return factory.CreateConnectionAsync().GetAwaiter().GetResult();
});

services.AddPulse(config => config
    .AddOutbox(
        options => options.Schema = "pulse",
        processorOptions => processorOptions.BatchSize = 100)
    .UseRabbitMqTransport(options =>
    {
        options.ExchangeName = "events";
    }));

3. Store events via IEventOutbox

Use IEventOutbox to store events reliably. The outbox processor picks them up and publishes each one to the configured RabbitMQ exchange:

public class OrderService
{
    private readonly IEventOutbox _outbox;

    public OrderService(IEventOutbox outbox) => _outbox = outbox;

    public async Task CreateOrderAsync(CreateOrderRequest request, CancellationToken ct)
    {
        // ... business logic ...

        // Stored reliably; published via RabbitMQ when the processor runs
        await _outbox.StoreAsync(new OrderCreatedEvent
        {
            OrderId = Guid.NewGuid(),
            CustomerId = request.CustomerId
        }, ct);
    }
}

Transaction Integration

For reliable at-least-once delivery guarantees, store outbox events within the same database transaction as your business data. Pair the RabbitMQ transport with a persistence provider that supports transaction enlistment:

public class OrderService
{
    private readonly ApplicationDbContext _context;
    private readonly IEventOutbox _outbox;

    public OrderService(ApplicationDbContext context, IEventOutbox outbox)
    {
        _context = context;
        _outbox = outbox;
    }

    public async Task CreateOrderAsync(CreateOrderRequest request, CancellationToken ct)
    {
        // Begin transaction
        await using var transaction = await _context.Database.BeginTransactionAsync(ct);

        try
        {
            // Business operation
            var order = new Order { CustomerId = request.CustomerId, Total = request.Total };
            _context.Orders.Add(order);
            await _context.SaveChangesAsync(ct);

            // Store event in outbox (same transaction)
            await _outbox.StoreAsync(new OrderCreatedEvent
            {
                OrderId = order.Id,
                CustomerId = order.CustomerId
            }, ct);

            // Commit both business data and event atomically
            await transaction.CommitAsync(ct);
        }
        catch
        {
            // Rollback discards both business data AND the outbox event
            await transaction.RollbackAsync(ct);
            throw;
        }
    }
}

Configuration

RabbitMqTransportOptions

Routing Key Resolution

By default, the ITopicNameResolver returns the Type.Name of OutboxMessage.EventType (a System.Type), which is used as the routing key.

Override the resolver for custom naming strategies:

services.AddSingleton<ITopicNameResolver, MyCustomTopicNameResolver>();

services.AddPulse(config => config
    .UseRabbitMqTransport(options =>
    {
        options.ExchangeName = "events";
    }));

Delivery Guarantees

A publish counts as successful only after the broker has confirmed it. The outbox marks a message Completed only then. Otherwise SendAsync / SendBatchAsync throws and the message stays pending for retry.

• Publisher confirms: every channel is created with CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true). BasicPublishAsync waits for the broker's basic.ack and throws PublishException on a basic.nack or basic.return. See Consumer Acknowledgements and Publisher Confirms.
• Persistent messages: every message is published with delivery mode 2 (Persistent = true). In durable queues it survives a broker restart; the broker confirms it only after writing it to disk.
• Mandatory routing: every message is published with mandatory: true. If the exchange cannot route a message to any queue, the publish fails instead of the broker dropping the message.
• Missing exchange: the broker closes the channel with 404 NOT_FOUND, and the pending publish fails. The closed channel is discarded instead of being returned to the pool.

Trade-offs:

• Each SendAsync waits one broker round trip for its confirmation. SendBatchAsync starts all publishes of a batch before awaiting them, so a batch pays that latency roughly once.
• A routing key that matches no binding now fails. The message is retried and eventually dead-lettered by the outbox. Bind every published event type to a queue, or configure an alternate exchange as a catch-all.
• With batch sending enabled, a failed batch is retried as a whole. Every other message of the batch has already been published, so all confirmed messages are delivered again on each retry, up to the retry limit. One unroutable message is enough to trigger this. Consumers must be idempotent; de-duplicate on MessageId, which is the outbox message id.
• No outstanding-confirms rate limiter is configured on the channels, so SendBatchAsync keeps every message of a batch in flight at once. The number of in-flight publishes is bounded by OutboxProcessorOptions.BatchSize.
• With confirmation tracking on, RabbitMQ.Client adds the x-dotnet-pub-seq-no header to every message. Consumers see this header.

These settings are always on and cannot be configured.

Exchange Setup

Example: Topic Exchange

# Create a topic exchange for event routing
rabbitmqadmin declare exchange name=events type=topic durable=true

# Create queues and bind them to specific event types
rabbitmqadmin declare queue name=order-service durable=true
rabbitmqadmin declare binding source=events destination=order-service routing_key="OrderCreated"

rabbitmqadmin declare queue name=payment-service durable=true
rabbitmqadmin declare binding source=events destination=payment-service routing_key="PaymentProcessed"

Example: Fanout Exchange

# Create a fanout exchange for broadcasting to all subscribers
rabbitmqadmin declare exchange name=notifications type=fanout durable=true

# Create queues and bind them (no routing key needed for fanout)
rabbitmqadmin declare queue name=email-service durable=true
rabbitmqadmin declare binding source=notifications destination=email-service

rabbitmqadmin declare queue name=sms-service durable=true
rabbitmqadmin declare binding source=notifications destination=sms-service

Consumer Integration

Consume messages using the official RabbitMQ .NET client or any compatible library:

var factory = new ConnectionFactory { HostName = "localhost" };
await using var connection = await factory.CreateConnectionAsync();
await using var channel = await connection.CreateChannelAsync();

await channel.QueueDeclareAsync(
    queue: "order-service",
    durable: true,
    exclusive: false,
    autoDelete: false);

await channel.QueueBindAsync(
    queue: "order-service",
    exchange: "events",
    routingKey: "OrderCreated");

var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (sender, ea) =>
{
    var body = ea.Body.ToArray();
    var json = Encoding.UTF8.GetString(body);
    var @event = JsonSerializer.Deserialize<OrderCreatedEvent>(json);
    
    // Handle the event
    Console.WriteLine($"Order created: {@event.OrderId}");
    
    await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
};

await channel.BasicConsumeAsync(
    queue: "order-service",
    autoAck: false,
    consumer: consumer);

How It Works

1. Your application stores events in the outbox via IEventOutbox.StoreAsync within a database transaction.
2. The Pulse background processor polls the outbox for pending messages.
3. For each message, RabbitMqMessageTransport publishes it to the configured exchange with a routing key resolved by ITopicNameResolver.
4. RabbitMQ routes the message to bound queues based on the routing key and exchange type.
5. On success, the message is marked as processed; on failure, it remains pending for the next poll cycle.

Performance Considerations

Batch Processing

Configure batch size and polling interval based on your throughput requirements:

.AddOutbox(configureProcessorOptions: options =>
{
    options.BatchSize = 100;                         // Messages per poll cycle
    options.PollingInterval = TimeSpan.FromSeconds(1);
})

Connection Management

Register IConnection as a singleton in your DI container. The RabbitMQ client library is thread-safe and designed for concurrent use, so a single shared connection is recommended.

Channel Management

Channels are rented from a pool of up to MaxChannelPoolSize channels and reused for subsequent sends. If a channel is closed, for example by the broker after a failed publish, it is discarded and a new one is created on the next send.

Requirements

• .NET 8.0, .NET 9.0, or .NET 10.0
• RabbitMQ 3.8+ (or compatible AMQP 0-9-1 broker)
• RabbitMQ.Client 7.0+ for async API support
• Microsoft.Extensions.DependencyInjection for service registration
• Microsoft.Extensions.Hosting for the background processor

Related Packages

• NetEvolve.Pulse - Core mediator and outbox abstractions
• NetEvolve.Pulse.Extensibility - Core contracts and abstractions
• NetEvolve.Pulse.EntityFramework - Entity Framework Core persistence provider
• NetEvolve.Pulse.SqlServer - SQL Server ADO.NET persistence provider
• NetEvolve.Pulse.Polly - Polly v8 resilience policies integration

Documentation

For complete documentation, please visit the official documentation.

Contributing

Contributions are welcome! Please read the Contributing Guidelines before submitting a pull request.

Support

• Issues: Report bugs or request features on GitHub Issues
• Documentation: Read the full documentation at https://github.com/dailydevops/pulse

License

This project is licensed under the MIT License - see the LICENSE file for details.