Name: EasyRabbitFlow
Author: Juan Carlos Torres Cuervo

EasyRabbitFlow v4.2.0 — nugetz

Feature	EasyRabbitFlow
Fluent, strongly-typed configuration	✅
Automatic queue / exchange / dead-letter generation	✅
Reflection-free per-message processing	✅
Configurable retry with exponential backoff	✅
Temporary batch processing with auto-cleanup	✅
Queue state & purge utilities	✅
Full DI integration (scoped/transient/singleton)	✅
Publisher confirms (single) & transactional batch	✅
Built-in idempotency support (auto `MessageId`)	✅
Rich `PublishResult` / `BatchPublishResult` types	✅
Thread-safe channel operations	✅
.NET Standard 2.1 (works with .NET 6, 7, 8, 9+)	✅

┌─────────────────────────────────────────────────────────────────────┐
│                         Your Application                            │
├────────────┬──────────────┬──────────────┬──────────────────────────┤
│ Publisher  │  Consumers   │   State /    │  Temporary Batch         │
│            │  (Hosted)    │   Purger     │  Processing              │
├────────────┴──────────────┴──────────────┴──────────────────────────┤
│                     EasyRabbitFlow Library                           │
│                                                                     │
│  ┌──────────────┐  ┌────────────────┐  ┌─────────────────────────┐ │
│  │ IRabbitFlow   │  │ ConsumerHosted │  │ IRabbitFlowTemporary    │ │
│  │ Publisher     │  │ Service        │  │ (batch processing)      │ │
│  └──────┬───────┘  └───────┬────────┘  └────────────┬────────────┘ │
│         │                  │                         │              │
│  ┌──────┴──────────────────┴─────────────────────────┴────────┐    │
│  │              RabbitMQ.Client (ConnectionFactory)            │    │
│  └────────────────────────────┬────────────────────────────────┘    │
└───────────────────────────────┼─────────────────────────────────────┘
                                │
                    ┌───────────┴───────────┐
                    │     RabbitMQ Broker    │
                    │                       │
                    │  ┌─────────────────┐  │
                    │  │   Exchanges     │  │
                    │  └────────┬────────┘  │
                    │           │            │
                    │  ┌────────▼────────┐  │
                    │  │     Queues      │  │
                    │  └────────┬────────┘  │
                    │           │            │
                    │  ┌────────▼────────┐  │
                    │  │  Dead-Letter    │  │
                    │  │    Queues       │  │
                    │  └─────────────────┘  │
                    └───────────────────────┘

  Publisher                    RabbitMQ                        Consumer
  ────────                    ────────                        ────────

  PublishAsync() ──────►  Exchange ──routing──► Queue ──────► HandleAsync()
                                                  │               │
                                                  │          (on failure)
                                                  │               │
                                                  │          Retry Policy
                                                  │          (exponential
                                                  │           backoff)
                                                  │               │
                                                  │          (exhausted)
                                                  │               │
                                                  └──► Dead-Letter Queue

dotnet add package EasyRabbitFlow

public class OrderCreatedEvent
{
    public string OrderId { get; set; } = string.Empty;
    public decimal Total { get; set; }
    public DateTime CreatedAt { get; set; }
}

public class OrderConsumer : IRabbitFlowConsumer<OrderCreatedEvent>
{
    private readonly ILogger<OrderConsumer> _logger;

    public OrderConsumer(ILogger<OrderConsumer> logger)
    {
        _logger = logger;
    }

    public async Task HandleAsync(OrderCreatedEvent message, CancellationToken cancellationToken)
    {
        _logger.LogInformation("Processing order {OrderId}, total: {Total}",
            message.OrderId, message.Total);

        // Your business logic here: save to DB, send email, call API, etc.
        await Task.CompletedTask;
    }
}

