AwadyLab.Mediator.RabbitMQ 1.7.0

AwadyLab.Mediator.RabbitMQ

RabbitMQ queue and distributed pub/sub broker transport for AwadyLab.Mediator.

Overview

AwadyLab.Mediator.RabbitMQ integrates RabbitMQ with AwadyLab.Mediator, supporting two distinct messaging topologies over a single shared connection:

flowchart TD
    Mediator{{"AwadyLab.Mediator"}}
    
    Mediator -->|"Delivery = Queue"| QMode["Queue Mode<br/>(RabbitMqMessageQueue)"]
    Mediator -->|"Delivery = Broker"| BMode["Broker Mode<br/>(RabbitMqBroker)"]
    
    QMode --> SingleQueue[("Durable Service Queue<br/>x-dead-letter-exchange")]
    BMode --> TopicEx["Topic / Fanout Exchange"]
    
    SingleQueue --> Pump["Internal Queue Pump"]
    TopicEx --> SubQ1[("Consumer Queue A")]
    TopicEx --> SubQ2[("Consumer Queue B")]
    
    SubQ1 --> H1{{"IExternalNotificationHandler&lt;&gt;"}}
    SubQ2 --> H2{{"IExternalNotificationHandler&lt;&gt;"}}

1. Queue Mode (NotificationDelivery.Queue): Replaces the default in-memory channel queue with a durable RabbitMQ queue (RabbitMqMessageQueue). Enqueued messages survive application restarts and are processed by local worker loops.
2. Broker Mode (NotificationDelivery.Broker): Full cross-service pub/sub messaging across independent microservices (RabbitMqBroker and RabbitMqSubscription) with automatic topology declaration, publisher confirms, and reconnect recovery.

You can enable Queue Mode, Broker Mode, or Both simultaneously over a single AMQP connection.

Features & Reliability

• Publisher Confirms: Enabled by default (PublisherConfirms = true). Publishes synchronously await RabbitMQ broker acknowledgment before returning.
• Channel Pooling: Publishes share a managed pool of channels (PublishChannelPoolSize = 8 by default) to eliminate channel churn and maximize throughput.
• Resilient Reconnections: Automatically detects network loss or broker restarts, restores connection, and recreates subscriptions without dropping workers.
• Header Sanitization: Prevents CRLF injection and Unicode directional override vulnerabilities across envelope headers.
• Dead-Letter Support: RabbitMQ rejected messages route to DeadLetterExchange, while exhausted retry attempts route to IDeadLetterSink.

Installation

dotnet add package AwadyLab.Mediator.RabbitMQ

Quickstart

1. Register in DI (Program.cs)

using AwadyLab.Mediator;
using AwadyLab.Mediator.RabbitMQ;
using AwadyLab.Mediator.RabbitMQ.Options;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddMediator([typeof(Program).Assembly], options =>
{
    options.UseRabbitMq(new RabbitMqOptions
    {
        ConnectionString = "amqp://guest:guest@localhost:5672/",
        
        // 1. Enable Broker Pub/Sub Mode
        Broker = new RabbitMqBrokerOptions
        {
            Prefetch = 20,
            DeadLetterExchange = "dlx.events"
        },
        
        // 2. (Optional) Enable Durable Queue Mode for local queue delivery
        Queue = new RabbitMqQueueOptions
        {
            QueueName = "orders-service-queue",
            Prefetch = 10
        }
    });
});

var app = builder.Build();

Broker Mode (Cross-Service Pub/Sub)

In Broker mode, events published with NotificationDelivery.Broker are serialized and broadcast to an exchange. Subscribing microservices consume events through their own durable queues.

1. Define Notification & Topology

Implement IExternalNotificationDefinition<T> to declare the exchange contract:

using AwadyLab.Mediator.Abstraction;
using AwadyLab.Mediator.Abstraction.Messaging;
using AwadyLab.Mediator.Models.Options;
using AwadyLab.Mediator.RabbitMQ;

// Domain event contract
public record OrderPlacedNotification(Guid OrderId, decimal Total) : INotification;

// Topology definition (exchange contract)
public sealed class OrderPlacedDefinition : IExternalNotificationDefinition<OrderPlacedNotification>
{
    public void Define(ExternalNotificationOptions options)
    {
        // Declares topic exchange and optional routing rules
        options.RabbitMq.Exchange = "orders.events";
        options.RabbitMq.ExchangeType = "topic";
        options.RabbitMq.RoutingKey = "orders.placed";
    }
}

2. Implement External Subscriber Handler

Consumers implement IExternalNotificationHandler<T> and specify a static QueueName:

using AwadyLab.Mediator.Abstraction.Messaging;
using AwadyLab.Mediator.Models.Options;

[ExternalSubscription(Prefetch = 25, MaxRetries = 3)]
public class BillingServiceHandler(IBillingService billing) 
    : IExternalNotificationHandler<OrderPlacedNotification>
{
    // Required: Identifies the subscription queue bound to the exchange
    public static string QueueName => "billing-service.orders";

    public async Task Handle(OrderPlacedNotification notification, CancellationToken ct)
    {
        await billing.GenerateInvoiceAsync(notification.OrderId, notification.Total, ct);
    }
}

3. Publishing to the Broker

await mediator.Publish(new OrderPlacedNotification(orderId, 250m), options =>
{
    options.Delivery = NotificationDelivery.Broker;
    
    // Optional: Set per-message RabbitMQ publish overrides
    options.RabbitMq.RoutingKey = "orders.placed.eu";
    options.RabbitMq.Priority = 2;
    options.RabbitMq.Expiration = TimeSpan.FromMinutes(30);
});

Queue Mode (Durable In-Process Channel)

When RabbitMqOptions.Queue is configured, NotificationDelivery.Queue routes envelopes through a durable RabbitMQ queue instead of an in-memory channel:

await mediator.Publish(new OrderPlacedNotification(orderId, 250m), options =>
{
    options.Delivery = NotificationDelivery.Queue;
});

• Survives service restarts.
• Drained by the mediator's background NotificationQueuePump workers with exponential backoff retries.

Options Reference

RabbitMqOptions

RabbitMqBrokerOptions

RabbitMqQueueOptions

License

Package by Mohamed Elawady

Copyright (c) Mohamed Elawady. All rights reserved.

This software is closed source and proprietary. Subject to the terms below,
you are granted a free, worldwide, non-exclusive license to:

  - Use this software, in source or compiled (NuGet package) form, in your
    own applications, including commercial applications, at no charge.

You may NOT:

  - Redistribute, sublicense, sell, or publish this software (or any
    modified version of it) as a standalone product, library, or package,
    including republishing it under a different name on NuGet.org or any
    other package repository.
  - Reverse engineer, decompile, or disassemble the compiled package except
    to the extent applicable law expressly permits this despite this
    limitation.
  - Remove or alter any copyright, trademark, or other proprietary notices.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.