Commify.Titan.API.RabbitMQ 0.13.0

Commify.Titan.API.RabbitMQ

Self-contained NuGet package that adds RabbitMQ messaging to any .NET 10 application. Supports event publishing/consuming with zero configuration, custom message routing, outbox pattern with PostgreSQL, and dead letter queues.

Contact: Titan API Team

Configuration

Add the following to appsettings.json:

{
  "TitanCommunication": {
    "RabbitMQ": {
      "Host": "host",
      "Port": 5672,
      "BasicCredentials": {
        "Username": "username",
        "Password": "password"
      }
    }
  }
}

• BasicCredentials enables basic authentication. When omitted, the package falls back to AAD token authentication.

Environment variables

Initialization

Register the Titan framework, then initialize RabbitMQ connections at startup:

var builder = Host.CreateApplicationBuilder(args);

builder.Services
    .AddTitanFramework(builder.Configuration);

// Register publishers and consumers (see sections below)

var app = builder.Build();

await app.InitializeAsync();                          // opens RabbitMQ connections
await app.ApplyEventPublisherMigrations();           // applies outbox EF Core migrations (only if using outbox)

await app.RunAsync();

Publishing

Event publishing

RegisterEventPublisher uses the shared event defaults (RabbitMqEventDefaults) and the outbox pattern. No manual settings required.

services.RegisterEventPublisher<LoginEventData>(
    context.Configuration,
    context.Configuration.GetConnectionString("AppDb")!);

Inject IOutboxMessagePublisher<T> and call SaveChangesAsync to transactionally save the message alongside your domain changes:

public class LoginService(
    IOutboxMessagePublisher<LoginEventData> publisher,
    IDbContextFactory<AppDbContext> dbContextFactory)
{
    public async Task HandleLoginAsync(LoginEventData data, CancellationToken ct)
    {
        var dbContext = await dbContextFactory.CreateDbContextAsync(ct);

        QueueMessage<LoginEventData> message = new()
        {
            Id = Guid.NewGuid(),
            Headers = new QueueHeaders
            {
                CorrelationId = Guid.NewGuid(),
                Source = "LoginService",
                Type = "LoginEventData",
                OccurredAt = DateTimeOffset.UtcNow
            },
            Payload = data
        };

        await publisher.SaveChangesAsync(dbContext, message, ct);
    }
}

A background service polls the outbox and publishes pending messages with exponential backoff.

Message publishing

RegisterMessagePublisher registers an IMessagePublisher<T> with custom exchange settings. Available in two overloads:

// Without IServiceProvider
services.RegisterMessagePublisher<Message>(context.Configuration, ()
    => new()
    {
        VirtualHost = "API",
        ExchangeName = "ex.mock",
        ExchangeType = "fanout"
    });

// With IServiceProvider
services.RegisterMessagePublisher<Message>(context.Configuration, (sp)
    => new()
    {
        VirtualHost = "API",
        ExchangeName = "ex.mock",
        ExchangeType = "fanout"
    });

Inject IMessagePublisher<T> and call PublishAsync:

await messagePublisher.PublishAsync(queueMessage, customTrace, cancellationToken);

Outbox message publishing

RegisterMessagePublisherWithOutbox combines custom exchange settings with the PostgreSQL outbox:

// Without IServiceProvider
services.RegisterMessagePublisherWithOutbox<Message>(
    context.Configuration,
    dbConnectionString,
    () => new()
    {
        VirtualHost = "API",
        ExchangeName = "ex.custom",
        ExchangeType = "direct",
        RoutingKey = "my.routing.key"
    });

// With IServiceProvider
services.RegisterMessagePublisherWithOutbox<Message>(
    context.Configuration,
    dbConnectionString,
    (sp) => new()
    {
        VirtualHost = "API",
        ExchangeName = "ex.custom",
        ExchangeType = "direct",
        RoutingKey = "my.routing.key"
    });

Inject IOutboxMessagePublisher<T> and use SaveChangesAsync.

Consuming

Event consuming

RegisterEventConsumer uses the shared event defaults (RabbitMqEventDefaults). No manual settings required.

services.RegisterEventConsumer<LoginEventData, LoginEventHandler>(context.Configuration);

Implement IMessageConsumer<T>:

public class LoginEventHandler : IMessageConsumer<LoginEventData>
{
    public async Task HandleMessageAsync(
        QueueMessage<LoginEventData> message,
        RabbitMQConsumerTrace customTrace,
        CancellationToken cancellationToken)
    {
        // Process the message
    }

    public async Task HandleExceptionAsync(
        Exception exception,
        RabbitMQConsumerTrace customTrace,
        CancellationToken cancellationToken)
    {
        // Handle consumer errors
    }
}

Message consuming

RegisterMessageConsumer requires custom settings. Available in two overloads:

// Without IServiceProvider
services.RegisterMessageConsumer<LoginEventData, LoginEventHandler>(context.Configuration, ()
    => new()
    {
        VirtualHost = "API",
        ExchangeNameToBind = "ex.events",
        ExchangeType = "fanout",
        QueueName = "q.my-service.login-events",   // optional, auto-generated from K8S_POD_NAME if omitted
        RoutingKey = "login.#"                       // optional
    });

// With IServiceProvider
services.RegisterMessageConsumer<LoginEventData, LoginEventHandler>(context.Configuration, (sp)
    => new()
    {
        VirtualHost = "API",
        ExchangeNameToBind = "ex.events",
        ExchangeType = "fanout"
    });

When QueueName is omitted, a queue is auto-created using K8S_POD_NAME and deleted when the application stops.

Dead letter queues

Failed messages can be routed to a dead letter queue by setting three properties on RabbitMqConsumerSettings:

All three must be set, or none. Partial configuration throws an ArgumentNullException at startup.

services.RegisterMessageConsumer<LoginEventData, LoginEventHandler>(context.Configuration, ()
    => new()
    {
        VirtualHost = "API",
        ExchangeNameToBind = "ex.events",
        ExchangeType = "fanout",
        DeadLetterQueueName = "q.dlx.login-events",
        DeadLetterExchangeName = "ex.dlx.events",
        DeadLetterRoutingKey = "dlx.login-events"
    });

When configured, any unhandled exception in HandleMessageAsync causes the message to be published to the dead letter exchange with the original headers plus a DeadLetterReason header containing the exception message. Dead letter exchanges always use the topic exchange type.

Custom implementation

For full control over publishing and consuming, use RegisterRabbitMQMessaging to register only the connection management layer:

services.RegisterRabbitMQMessaging(context.Configuration);

This registers a singleton IConnectionManagement that you can inject to create channels directly:

var channel = await connectionManagement.CreateChannelAsync("API", customTrace, cancellationToken);

Key types