EasyRabbitFlow 2.2.4

RabbitFlow Documentation

Welcome to the documentation for the RabbitFlow library! This guide will walk you through the process of configuring and using RabbitMQ consumers in your application using RabbitFlow.

Table of Contents

1. Introduction
2. Install
3. Configuration
   • Host Configuration
   • JSON Serialization Options
   • Publisher Options
4. Consumers
   • Adding Consumers
   • Retry Policies
   • Consumer Interface Implementation
5. Initialize Consumers
6. Publishing Messages
7. Queue State

1. Introduction

RabbitFlow is an intuitive library designed to simplify the management of RabbitMQ consumers within your application. This documentation provides step-by-step instructions for setting up, configuring, and using RabbitFlow effectively in your projects.

RabbitFlow is specifically designed to handle two approaches: pre-defined exchanges and queues within RabbitMQ, as well as the dynamic creation of new ones as needed. The rationale behind this approach is to offer flexibility in managing RabbitMQ infrastructure while maintaining simplicity in usage.

Now, let's dive into the details of setting up, configuring, and using RabbitFlow in your projects.

2. Install

To install the RabbitFlow library into your project, you can use the NuGet package manager:

dotnet add package EasyRabbitFlow

3. Configuration

To configure RabbitFlow in your application, use the AddRabbitFlow method:

builder.Services.AddRabbitFlow(opt =>
{
    // Configure host settings
    opt.ConfigureHost(hostSettings =>
    {
        hostSettings.Host = "rabbitmq.example.com";
        hostSettings.Username = "guest";
        hostSettings.Password = "guest";
    });

    // Configure JSON serialization options. [OPTIONAL]
    opt.ConfigureJsonSerializerOptions(jsonSettings =>
    {
        jsonSettings.PropertyNameCaseInsensitive = true;
        jsonSettings.PropertyNamingPolicy = JsonNamingPolicy.CamelCase;
        jsonSettings.DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull;
    });

    // Configure Publisher. [OPTIONAL]
    opt.ConfigurePublisher(publisherSettings => publisherSettings.DisposePublisherConnection = false);

    // Add and configure consumers
     settings.AddConsumer<TConsumer>(consumer=>{...});
});

• 3.1 Host Configuration

The ConfigureHost method allows you to specify the connection details for your RabbitMQ host:

opt.ConfigureHost(hostSettings =>
{
    hostSettings.Host = "rabbitmq.example.com";
    hostSettings.Username = "guest";
    hostSettings.Password = "guest";
});

• 3.2 JSON Serialization Options

This option allows you to globally configure how JSON serialization should be handled. This configuration is optional; if not provided, the default JsonSerializerOptions will be used.

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

• 3.3 Publisher Options

Optionally, you may configure the publisher that you intend to use by defining the 'DisposePublisherConnection' variable.

This variable determines whether the connection established by the publisher with RabbitMQ should be kept alive or terminated upon the completion of the process. The default value is false.

 opt.ConfigurePublisher(publisherSettings => publisherSettings.DisposePublisherConnection = true);

4. Consumers

• 4.1 Adding Consumers

Define and configure consumers for specific queues using the AddConsumer method:

All consumers must implement the interface IRabbitFlowConsumer<TEvent> where TEvent us the event or message model.

Using the AddConsumer method, all required services and configurations will be created and registered into the DI container, ready for use.

opt.AddConsumer<EmailConsumer>("email-queue", consumerSettings =>
{
    consumerSettings.PrefetchCount = 1;
    consumerSettings.Timeout = TimeSpan.FromMilliseconds(500);
    consumerSettings.AutoGenerate = true;
    consumerSettings.ConfigureAutoGenerate(opt =>
      {
     	opt.DurableQueue = true;
     	opt.DurableExchange = true;
     	opt.ExclusiveQueue = false;
     	opt.AutoDeleteQueue = false;
     	opt.GenerateDeadletterQueue = true;
     	opt.ExchangeType = ExchangeType.Direct;
    	// ... other settings ...
      });

4.2 Retry Policies

You can configure a retry policy to handle message processing failures effectively.

By default, all exceptions related to timeout issues will be automatically retried if the retry mechanism is enabled.

Additionally, you can customize the retry logic by defining your own rules for handling specific use cases using the TranscientException class from the EasyRabbitFlow.Exceptions namespace.

consumerSettings.ConfigureRetryPolicy(retryPolicy =>
{
    retryPolicy.MaxRetryCount = 3;
    retryPolicy.RetryInterval = 1000;
    retryPolicy.ExponentialBackoff = true;
    retryPolicy.ExponentialBackoffFactor = 2;
});

4.3 Consumer Interface Implementation

Consumers must implement the IRabbitFlowConsumer<TEvent> interface:

// Consumers must implement the IRabbitFlowConsumer<TEvent> interface:

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

// Example EmailConsumer

public class EmailConsumer : IRabbitFlowConsumer<EmailEvent>
{
    private readonly ILogger<EmailConsumer> _logger;

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

    public async Task HandleAsync(EmailEvent message, CancellationToken cancellationToken)
    {
        await Task.CompletedTask;

        _logger.LogInformation("New email event received. Event:{event}", JsonSerializer.Serialize(message));
    }
}

5. Initialize Consumers

Use the InitializeConsumer extension method to efficiently process messages using registered consumers.

This method is an extension of IServiceProvider. It is intended for use when the application's ServiceProvider is already built, ensuring no second container is created to handle incoming messages.

This method resolves all required services and configurations, starting the message listener for the queue.

You can configure whether the consumer will be active and whether a new instance of the consumer should be created for each message. This ensures isolated processing in a scoped environment or a single instance for all incoming messages. Use this feature according to your needs.

var app = builder.Build();

app.Services.InitializeConsumer<EmailEvent, EmailConsumer>();

app.Services.InitializeConsumer<WhatsAppEvent, WhatsAppConsumer>(opt =>
{
    opt.PerMessageInstance = true;
    opt.Active = false;
});

6. Publishing Messages

Publisher Interface

Use the IRabbitFlowPublisher interface to publish messages to a RabbitMQ:

JsonSerializerOptions can be overridden from global settings.

The publisherId parameter is intended to identify the connection created with RabbitMQ.

public interface IRabbitFlowPublisher
{
    Task<bool> PublishAsync<TEvent>(TEvent message, string exchangeName, string routingKey, string publisherId = "", JsonSerializerOptions? jsonOptions = null) where TEvent : class;

    Task<bool> PublishAsync<TEvent>(TEvent message, string queueName, string publisherId = "", JsonSerializerOptions? jsonOptions = null) where TEvent : class;
}

7. Queue State

The IRabbitFlowState interface allows you to access queue status information:

public interface IRabbitFlowState
{
    bool IsEmptyQueue(string queueName);
    uint GetQueueLength(string queueName);
    uint GetConsumersCount(string queueName);
    bool QueueHasConsumers(string queueName);
}