builder.Services
    .AddRabbitFlow(cfg =>
    {
        cfg.ConfigureHost(host =>
        {
            host.Host = "localhost";
            host.Port = 5672;
            host.Username = "guest";
            host.Password = "guest";
        });

        cfg.AddConsumer<OrderConsumer>("orders-queue", c =>
        {
            c.AutoGenerate = true;
            c.PrefetchCount = 10;
            c.Timeout = TimeSpan.FromSeconds(30);
            c.ConfigureRetryPolicy(r =>
            {
                r.MaxRetryCount = 3;
                r.ExponentialBackoff = true;
                r.ExponentialBackoffFactor = 2;
            });
        });
    })
    .UseRabbitFlowConsumers(); // Starts background consumer automatically

app.MapPost("/orders", async (OrderCreatedEvent order, IRabbitFlowPublisher publisher) =>
{
    var result = await publisher.PublishAsync(order, "orders-queue");
    return result.Success
        ? Results.Ok(new { result.Destination, result.MessageId, result.TimestampUtc })
        : Results.Problem(result.Error?.Message);
});

builder.Services.AddRabbitFlow(cfg =>
{
    cfg.ConfigureHost(...);                    // Connection settings
    cfg.ConfigureJsonSerializerOptions(...);   // Serialization (optional)
    cfg.ConfigurePublisher(...);               // Publisher behavior (optional)
    cfg.AddConsumer<T>(...);                   // Register consumers
});

cfg.ConfigureHost(host =>
{
    host.Host = "rabbitmq.example.com";
    host.Port = 5672;
    host.Username = "admin";
    host.Password = "secret";
    host.VirtualHost = "/";
    host.AutomaticRecoveryEnabled = true;
    host.NetworkRecoveryInterval = TimeSpan.FromSeconds(10);
    host.RequestedHeartbeat = TimeSpan.FromSeconds(30);
});

Property	Type	Default	Description
`Host`	string	`"localhost"`	RabbitMQ server hostname or IP
`Port`	int	`5672`	AMQP port
`Username`	string	`"guest"`	Authentication username
`Password`	string	`"guest"`	Authentication password
`VirtualHost`	string	`"/"`	RabbitMQ virtual host
`AutomaticRecoveryEnabled`	bool	`true`	Auto-reconnect after failures
`TopologyRecoveryEnabled`	bool	`true`	Auto-recover queues/exchanges after reconnect
`NetworkRecoveryInterval`	TimeSpan	`10s`	Wait time between recovery attempts
`RequestedHeartbeat`	TimeSpan	`30s`	Heartbeat interval for connection health

cfg.ConfigureJsonSerializerOptions(json =>
{
    json.PropertyNameCaseInsensitive = true;
    json.PropertyNamingPolicy = JsonNamingPolicy.CamelCase;
    json.DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull;
});

cfg.ConfigurePublisher(pub =>
{
    pub.DisposePublisherConnection = false; // Keep connection alive (default)
    pub.IdempotencyEnabled = true;          // Auto-assign MessageId to every message
});

Property	Type	Default	Description
`DisposePublisherConnection`	bool	`false`	Dispose connection after each publish
`IdempotencyEnabled`	bool	`false`	Auto-assign a unique `MessageId` to every published message for deduplication

public interface IRabbitFlowConsumer<TEvent>
{
    Task HandleAsync(TEvent message, CancellationToken cancellationToken);
}

public class EmailConsumer : IRabbitFlowConsumer<NotificationEvent>
{
    private readonly IEmailService _emailService;
    private readonly ILogger<EmailConsumer> _logger;

    public EmailConsumer(IEmailService emailService, ILogger<EmailConsumer> logger)
    {
        _emailService = emailService;
        _logger = logger;
    }

    public async Task HandleAsync(NotificationEvent message, CancellationToken cancellationToken)
    {
        _logger.LogInformation("Sending email to {Recipient}", message.Email);
        await _emailService.SendAsync(message.Email, message.Subject, message.Body, cancellationToken);
    }
}

