ToolsToLive.RabbitMQ 10.0.0

ToolsToLive.RabbitMQ

A .NET 10 library for routed messages, delayed delivery and publish/subscribe events over RabbitMQ. Messages are serialized as JSON, with enum values written as strings.

Versioning

Starting with 10.0.0, versions follow the same scheme as ToolsToLive.Logging: the first number is the supported .NET version, the second is the library major version, and the third is the library minor version.

Configuration

appsettings.json:

{
  "RabbitMqSettings": {
    "Host": "<rabbitmq-host>",
    "Port": 5672,
    "UserName": "<user>",
    "Password": "<password>",
    "AppName": "sample",
    "AppVersion": "v1",
    "UseBuiltInRecovery": true
  }
}

Alternatively, replace Host, Port, UserName and Password with a connection string. URI-encode special characters in its credentials:

{
  "RabbitMqSettings": {
    "ConnectionString": "amqp://<user>:<password>@<rabbitmq-host>:5672/",
    "AppName": "sample",
    "AppVersion": "v1",
    "UseBuiltInRecovery": true
  }
}

See RabbitMqSettings.cs for all settings, including TLS and delay durations. Supply real credentials through your application's configuration providers rather than committing them.

Startup and routes

ASP.NET Core Program.cs:

using Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.DependencyInjection;
using ToolsToLive.RabbitMQ;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddRabbitMqMessageHandler(
    builder.Configuration.GetSection("RabbitMqSettings"));
builder.Services.AddHostedService<MessageConsumer>();

var app = builder.Build();
await app.Services.ConfigureRabbitMQAsync<Routes>();
await app.RunAsync();

ConfigureRabbitMQAsync<TRoute> creates the exchanges, delay queues and one queue for each value of the route enum. Call it before sending messages or starting listeners; the broker must be available. EventBus uses the same exchange, so it also needs this initialization. An EventBus-only application can use an empty route enum.

Shared contracts used by the examples:

public enum Routes
{
    SendModel
}

public sealed class Model
{
    public string SomeField { get; set; } = "";
}

public sealed class SomethingCreatedEvent
{
    public int Id { get; set; }
}

With the configuration above, the exchange is sample-v1-exchange and the route queue is sample-v1-send-model-queue. Producers and consumers must agree on route names and message contracts.

Sending messages

Inject IMessageSender into an application service:

using ToolsToLive.RabbitMQ.Interfaces;
using ToolsToLive.RabbitMQ.Model;

public sealed class MessagePublisher(IMessageSender sender)
{
    public Task SendAsync(CancellationToken cancellationToken = default)
    {
        var message = new Model { SomeField = "Some content" };
        return sender.SendAsync(message, Routes.SendModel, cancellationToken);
    }

    public Task SendLaterAsync(CancellationToken cancellationToken = default)
    {
        var message = new Model { SomeField = "Some content" };
        return sender.SendAsync(message, Routes.SendModel, SendingDelay.Delay4,
            cancellationToken);
    }
}

Delay4 defaults to five minutes and is configured by RabbitMqSettings.Delay4. Delayed messages pass through TTL/dead-letter queues; no delayed-message plugin is required. Successful sending does not mean the consumer has finished processing.

Receiving messages

The hosted service registered at startup keeps its listener until the application stops:

using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using ToolsToLive.RabbitMQ.Interfaces;

public sealed class MessageConsumer(
    IMessageListenerFactory listenerFactory,
    ILogger<MessageConsumer> logger) : IHostedService
{
    private IMessageListener? _listener;

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        _listener = await listenerFactory.CreateAndStartListeningAsync<Routes, Model>(
            Routes.SendModel, HandleAsync, prefetchCount: 5,
            cancellationToken: cancellationToken);
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        if (_listener is not null)
            await _listener.DisposeAsync();
    }

    private Task HandleAsync(Model message)
    {
        logger.LogInformation("Received {SomeField}", message.SomeField);
        return Task.CompletedTask;
    }
}

prefetchCount limits unacknowledged messages per listener; it does not set processing concurrency. The callback is Func<TMessage, Task> and receives no cancellation token. Resolve scoped application services in a separate DI scope per message.

Deserialization and handler exceptions are logged, and the message is then acknowledged. Throwing from the handler does not trigger redelivery; implement retries in the application if needed. Handlers should also tolerate duplicate deliveries, for example after a connection failure before acknowledgement.

EventBus

Inject IEventBus to publish events:

using ToolsToLive.RabbitMQ.Interfaces;

public sealed class EventPublisher(IEventBus eventBus)
{
    public Task PublishAsync(int id, CancellationToken cancellationToken = default)
    {
        return eventBus.PublishAsync("something_created",
            new SomethingCreatedEvent { Id = id }, cancellationToken);
    }
}

To receive events, add builder.Services.AddHostedService<EventConsumer>(); before builder.Build():

using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using ToolsToLive.RabbitMQ.Interfaces;
using ToolsToLive.RabbitMQ.Model;

public sealed class EventConsumer(
    IEventBus eventBus,
    ILogger<EventConsumer> logger) : IHostedService
{
    private Subscription? _subscription;

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        _subscription = await eventBus.SubscribeAsync<SomethingCreatedEvent>(
            "something_created", "app1", HandleAsync, prefetchCount: 1,
            cancellationToken: cancellationToken);
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        if (_subscription is not null)
            await _subscription.DisposeAsync();
    }

    private Task HandleAsync(SomethingCreatedEvent message)
    {
        logger.LogInformation("Something created: {Id}", message.Id);
        return Task.CompletedTask;
    }
}

SubscribeAsync creates and binds a durable queue. Different appKey values, such as app1 and app2, get separate queues and each receives a copy of the event. Subscribers with the same key share a queue and compete for its messages. Keep event names and keys consistent in spelling and case.

Keep the subscription alive while receiving events. DisposeAsync (or UnsubscribeAsync) stops listening and keeps the queue and pending messages for the next start. DeleteQueueAsync(eventType, appKey) deletes the queue and its messages; use it only when permanently removing a subscription. Only queues already bound at publication receive an event.

Health checks and tracing

After registering the library, add a health check with builder.Services.AddHealthChecks().AddRabbitMQ(); (namespace ToolsToLive.RabbitMQ.HealthCheck). It checks the connection by opening a channel; it does not verify message processing.

For OpenTelemetry tracing, include ToolsToLive.RabbitMQ.Observability.Tracing.ActivitySourceName in your tracer provider's AddSource(...) configuration. The library emits producer/consumer activities and propagates trace context through message headers.