RabbitMQ.Client.Core.DependencyInjection

This repository contains the library that provides functionality for wrapping RabbitMQ.Client code and adding it in your application via the dependency injection mechanism. That wrapper provides easy, managed message queue consume and publish operations. The library targets netstandard2.1.

(!)Current version of documentation is out of date for 5.0.0. Stay updated.(!)

Usage

This section contains only an example of a basic usage of the library. You can find the detailed documentation in the docs directory where all functionality fully covered.

Producer

To produce messages in the RabbitMQ queue you have to go through the routine of configuring a RabbitMQ connection and exchanges. In your Startup file you can do it simply calling couple methods in a fluent-Api way.

public static IConfiguration Configuration { get; set; }

public void ConfigureServices(IServiceCollection services)
{
    var rabbitMqSection = Configuration.GetSection("RabbitMq");
    var exchangeSection = Configuration.GetSection("RabbitMqExchange");

    services.AddRabbitMqClient(rabbitMqSection)
        .AddProductionExchange("exchange.name", exchangeSection);
}

By calling AddRabbitMqClient you add IQueueService that provides functionality of sending messages to queues. AddProductionExchange configures exchange to queues bindings (presented in json configuration) that allow messages to route properly. Example of appsettings.json is two sections below. You can also configure everything manually. For more information, see rabbit-configuration and exchange-configuration documentation files.

Now you can inject an instance of IQueueService inside anything you want.

[Route("api/[controller]")]
public class HomeController : Controller
{
    readonly IQueueService _queueService;
    public HomeController(IQueueService queueService)
    {
        _queueService = queueService;
    }
}

Now you can send messages using Send or SendAsync methods.

var messageObject = new
{
    Id = 1,
    Name = "RandomName"
};

await _queueService.SendAsync(
    @object: messageObject,
    exchangeName: "exchange.name",
    routingKey: "routing.key");

You can also send delayed messages.

await _queueService.SendAsync(
    @object: messageObject,
    exchangeName: "exchange.name",
    routingKey: "routing.key",
    millisecondsDelay: 10);

The mechanism of sending delayed messages described in the documentation. Dive into it for more detailed information.

Consumer

After making the message production possible let's make the consumption possible too! Imagine that a consumer is a simple console application.

class Program
{
    const string ExchangeName = "exchange.name";
    public static IConfiguration Configuration { get; set; }

    static void Main()
    {
        var builder = new ConfigurationBuilder()
            .SetBasePath(Directory.GetCurrentDirectory())
            .AddJsonFile("appsettings.json", optional: false, reloadOnChange: true);
        Configuration = builder.Build();

        var serviceCollection = new ServiceCollection();
        ConfigureServices(serviceCollection);

        var serviceProvider = serviceCollection.BuildServiceProvider();
        var queueService = serviceProvider.GetRequiredService<IQueueService>();
        queueService.StartConsuming();
    }

    static void ConfigureServices(IServiceCollection services)
    {
        var rabbitMqSection = Configuration.GetSection("RabbitMq");
        var exchangeSection = Configuration.GetSection("RabbitMqExchange");

        services.AddRabbitMqClient(rabbitMqSection)
            .AddConsumptionExchange("exchange.name", exchangeSection)
            .AddMessageHandlerSingleton<CustomMessageHandler>("routing.key");
    }
}

You have to configure everything almost the same way as you have already done previously with the producer. The main differences are that you need to declare (configure) a consumption exchange calling AddConsumptionExchange instead of a production exchange. For detailed information about difference in exchange declarations you may want to see the documentation. The other important part is adding custom message handlers by implementing the IMessageHandler interface and calling AddMessageHandlerSingleton<T> or AddMessageHandlerTransient<T> methods. IMessageHandler is a simple subscriber, which receives messages from a queue by selected routing key. You are allowed to set multiple message handlers for one routing key (e.g. one is writing it in a database, and the other does the business logic).

You can also use pattern matching while adding message handlers where * (star) can substitute for exactly one word and # (hash) can substitute for zero or more words. You are also allowed to specify the exact exchange which will be "listened" by the selected message handler with the given routing key (or a pattern).