cfg.AddConsumer<EmailConsumer>("email-queue", c =>
{
    c.Enable = true;                              // Enable/disable this consumer
    c.PrefetchCount = 5;                          // Messages fetched in parallel
    c.Timeout = TimeSpan.FromSeconds(30);         // Per-message processing timeout
    c.AutoAckOnError = false;                     // Don't ack failed messages
    c.ExtendDeadletterMessage = true;             // Add error details to DLQ messages
    c.ConsumerId = "email-consumer-1";            // Custom connection identifier
});

Property	Type	Default	Description
`Enable`	bool	`true`	Whether this consumer is active
`QueueName`	string	(set in constructor)	Queue to consume from
`ConsumerId`	string?	`null`	Custom connection ID (falls back to queue name)
`PrefetchCount`	ushort	`1`	How many messages to prefetch
`Timeout`	TimeSpan	`30s`	Processing timeout per message
`AutoAckOnError`	bool	`false`	Auto-acknowledge on error (message lost)
`AutoGenerate`	bool	`false`	Auto-create queue/exchange/DLQ
`ExtendDeadletterMessage`	bool	`false`	Enrich dead-letter messages with error details

cfg.AddConsumer<OrderConsumer>("orders-queue", c =>
{
    c.AutoGenerate = true;
    c.ConfigureAutoGenerate(ag =>
    {
        ag.ExchangeName = "orders-exchange";      // Custom exchange name
        ag.ExchangeType = ExchangeType.Fanout;    // Direct | Fanout | Topic | Headers
        ag.RoutingKey = "orders-routing-key";      // Routing key for binding
        ag.GenerateExchange = true;                // Create the exchange
        ag.GenerateDeadletterQueue = true;         // Create a dead-letter queue
        ag.DurableQueue = true;                    // Queue survives broker restart
        ag.DurableExchange = true;                 // Exchange survives broker restart
        ag.ExclusiveQueue = false;                 // Not limited to one connection
        ag.AutoDeleteQueue = false;                // Don't delete when last consumer disconnects
    });
});

                        ┌──────────────────────┐
                        │  orders-exchange      │
                        │  (fanout)             │
                        └──────────┬───────────┘
                                   │ routing-key
                        ┌──────────▼───────────┐
                        │  orders-queue         │──── args: x-dead-letter-exchange
                        │  (durable)            │         x-dead-letter-routing-key
                        └──────────┬───────────┘
                                   │ (on failure)
                  ┌────────────────▼─────────────────┐
                  │  orders-queue-deadletter-exchange │
                  │  (direct)                         │
                  └────────────────┬─────────────────┘
                                   │
                  ┌────────────────▼─────────────────┐
                  │  orders-queue-deadletter          │
                  └──────────────────────────────────┘

Property	Type	Default	Description
`GenerateExchange`	bool	`true`	Create the exchange
`GenerateDeadletterQueue`	bool	`true`	Create dead-letter queue and exchange
`ExchangeType`	ExchangeType	`Direct`	`Direct`, `Fanout`, `Topic`, or `Headers`
`ExchangeName`	string?	`null`	Custom name (defaults to `{queue}-exchange`)
`RoutingKey`	string?	`null`	Custom routing key (defaults to `{queue}-routing-key`)
`DurableExchange`	bool	`true`	Exchange survives broker restart
`DurableQueue`	bool	`true`	Queue survives broker restart
`ExclusiveQueue`	bool	`false`	Queue limited to declaring connection
`AutoDeleteQueue`	bool	`false`	Delete queue when last consumer disconnects
`Args`	IDictionary?	`null`	Additional RabbitMQ arguments

c.ConfigureRetryPolicy(r =>
{
    r.MaxRetryCount = 5;               // Number of attempts
    r.RetryInterval = 1000;            // Base delay in ms
    r.ExponentialBackoff = true;        // Enable exponential backoff
    r.ExponentialBackoffFactor = 2;     // Multiply delay by this factor
    r.MaxRetryDelay = 30_000;           // Cap delay at 30 seconds
});

