MVFC.Messaging.RabbitMQ 3.0.2

MVFC.Messaging.RabbitMQ

🇧🇷 Leia em Português

A .NET messaging provider for RabbitMQ, built on top of MVFC.Messaging.Core. Provides RabbitMqPublisher<T> and RabbitMqConsumer<T> for publishing and consuming JSON-serialized messages in RabbitMQ queues with persistent delivery, QoS prefetch, and explicit acknowledgment.

Package

Package	Downloads
MVFC.Messaging.RabbitMQ

Installation

dotnet add package MVFC.Messaging.RabbitMQ

This package depends on MVFC.Messaging.Core (installed automatically) and RabbitMQ.Client.

Configuration

Connection String

Both RabbitMqPublisher<T> and RabbitMqConsumer<T> use an async factory pattern (CreateAsync) and accept a connection string (AMQP URI) and a queue name.

amqp://guest:guest@localhost:5672/
amqp://user:password@rabbitmq-host:5672/vhost

Queue Configuration

The queue is automatically declared during creation with the following settings:

Consumer QoS

The consumer sets a prefetch count of 10, which means the broker delivers up to 10 unacknowledged messages to the consumer at a time.

appsettings.json Example

{
  "RabbitMQ": {
    "ConnectionString": "amqp://guest:guest@localhost:5672/",
    "QueueName": "orders"
  }
}

var connectionString = builder.Configuration["RabbitMQ:ConnectionString"]!;
var queueName = builder.Configuration["RabbitMQ:QueueName"]!;

Usage

Async Factory Pattern

Unlike other providers, RabbitMQ uses an async factory method instead of a regular constructor, because establishing the connection and declaring the queue are async operations:

// ✅ Correct — use CreateAsync
await using var publisher = await RabbitMqPublisher<OrderCreated>.CreateAsync(connectionString, queueName);
await using var consumer = await RabbitMqConsumer<OrderCreated>.CreateAsync(connectionString, queueName);

// ❌ Wrong — constructor is private
// var publisher = new RabbitMqPublisher<OrderCreated>(...);

Publishing a Single Message

using MVFC.Messaging.RabbitMQ.Rabbit;

var connectionString = "amqp://guest:guest@localhost:5672/";
var queueName = "orders";

await using var publisher = await RabbitMqPublisher<OrderCreated>.CreateAsync(connectionString, queueName);

var order = new OrderCreated(1, "Keyboard", 149.90m);
await publisher.PublishAsync(order);

Messages are serialized to JSON, encoded as UTF-8, and published with Persistent = true to survive broker restarts.

Publishing a Batch

Batch publishing sends each message sequentially using the same channel:

var orders = new[]
{
    new OrderCreated(1, "Keyboard", 149.90m),
    new OrderCreated(2, "Mouse", 59.90m),
    new OrderCreated(3, "Monitor", 899.00m)
};

await publisher.PublishBatchAsync(orders);

Consuming Messages

The consumer uses AsyncEventingBasicConsumer for event-driven message delivery with explicit acknowledgment:

using MVFC.Messaging.RabbitMQ.Rabbit;

var connectionString = "amqp://guest:guest@localhost:5672/";
var queueName = "orders";

await using var consumer = await RabbitMqConsumer<OrderCreated>.CreateAsync(connectionString, queueName);

await consumer.StartAsync(async (message, ct) =>
{
    Console.WriteLine($"Processing order #{message.OrderId}: {message.Product}");
    // Your business logic here
}, cancellationToken);

// ... later, when shutting down:
await consumer.StopAsync();

Consumer behavior:

• Uses AsyncEventingBasicConsumer for fully asynchronous message handling.
• autoAck is set to false — messages must be explicitly acknowledged.
• On successful handler execution, BasicAckAsync is called — the message is removed from the queue.
• On handler exception, BasicNackAsync is called with requeue: true — the message returns to the queue for redelivery.
• QoS prefetch count is 10 — the broker delivers up to 10 messages at a time before waiting for acks.

Complete Publish + Consume Example

using MVFC.Messaging.RabbitMQ.Rabbit;

var connectionString = "amqp://guest:guest@localhost:5672/";
var queueName = "orders";

await using var publisher = await RabbitMqPublisher<OrderCreated>.CreateAsync(connectionString, queueName);
await using var consumer = await RabbitMqConsumer<OrderCreated>.CreateAsync(connectionString, queueName);

// Start consuming
var received = new TaskCompletionSource<OrderCreated>();
await consumer.StartAsync(async (msg, ct) =>
{
    Console.WriteLine($"Received: Order #{msg.OrderId} — {msg.Product}");
    received.SetResult(msg);
}, CancellationToken.None);

await Task.Delay(1000); // Wait for consumer initialization

// Publish
await publisher.PublishAsync(new OrderCreated(42, "Keyboard", 149.90m));

// Wait for the message to be consumed
var result = await received.Task.WaitAsync(TimeSpan.FromSeconds(10));

// Cleanup
await consumer.StopAsync();

API Reference

RabbitMqPublisher<T>

RabbitMqConsumer<T>

Requirements

• .NET 9.0+
• RabbitMQ.Client (installed automatically)
• A running RabbitMQ server (or Docker: docker run -p 5672:5672 -p 15672:15672 rabbitmq:management)

License

Apache-2.0