Oragon.RabbitMQ 1.0.0

Oragon.RabbitMQ

Official Release

Others

Tech / Skill

Opinionated and Simplified Minimal APIs for Consuming Messages from RabbitMQ, Ensuring No Crucial Configurations Are Hidden.

What is Oragon.RabbitMQ?

Oragon.RabbitMQ is a Minimal API implementation for Consume RabbitMQ Queues.

Oragon.RabbitMQ provides everything you need to create resilient RabbitMQ consumers without the need to study numerous books and articles or introduce unknown risks to your environment.

Get Started

Add Consumer and Serializer packages

dotnet add package Oragon.RabbitMQ
dotnet add package Oragon.RabbitMQ.Serializer.SystemTextJson

Implement your service (with or without interface)

public class BusinessService
{
    public async Task DoSomethingAsync(BusinessCommandOrEvent commandOrEvent)
    {
        ... business core ...
    }
}

Configuring Dependency Injection

Basic Setup

var builder = WebApplication.CreateBuilder(args); //or Host.CreateApplicationBuilder(args);

.../*your dependencies configuration*/...

builder.AddRabbitMQConsumer();

builder.Services.AddSingleton<BusinessService>();

builder.Services.AddSingleton<IAMQPSerializer>(sp => new SystemTextJsonAMQPSerializer(new System.Text.Json.JsonSerializerOptions(System.Text.Json.JsonSerializerDefaults.General){ ... }));

Inject a valid IConnection on dependency injection

The consumer will use the dependency injection to get an valid instance of RabbitMQ.Client.IConnection. If you does not provider one, above you can see how to create a connection configuration.

...
builder.Services.AddSingleton<IConnectionFactory>(sp => new ConnectionFactory()
{
    Uri = new Uri("amqp://rabbitmq:5672"),
    DispatchConsumersAsync = true
});

builder.Services.AddSingleton(sp => sp.GetRequiredService<IConnectionFactory>().CreateConnectionAsync().GetAwaiter().GetResult());
...

If you are using .NET Aspire, you can use .NET Aspire with:

...
builder.AddRabbitMQClient("rabbitmq");
...

🎯 Map your Queue 🎯

To map your queue using this package, follow these steps:

1. Build the application: First, you need to build your application using the builder pattern. This initializes the application and prepares it for further configuration.

   var app = builder.Build();
   

2. Map the queue: Next, map your queue to a specific service and command/event. This step involves configuring how the service will handle incoming messages from the queue.

   app.MapQueue<BusinessService, BusinessCommandOrEvent>(config => config
       .WithDispatchInRootScope()  // Use for singleton service
       .WithDispatchInChildScope() // Use for scoped service
       .WithAdapter((svc, msg) => svc.DoSomethingAsync(msg)) // Define how the service handles the message
       .WithQueueName("events") // Set the queue name
       .WithPrefetchCount(System.Environment.ProcessorCount * 16 * 10) // Set the prefetch count
   );
   

   • WithDispatchInRootScope(): Use this for singleton services that should be instantiated once and shared across the application.
   • WithDispatchInChildScope(): Use this for scoped services that should be instantiated per request or per message.
   • WithAdapter((svc, msg) => svc.DoSomethingAsync(msg)): Define how the service handles the incoming message. This is where you specify the method to process the message.
   • WithQueueName("events"): Set the name of the queue from which the messages will be consumed.
   • WithPrefetchCount(System.Environment.ProcessorCount * 16 * 10): Set the prefetch count to control how many messages the consumer can fetch at a time. This can help optimize performance.

3. Run the application: Finally, run the application to start processing messages from the queue.

   app.Run();
   

Full Example

var builder = WebApplication.CreateBuilder(args);

builder.AddRabbitMQConsumer();

builder.Services.AddSingleton<BusinessService>();

builder.Services.AddSingleton<IAMQPSerializer>(sp => new SystemTextJsonAMQPSerializer(new System.Text.Json.JsonSerializerOptions(System.Text.Json.JsonSerializerDefaults.General){ ... }));

builder.Services.AddSingleton<IConnectionFactory>(sp => new ConnectionFactory(){ Uri = new Uri("amqp://rabbitmq:5672"), DispatchConsumersAsync = true });

builder.Services.AddSingleton(sp => sp.GetRequiredService<IConnectionFactory>().CreateConnectionAsync().GetAwaiter().GetResult());

var app = builder.Build();

app.MapQueue<BusinessService, BusinessCommandOrEvent>(config => config
    .WithDispatchInRootScope()  // ->  for singleton service
    .WithDispatchInChildScope() // ->  for scoped service
    .WithAdapter((svc, msg) => svc.DoSomethingAsync(msg))
    .WithQueueName("queue")
    .WithPrefetchCount(System.Environment.ProcessorCount * 16 * 10)
);

app.Run();

Concepts

Decoupling Business Logic from Infrastructure

This approach is designed to decouple RabbitMQ consumers from business logic, ensuring that business code remains unaware of the queue consumption context.

The result is incredibly simple, decoupled, agnostic, more reusable, and highly testable code.

Opinionated Design: Why?

This consumer is focused on creating a resilient consumer using manual acknowledgments.

• The flow produces a BasicReject without requeue for serialization failures (e.g., incorrectly formatted messages), you will use dead-lettering to ensure these messages are not lost.
• The flow produces a BasicNack with requeue for processing failures, allowing for message reprocessing.
• Minimal API design style made without reflection
• Extensible with support for custom serializers and encoders

RabbitMQ Tracing com OpenTelemetry

Full support for OpenTelemetry on publishing or consuming RabbitMQ messages.

<img src="./docs/playground.gif">

Refactored to use RabbitMQ.Client 7x (with IChannel instead IModel)

Stages and Requirements for Launch

• Migrate Demo to Library Project
• Core: Queue Consumer
• Core: Rpc Queue Consumer
• Core: Support Keyed Services
• Core: Support of new design of RabbitMQ.Client
• Create Samples
• Review All SuppressMessageAttribute
• Create Docs
• Benchmarks
• Automate Badges
• Add SonarCloud
• Code Coverage > 80%
• Add CI/CD
• Add Unit Tests
• Add Integrated Tests with TestContainers
• Test CI/CD Flow: MyGet Alpha Packages with Symbols
• Test CI/CD Flow: MyGet Packages without Symbols
• Test CI/CD Flow: Nuget Packages without Symbols