Attempt 1 → fail → wait 1000ms
Attempt 2 → fail → wait 2000ms
Attempt 3 → fail → wait 4000ms
Attempt 4 → fail → wait 8000ms
Attempt 5 → fail → sent to dead-letter queue

Property	Type	Default	Description
`MaxRetryCount`	int	`1`	Total retry attempts
`RetryInterval`	int	`1000`	Base delay between retries (ms)
`ExponentialBackoff`	bool	`false`	Enable exponential backoff
`ExponentialBackoffFactor`	int	`1`	Multiplier for exponential growth
`MaxRetryDelay`	int	`60000`	Upper bound for delay (ms)

c.ConfigureCustomDeadletter(dl =>
{
    dl.DeadletterQueueName = "custom-errors-queue";
});

{
  "dateUtc": "2026-01-15T10:30:00Z",
  "messageType": "OrderCreatedEvent",
  "messageData": { "orderId": "ORD-123", "total": 99.99 },
  "exceptionType": "TimeoutException",
  "errorMessage": "The operation was canceled.",
  "stackTrace": "...",
  "source": "OrderService",
  "innerExceptions": []
}

public class OrderService
{
    private readonly IRabbitFlowPublisher _publisher;

    public OrderService(IRabbitFlowPublisher publisher)
    {
        _publisher = publisher;
    }

    // Publish directly to a queue
    public async Task CreateOrderAsync(OrderCreatedEvent order)
    {
        var result = await _publisher.PublishAsync(order, queueName: "orders-queue");

        if (!result.Success)
            throw new Exception($"Publish failed: {result.Error?.Message}");
    }

    // Publish to an exchange with routing key
    public async Task BroadcastNotificationAsync(NotificationEvent notification)
    {
        var result = await _publisher.PublishAsync(
            notification,
            exchangeName: "notifications",
            routingKey: "user.created");

        Console.WriteLine($"Published to {result.Destination} at {result.TimestampUtc}");
    }
}

// Publish to exchange (with routing key) — always uses publisher confirms
Task<PublishResult> PublishAsync<TEvent>(TEvent message, string exchangeName, string routingKey,
    string publisherId = "", JsonSerializerOptions? jsonOptions = null,
    CancellationToken cancellationToken = default) where TEvent : class;

// Publish directly to queue — always uses publisher confirms
Task<PublishResult> PublishAsync<TEvent>(TEvent message, string queueName,
    string publisherId = "", JsonSerializerOptions? jsonOptions = null,
    CancellationToken cancellationToken = default) where TEvent : class;

Property	Type	Description
`Success`	bool	Whether the broker confirmed the message
`MessageId`	string?	Unique ID (when `IdempotencyEnabled = true`)
`Destination`	string	Target exchange or queue name
`RoutingKey`	string	Routing key used (empty for queue publishes)
`TimestampUtc`	DateTime	When the publish was executed
`Error`	Exception?	The exception if `Success` is `false`

// Atomic batch (Transactional — default)
var result = await publisher.PublishBatchAsync(
    orders,
    exchangeName: "orders-exchange",
    routingKey: "new-order");

Console.WriteLine($"Batch: {result.MessageCount} messages, success={result.Success}");

// Non-atomic batch (Confirm mode — higher throughput)
var result = await publisher.PublishBatchAsync(
    orders,
    queueName: "orders-queue",
    channelMode: ChannelMode.Confirm);

// Batch to exchange
Task<BatchPublishResult> PublishBatchAsync<TEvent>(IReadOnlyList<TEvent> messages,
    string exchangeName, string routingKey,
    ChannelMode channelMode = ChannelMode.Transactional,
    string publisherId = "", JsonSerializerOptions? jsonOptions = null,
    CancellationToken cancellationToken = default) where TEvent : class;

