A new RabbitFlowMessageContext parameter was added to provide AMQP metadata (MessageId, CorrelationId, headers, delivery info) directly to each consumer.
How to migrate: Add the RabbitFlowMessageContext context parameter to every HandleAsync implementation. If you don't need the context, simply ignore it:
public Task HandleAsync(MyEvent message, RabbitFlowMessageContext context, CancellationToken ct)
{
// context is available but not required to use
return ProcessAsync(message, ct);
}
Other changes in v5.0.0:
PublishAsync / PublishBatchAsync now accept an optional correlationId parameter.
PublishAsync returns PublishResult (instead of bool). Use result.Success instead of the raw return value.
PublishBatchAsync is new — publish multiple messages atomically (Transactional) or individually confirmed ().
Confirm
ChannelMode was removed from PublisherOptions — it is now a per-call parameter on PublishBatchAsync (defaults to Transactional). Single-message publishes always use publisher confirms.
PublisherOptions.IdempotencyEnabled is new — auto-assigns a unique MessageId to every published message.
public class OrderCreatedEvent
{
public string OrderId { get; set; } = string.Empty;
public decimal Total { get; set; }
public DateTime CreatedAt { get; set; }
}
2. Implement a consumer:
public class OrderConsumer : IRabbitFlowConsumer<OrderCreatedEvent>
{
private readonly ILogger<OrderConsumer> _logger;
public OrderConsumer(ILogger<OrderConsumer> logger)
{
_logger = logger;
}
public async Task HandleAsync(OrderCreatedEvent message, RabbitFlowMessageContext context, CancellationToken cancellationToken)
{
_logger.LogInformation("Processing order {OrderId}, total: {Total}, correlationId: {CorrelationId}",
message.OrderId, message.Total, context.CorrelationId);
// Your business logic here: save to DB, send email, call API, etc.
await Task.CompletedTask;
}
}
The RabbitFlowMessageContext parameter provides AMQP metadata (MessageId, CorrelationId, headers, etc.) for the message being processed. See Message Context for details.
Consumers support full dependency injection — you can inject any service (scoped, transient, singleton):
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
Retry Policies
Configure how failed messages are retried:
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
});
Example: retry timeline with exponential backoff (factor=2, base=1000ms):
Inject IRabbitFlowPublisher to publish messages. All single-message publishes use publisher confirms — the await only completes after the broker confirms receipt.
Single Message
Returns a PublishResult with Success, Destination, RoutingKey, MessageId, TimestampUtc, and Error:
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 and correlation ID
public async Task BroadcastNotificationAsync(NotificationEvent notification, string correlationId)
{
var result = await _publisher.PublishAsync(
notification,
exchangeName: "notifications",
routingKey: "user.created",
correlationId: correlationId);
Console.WriteLine($"Published to {result.Destination} at {result.TimestampUtc}");
}
}
When enabled, every published message (single or batch) gets a unique MessageId set in BasicProperties.MessageId. The ID is also returned in PublishResult.MessageId or BatchPublishResult.MessageIds.
Correlation
Pass a correlationId when publishing to trace related messages end-to-end:
// Single message with correlation
await publisher.PublishAsync(order, "orders-queue", correlationId: "req-abc-123");
// Exchange publish with correlation
await publisher.PublishAsync(event, "notifications", routingKey: "new", correlationId: requestId);
// Batch — same correlationId shared across all messages
await publisher.PublishBatchAsync(events, "orders-queue", correlationId: batchId);
The correlationId is set on BasicProperties.CorrelationId and received by consumers via RabbitFlowMessageContext.CorrelationId.
Message Context
Every consumer receives a RabbitFlowMessageContext as a parameter of HandleAsync, providing access to AMQP metadata of the message being processed:
public class OrderConsumer : IRabbitFlowConsumer<OrderCreatedEvent>
{
public async Task HandleAsync(OrderCreatedEvent message, RabbitFlowMessageContext context, CancellationToken ct)
{
// Idempotency check using MessageId
if (context.MessageId != null && await _db.ExistsAsync(context.MessageId))
{
_logger.LogWarning("Duplicate message {Id}, skipping", context.MessageId);
return;
}
// Trace correlation across services
_logger.LogInformation("Processing order. CorrelationId={CorrelationId}", context.CorrelationId);
// Check if this is a redelivery
if (context.Redelivered)
{
_logger.LogWarning("Redelivered message, DeliveryTag={Tag}", context.DeliveryTag);
}
await ProcessOrderAsync(message, ct);
}
}
RabbitFlowMessageContext properties:
Property
Type
Description
MessageId
string?
Unique ID from BasicProperties.MessageId (set when IdempotencyEnabled = true)
CorrelationId
string?
Correlation ID from BasicProperties.CorrelationId (set via correlationId parameter)
Exchange
string?
Exchange that delivered the message (empty for direct queue publishes)
RoutingKey
string?
Routing key used when the message was published
Headers
IDictionary?
Custom AMQP headers from BasicProperties.Headers
DeliveryTag
ulong
Broker-assigned delivery tag for this message
Redelivered
bool
Whether the broker redelivered this message
Note:RabbitFlowMessageContext is immutable — all properties are read-only and populated automatically from BasicDeliverEventArgs before HandleAsync is invoked.
Queue State Inspection
Use IRabbitFlowState to query queue metadata at runtime:
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?
Queue Purging
Use IRabbitFlowPurger to remove all messages from queues:
// Purge a single queue
await purger.PurgeMessagesAsync("orders-queue");
// Purge multiple queues at once
await purger.PurgeMessagesAsync(new[] { "orders-queue", "emails-queue", "notifications-queue" });
Temporary Batch Processing
IRabbitFlowTemporary is designed for fire-and-forget batch workflows — process a collection of messages through RabbitMQ with automatic queue creation and cleanup.
Ideal for:
Background jobs (PDF generation, email sending, report calculation)
One-time batch processing (database cleanup, data migration)
By default, timeouts and RabbitFlowTransientException trigger retries. Throw RabbitFlowTransientException from your consumer to signal a retryable error:
using EasyRabbitFlow.Exceptions;
public class PaymentConsumer : IRabbitFlowConsumer<PaymentEvent>
{
public async Task HandleAsync(PaymentEvent message, RabbitFlowMessageContext context, 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 types:
Exception
Purpose
RabbitFlowTransientException
Signals a retryable error — triggers retry policy
RabbitFlowException
General library error
RabbitFlowOverRetriesException
Thrown internally when all retry attempts are exhausted
EasyRabbitFlow is designed for high-throughput scenarios:
Zero per-message reflection — consumer handlers are compiled via expression trees at startup, not resolved per message.
Connection pooling — publisher reuses a single connection by default.
Prefetch control — tune PrefetchCount for optimal throughput vs. memory usage.
Thread-safe channel operations — all channel I/O (ACK/NACK/Publish) is serialized via per-channel semaphores, preventing race conditions when PrefetchCount > 1.