services.AddRabbitMqClient(rabbitMqSection)
    .AddConsumptionExchange("exchange.name", exchangeSection)
    .AddMessageHandlerSingleton<CustomMessageHandler>("*.*.key")
    .AddMessageHandlerSingleton<AnotherCustomMessageHandler>("#", "exchange.name");

The very last step is to start "listening" (consuming) by simply calling StartConsuming method of IQueueService. After that you will start getting messages, and you can handle them in any way you want.

A message handler example.

public class CustomMessageHandler : IMessageHandler
{
    readonly ILogger<CustomMessageHandler> _logger;
    public CustomMessageHandler(ILogger<CustomMessageHandler> logger)
    {
        _logger = logger;
    }

    public void Handle(BasicDeliverEventArgs eventArgs, string matchingRoute)
    {
        // Do whatever you want with the message.
        _logger.LogInformation("Hello world");
    }
}

If you want to use an async magic then implement your custom IAsyncMessageHandler.

public class CustomAsyncMessageHandler : IAsyncMessageHandler
{
    readonly ILogger<CustomAsyncMessageHandler> _logger;
    public CustomAsyncMessageHandler(ILogger<CustomAsyncMessageHandler> logger)
    {
        _logger = logger;
    }

    public async Task Handle(BasicDeliverEventArgs eventArgs, string matchingRoute)
    {
	   // await something.
    }
}

If you want to send messages from the Handle method inside your message handler then use INonCyclicMessageHandler. You can't inject IQueueService inside any message handlers, otherwise you will get a cyclic dependency exception. But INonCyclicMessageHandler allows you to avoid this as it accepts an instance of IQueueService as a parameter of the method Handle.

public class MyNonCyclicMessageHandler : INonCyclicMessageHandler
{
    // Inject anything you want except IQueueService.
    readonly ILogger<MyNonCyclicMessageHandler> _logger;
    public MyNonCyclicMessageHandler(ILogger<MyNonCyclicMessageHandler> logger)
    {
        _logger = logger;
    }

    public void Handle(BasicDeliverEventArgs eventArgs, string matchingRoute, IQueueService queueService)
    {
        // Send anything you want using IQueueService instance.
        var anotherMessage = new MyMessage { Foo = "Bar" };
        queueService.Send(anotherMessage, "exchange.name", "routing.key");
    }
}

INonCyclicMessageHandler has its asynchronous analogue IAsyncNonCyclicMessageHandler.

public class MyAsyncNonCyclicMessageHandler : IAsyncNonCyclicMessageHandler
{
    // Inject anything you want except IQueueService.
    readonly ILogger<MyAsyncNonCyclicMessageHandler> _logger;
    public MyAsyncNonCyclicMessageHandler(ILogger<MyAsyncNonCyclicMessageHandler> logger)
    {
        _logger = logger;
    }

    public async Task Handle(BasicDeliverEventArgs eventArgs, string matchingRoute, IQueueService queueService)
    {
        // Do async stuff.
        var anotherMessage = new MyMessage { Foo = "Bar" };
        await queueService.SendAsync(anotherMessage, "exchange.name", "routing.key");
    }
}

You can register any of those message handlers the same way you registered IMessageHandler before. There are similar extension methods for each type of message handlers. For more information, see the message-consuming documentation file.

You can also find example projects in the repository inside the examples directory.

Configuration

In both cases for producing and consuming messages the configuration file is the same. appsettings.json consists of those sections: (1) settings to connect to the RabbitMQ server and (2) sections that configure exchanges and queue bindings. You can have multiple exchanges and one configuration section per each exchange. Exchange sections define how to bind queues and exchanges with each other using specified routing keys (or patterns). You allowed to bind a queue to an exchange with more than one routing key, but if there are no routing keys in the queue section, then that queue will be bound to the exchange with its name.

{
  "RabbitMq": {
    "HostName": "127.0.0.1",
    "Port": "5672",
    "UserName": "guest",
    "Password": "guest"
  },
  "RabbitMqExchan

RabbitMQ.Client.Core.DependencyInjection

Install / Use

README

RabbitMQ.Client.Core.DependencyInjection

Usage

Producer

Consumer

Configuration