// Batch to queue
Task<BatchPublishResult> PublishBatchAsync<TEvent>(IReadOnlyList<TEvent> messages,
    string queueName,
    ChannelMode channelMode = ChannelMode.Transactional,
    string publisherId = "", JsonSerializerOptions? jsonOptions = null,
    CancellationToken cancellationToken = default) where TEvent : class;

Property	Type	Description
`Success`	bool	Whether all messages were published
`MessageCount`	int	Number of messages in the batch
`MessageIds`	IReadOnlyList<string>	IDs per message (when `IdempotencyEnabled = true`)
`Destination`	string	Target exchange or queue name
`RoutingKey`	string	Routing key used
`ChannelMode`	ChannelMode	Mode used (`Transactional` or `Confirm`)
`TimestampUtc`	DateTime	When the batch was executed
`Error`	Exception?	The exception if `Success` is `false`

cfg.ConfigurePublisher(pub => pub.IdempotencyEnabled = true);

public class HealthCheckService
{
    private readonly IRabbitFlowState _state;

    public HealthCheckService(IRabbitFlowState state)
    {
        _state = state;
    }

    public async Task<object> GetQueueHealthAsync(string queueName)
    {
        return new
        {
            IsEmpty = await _state.IsEmptyQueueAsync(queueName),
            MessageCount = await _state.GetQueueLengthAsync(queueName),
            ConsumerCount = await _state.GetConsumersCountAsync(queueName),
            HasConsumers = await _state.HasConsumersAsync(queueName)
        };
    }
}

Method	Returns	Description
`IsEmptyQueueAsync(queueName)`	`Task<bool>`	Is the queue empty?
`GetQueueLengthAsync(queueName)`	`Task<uint>`	Number of messages in the queue
`GetConsumersCountAsync(queueName)`	`Task<uint>`	Number of active consumers
`HasConsumersAsync(queueName)`	`Task<bool>`	Does the queue have any consumers?

// Purge a single queue
await purger.PurgeMessagesAsync("orders-queue");

// Purge multiple queues at once
await purger.PurgeMessagesAsync(new[] { "orders-queue", "emails-queue", "notifications-queue" });

public class InvoiceService
{
    private readonly IRabbitFlowTemporary _temporary;

    public InvoiceService(IRabbitFlowTemporary temporary)
    {
        _temporary = temporary;
    }

    public async Task ProcessInvoiceBatchAsync(List<Invoice> invoices)
    {
        int processed = await _temporary.RunAsync(
            invoices,
            onMessageReceived: async (invoice, ct) =>
            {
                Console.WriteLine($"Processing invoice {invoice.Id}...");
                await Task.Delay(500, ct); // Simulate work
            },
            onCompleted: (total, errors) =>
            {
                Console.WriteLine($"Done! Processed: {total}, Errors: {errors}");
            },
            options: new RunTemporaryOptions
            {
                PrefetchCount = 10,
                Timeout = TimeSpan.FromSeconds(30),
                CorrelationId = Guid.NewGuid().ToString()
            });
    }
}

int processed = await _temporary.RunAsync<Invoice, InvoiceResult>(
    invoices,
    onMessageReceived: async (invoice, ct) =>
    {
        var result = await ProcessInvoiceAsync(invoice, ct);
        return new InvoiceResult { InvoiceId = invoice.Id, Status = "Completed" };
    },
    onCompletedAsync: async (count, results) =>
    {
        // results is a ConcurrentQueue<InvoiceResult> with all collected results
        Console.WriteLine($"Processed {count} invoices, collected {results.Count} results");
        await SaveResultsAsync(results);
    });

  Your Code                   RabbitMQ (Temporary)            Handler
  ─────────                   ──────────────────              ───────

  RunAsync(messages) ─────►  Create temp queue  ───────────►  onMessageReceived()
       │                     Publish all msgs                      │
       │                           │                               │
       │                     Consume & process  ◄──────────────────┘
       │                           │
       │                     All processed?
       │                           │ yes
       ◄──────────────────── Delete temp queue
       │                     Call onCompleted()
       │
  return count

