RabbitMQ.Services 2026.10.8.1

RabbitMQ.Services

A .NET library that simplifies integrating RabbitMQ into ASP.NET Core applications. It provides dependency injection extensions, a consumer hosted service, an outbox pattern implementation, and a URI-based endpoint configuration model.

Features

• URI-based RabbitMQ endpoint configuration
• Consumer hosted service with automatic retry and delay support
• Automatic reconnection of consumers after a broker restart
• Outbox pattern for reliable message delivery via Entity Framework Core
• DNS-based cluster resolution
• OpenTelemetry activity propagation

Requirements

• .NET 9 or .NET 10
• RabbitMQ.Client 7.x
• Microsoft.EntityFrameworkCore.Relational (for the outbox pattern)

Installation

dotnet add package RabbitMQ.Services

Getting Started

1. Register Core Services

Call AddRabbitMQ in your Program.cs or Startup.cs:

services.AddRabbitMQ();

This registers all internal infrastructure services (connection factory, endpoint parser, DNS resolver, etc.).

2. Consuming Messages

Define a Message

public class OrderCreatedMessage : BaseMessage
{
    public int OrderId { get; set; }
}

BaseMessage provides TraceId and SpanId properties used for distributed tracing.

Implement a Handler

Extend BaseMessageHandler<T> and override HandleAsync:

public class OrderCreatedHandler : BaseMessageHandler<OrderCreatedMessage>
{
    public override async Task HandleAsync(OrderCreatedMessage message, CancellationToken token)
    {
        // process the message
    }
}

Override HandleAsync(T message, int retryAttempt, CancellationToken token) or the full overload with delayAttempt if you need retry/delay context.

Tracing the Handler

BaseMessage carries the TraceId and SpanId of the publisher's Activity (they are stored by the outbox together with the message). To continue that trace in the handler, open the consumer span through IActivityBuilder:

public class OrderCreatedHandler(IActivityBuilder activityBuilder) : BaseMessageHandler<OrderCreatedMessage>
{
    public override async Task HandleAsync(OrderCreatedMessage message, CancellationToken token)
    {
        using var activity = activityBuilder.StartNewChildActivity<OrderCreatedHandler>(message);
        // process the message: spans opened here (HttpClient, database, ...) belong to the publisher's trace
    }
}

The span comes from the ActivitySource named ActivityBuilder.ActivitySourceName ("RabbitMQ.Services"), so subscribe your tracer to it, for example with OpenTelemetry:

services.AddOpenTelemetry().WithTracing(tracing => tracing.AddSource(ActivityBuilder.ActivitySourceName));

Without a listener the builder still starts a plain Activity (so Activity.Current is available for log correlation) and marks it recorded, so spans created under it by other instrumented libraries are sampled.

Register the Consumer Hosted Service

services.AddConsumerHostedService<OrderCreatedMessage, OrderCreatedHandler>(configuration);

Configuration (appsettings.json)

{
  "RabbitMQ": {
    "Consumers": {
      "Defaults": {
        "MaxRetryCount": 3
      },
      "OrderCreatedMessage": {
        "Url": "amqp://user:password@localhost/orders/order.created"
      }
    }
  }
}

The section name under Consumers must match the message type name (typeof(TMessage).Name).

You can also configure the consumer programmatically using the optional configureOptions action:

services.AddConsumerHostedService<OrderCreatedMessage, OrderCreatedHandler>(
    configuration,
    config =>
    {
        config.Url = "amqp://user:password@localhost/orders/order.created";
        config.MaxRetryCount = 5;
    });

Consumer Configuration Options

Delaying Message Processing

Throw DelayMessageException from a handler to reschedule the message:

throw new DelayMessageException(TimeSpan.FromSeconds(30), "Not ready yet");

Connection Recovery

The client's built-in automatic recovery is switched off; ConsumerHostedService supervises the consumer instead. It connects with retries on startup, and when the broker connection drops it stops the consumer and connects again, retrying every second until the broker is back or the application shuts down. Nothing has to be configured, and a broker restart no longer requires restarting the application.

Every step of that path is bounded, so a broker that accepts TCP but never answers cannot hang a consumer silently:

• A connect attempt is cancelled after 60 seconds. Consumers that share a ConnectionName share one attempt and wait for its result rather than for a lock, so one slow connect never queues the others.
• A start attempt (connect plus channel setup) is abandoned after 90 seconds and retried.
• While connected, the supervisor checks IsConnected every 30 seconds and reconnects if the connection or the channel is closed even though no event reported it.

IAsyncMessageConsumer<T> reports the drop through its ConnectionLost event (raised for both a connection and a channel shutdown) and never recovers on its own. Handlers of that event run on the client's event dispatcher and must return immediately. StartAsync and StopAsync take a CancellationToken; IsConnected is a snapshot of the connection and channel state.

3. Sending Messages (Outbox Pattern)

The outbox pattern stores messages in your database within the same transaction as your business data, guaranteeing at-least-once delivery.

Implement IOutboxDbContext

Your DbContext must implement IOutboxDbContext:

public class AppDbContext : DbContext, IOutboxDbContext
{
    // EF Core already provides Set<T>() and SaveChangesAsync
}

Add the Outbox Table

In OnModelCreating, call:

protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.AddOutboxEntities("outbox_messages");
    // optionally specify a schema: modelBuilder.AddOutboxEntities("outbox_messages", "myschema");
}

Register Outbox Services

services.AddOutboxServices<AppDbContext>(configuration);

Configuration (appsettings.json)

{
  "RabbitMQ": {
    "Outbox": {
      "ConnectionName": "my-service-producer",
      "Namespace": "my-service"
    }
  }
}

Send a Message

Inject IAsyncMessageSender and call SendMessageAsync:

public class OrderService(IAsyncMessageSender sender)
{
    public async Task CreateOrderAsync()
    {
        // ... business logic ...

        await sender.SendMessageAsync(
            "amqp://user:password@localhost/orders/order.created",
            new OrderCreatedMessage { OrderId = 42 });

        // Commit both business data and outbox message in one transaction
        await dbContext.SaveChangesAsync();
    }
}

Deliver Outbox Messages

Inject IMessageDeliveryService and call SendMessagesAsync periodically (e.g., from a background service or a scheduled job):

await deliveryService.SendMessagesAsync();

RabbitMQ Endpoint URI Format

Endpoints are specified as URIs with the amqp:// or rabbitmq:// scheme:

amqp://[user:password@]host[:port]/[vhost/]queue[?param=value&...]

Examples

amqp://localhost/myqueue
amqp://user:pass@localhost/myvhost/myqueue
amqp://localhost/myqueue?exchangetype=direct&route=my.routing.key
amqp://localhost/myqueue?consumers=4&prefetch=10
amqp://localhost/myqueue?temporary=true
amqp://localhost/myqueue?hosts=host1:5672,host2:5672

Credentials from the URI are masked when the endpoint is written to logs, e.g. amqp://***:***@localhost/myvhost/myqueue.

Query Parameters

License

This project is licensed under the MIT License.