Option	Type	Default	Description
`PrefetchCount`	ushort	`1`	Parallel message processing (>0)
`Timeout`	TimeSpan?	`null`	Per-message timeout
`QueuePrefixName`	string?	`null`	Custom prefix for the temp queue name
`CorrelationId`	string?	`Guid`	Correlation ID for tracing/logging

using EasyRabbitFlow.Exceptions;

public class PaymentConsumer : IRabbitFlowConsumer<PaymentEvent>
{
    public async Task HandleAsync(PaymentEvent message, CancellationToken cancellationToken)
    {
        try
        {
            await ProcessPaymentAsync(message);
        }
        catch (HttpRequestException ex) when (ex.StatusCode == HttpStatusCode.ServiceUnavailable)
        {
            // This will trigger the retry policy
            throw new RabbitFlowTransientException("Payment gateway temporarily unavailable", ex);
        }
        // Any other exception → no retry, sent to dead-letter queue
    }
}

Exception	Purpose
`RabbitFlowTransientException`	Signals a retryable error — triggers retry policy
`RabbitFlowException`	General library error
`RabbitFlowOverRetriesException`	Thrown internally when all retry attempts are exhausted

Interface	Lifetime	Description
`IRabbitFlowPublisher`	Singleton	Publish messages to exchanges or queues
`IRabbitFlowState`	Singleton	Query queue metadata
`IRabbitFlowTemporary`	Singleton	Temporary batch processing
`IRabbitFlowPurger`	Singleton	Purge queue messages
`ConsumerHostedService`	Hosted	Background consumer lifecycle (via `UseRabbitFlowConsumers`)

// Register all EasyRabbitFlow services
IServiceCollection AddRabbitFlow(this IServiceCollection services,
    Action<RabbitFlowConfigurator>? configurator = null);

// Start background consumer processing
IServiceCollection UseRabbitFlowConsumers(this IServiceCollection services);

Method	Description
`ConfigureHost(Action<HostSettings>)`	Set RabbitMQ connection details
`ConfigureJsonSerializerOptions(Action<JsonSerializerOptions>)`	Customize JSON serialization
`ConfigurePublisher(Action<PublisherOptions>?)`	Configure publisher behavior
`AddConsumer<TConsumer>(string queueName, Action<ConsumerSettings<TConsumer>>)`	Register a consumer

cfg.AddConsumer<MyConsumer>("high-volume-queue", c =>
{
    c.PrefetchCount = 50;                          // Process 50 messages concurrently
    c.Timeout = TimeSpan.FromSeconds(60);          // Generous timeout for heavy processing
    c.ConfigureRetryPolicy(r =>
    {
        r.MaxRetryCount = 3;
        r.RetryInterval = 500;
        r.ExponentialBackoff = true;
        r.ExponentialBackoffFactor = 2;
        r.MaxRetryDelay = 10_000;                  // Cap at 10 seconds
    });
});

cfg.ConfigurePublisher(pub =>
{
    pub.DisposePublisherConnection = false;         // Reuse connection
    pub.IdempotencyEnabled = true;                  // Auto-assign MessageId
});

dev.juanca/EasyRabbitFlowv4.2.0

Get Started

Readme

EasyRabbitFlow

Table of Contents

Why EasyRabbitFlow?

Architecture Overview

Message Flow

Getting Started

Installation

Quick Start

Configuration

Host Settings

JSON Serialization

Publisher Options

Consumers

Implementing a Consumer

Registering Consumers

Auto-Generate Topology

Retry Policies

Custom Dead-Letter Queues

Publishing Messages

Single Message

Batch Publishing

Idempotency

Queue State Inspection

Queue Purging

Temporary Batch Processing

Basic Usage

With Result Collection

How It Works

Transient Exceptions & Custom Retry Logic

Full API Reference

Registered Services

Extension Methods

RabbitFlowConfigurator Methods

Performance Notes

License