Sample Projects: RabbitLight.Samples

RabbitLight

RabbitLight is a RabbitMQ Client for .NET developed with simplicity in mind.

Messages are routed to their respective consumers using Attributes, similar to the [Route("my-route")] attribute used on AspNetCore projects.

It also comes with Auto Scaling and Self Healing on the client side to ensure that your application is always connected to the broker, while optimizing the usage of the machine's resources with parallel processing.

To create a consumer, you just have to:

1. Create a class that inherits from ConsumerBase

2. Use [Exchange("my-exchange")] to bind a exchange to that class

3. Add [Queue("my-queue")] to bind a queue to a method from that class

[Exchange("my-exchange")]
public class MyConsumer : ConsumerBase
{
    [Queue("my-queue")]
    public async Task MyQueue(MessageContext<MyMessage> context)
    {
        // ...
    }
}

And to publish a message:

1. Inject the context, and get the publisher:

public class ExampleController : ControllerBase
{
    private readonly IPublisher _publisher;

    public ExampleController(ExampleContext busContext)
    {
        _publisher = busContext.Publisher;
    }

    // ...
}

2. Publish a message:

await _publisher.PublishString("my-exchange", "routing-key", "Hello, World!");

How to use

1. Create a Context

Think of a context as a unique instance of a client, that listens and/or publishes to a specific RabbitMQ server.

You may have multiple contexts in your application, but a context should only be added once per application (if it's added multiple times, only the first registration is considered).

public class ExampleContext : RabbitLightContext
{
    public ExampleContext(IServiceProvider sp, ContextConfig config) : base(sp, config)
    {
    }
}

2. On your Startup class, add the context configuration:

// ...
using RabbitLight.Extensions;
// ...

public void ConfigureServices(IServiceCollection services)
{
    // ...

    services.AddRabbitLightContext<ExampleContext>(config =>
    {
        // Context Alias (used for logging)
        config.Alias = nameof(ExampleContext);

        // Use IHostedService vs IHostApplicationLifetime (requires IApplicationBuilder.UseRabbitLight)
        config.UseHostedService = true;

        // Connection Config (from appsettings.json, explained later on this README.md)
        config.ConnConfig = ConnectionConfig.FromConfig(Configuration.GetSection("RabbitLight"));

        // Consumer Types to include
        config.Consumers = Assembly.GetEntryAssembly().GetTypes();

        // Callback called after all necessary configuration has been completed
        config.OnConfig = async (sp) =>
        {
            var context = sp.GetRequiredService<AspNetAppContext>();
            await context.Api.CreateExchange("manual-exchange");
            await context.Api.CreateQueue("manual-queue");
            await context.Api.CreateBind("manual-queue", "manual-exchange", "manual-key");
        };

        // Callback called before a consumer is invoked
        config.OnStart = (sp, consumer, ea) => Task.Run(() =>
            consumer.Logger.LogInformation($"Starting {consumer.Type.Name}: {ea.DeliveryTag}"));

        // Callback called after a consumer is successfully invoked
        config.OnEnd = (sp, consumer, ea) => Task.Run(() =>
            consumer.Logger?.LogInformation($"Ending {consumer.Type.Name}: {ea.DeliveryTag}"));

        // Callback called after the ACK message is sent
        config.OnAck = (sp, consumer, ea) => Task.Run(() =>
            consumer.Logger?.LogInformation($"Acked {consumer.Type.Name}: {ea.DeliveryTag}"));

        // Global error handler, whose return identifies the requeue strategy
        // null -> Nack w/o requeue; TimeSpan == 0 -> Requeue immediately; TimeSpan > 0 -> Requeue after delay;
        config.OnError = (sp, consumer, ea, ex) => Task.Run(() =>
        {
            consumer.Logger?.LogError($"Handled error in {consumer.Type.Name}: {ea.DeliveryTag}");

            // Requeue if the queue doesn't have a Dead Letter as fallback
            var requeue = !(consumer?.Queue?.Arguments?.ContainsKey("x-dead-letter-exchange")).GetValueOrDefault();

            var requeueDelay = config.ConnConfig.RequeueDelay ?? TimeSpan.FromSeconds(30);
            return requeue ? requeueDelay : default(TimeSpan?);
        });

        // Optional callback called after a publisher receives a NACK from the server
        config.OnPublisherNack = (sp, sender, ea) => Task.Run(() =>
        {
            var logger = sp.GetService<ILoggerFactory>()?.CreateLogger<IPublisher>();
            logger?.LogError($"Publisher error: {ea.DeliveryTag}");
        });
    });

    // ...
}

public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
    // ...

    // Only needed if config.UseHostedService == false
    app.UseRabbitLight<ExampleContext>();

    // ...
}

3. Create a consumer:

// ...
using RabbitLight.Consumer;
using RabbitLight.Exceptions;
// ...

[Exchange("exchange", ExchangeType.Topic)] // ExchangeTypes accepted by RabbitMQ (Direct, Fanout, Headers, Topic)
public class ExampleConsumer : ConsumerBase
{
    private readonly ILogger<ExampleConsumer> _logger;

    public ExampleConsumer(ILogger<ExampleConsumer> logger) // Any injected services
    {
        _logger = logger;
    }

    [Queue("queue1")] // routing key defaults to "#"
    public async Task Example(MessageContext<ExampleMessage> context)
    {
        // Routes:
        // EXCHANGE      ROUTING KEY      QUEUE
        // exchange  ->       #       ->  queue1

        // Get headers
        var headers = context.Headers();

        // Get the message
        var msgBytes = context.MessageBytes();
        var msgStr = context.MessageString();
        var msgXml = context.MessageXml();
        var msgJson = context.MessageJson();

        // Automatically chooses a parser based on the Content  Type Header
        var msg = context.Message();

        // Your code here...
    }

    [Queue("queue2", "key2")]
    [Queue("queue3", "key3", "key30")]
    public void ExampleMulti(MessageContext<ExampleMessage> context)
    {
        // Routes:
        // EXCHANGE      ROUTING KEY      QUEUE
        // exchange  ->     key2      ->  queue2
        // exchange  ->     key3      ->  queue3
        // exchange  ->     key30     ->  queue3

        var msg = context.Message();

        // Your code here...
    }

    [Queue("queue4", maxChannels: 1, routingKeys: "key4")]
    public void ExampleSerially(MessageContext<ExampleMessage> context)
    {
        // Routes:
        // EXCHANGE      ROUTING KEY      QUEUE
        // exchange  ->     key4      ->  queue4

        var msg = context.Message();

        // Your code here...
    }
}

public class ExampleMessage
{
    public string Text { get; set; }
}

Tip

Instead of using hard-coded strings, you can also use defined constants:

public const string MyExchange = "my-exchange";
[Exchange(Exchanges.MyExchange)]

Note

Remember that if another consumer class is listening to the same exchange and/or queues the messages will be routed acording to the ExchangeType selected.

Example:

[Exchange("exchangeA")]
public class ConsumerA : ConsumerBase
{
    [Queue("queue1", "routingA")]
    public async Task Example(MessageContext<ExampleMessage> context)
    {
        // ...
    }
}

[Exchange("exchangeB")]
public class ConsumerB : ConsumerBase
{
    [Queue("queue1", "routingB")]
    public async Task Example(MessageContext<ExampleMessage> context)
    {
        // ...
    }
}

When a message with the routingA routing key is received, there is no guarantee that it will be routed to ConsumerA rather than ConsumerB, as they share the same destination queue. If there is need to check the routing key from a message, use context.EventArgs.RoutingKey.

4. Create a publisher:

// ...
using RabbitLight.Publisher;
// ...

[ApiController]
[Route("[controller]")]
public class ExampleController : ControllerBase
{
    private readonly IPublisher _publisher;

    public ExampleController(ExampleContext busContext)
    {
        _publisher = busContext.Publisher;
    }

    [HttpPost]
    public async Task<string> SendMessage()
    {
        var body = new Example { Text = "Hello, World!" };

        // Publish byte[]
        await _publisher.Publish("exchange", "key1", new byte[] { });

        // Publish string
        await _publisher.PublishString("exchange", "key1", "Hello, World!");

        // Publish Json
        await _publisher.PublishJson("exchange", "key1", body);

        // Publish Xml
        await _publisher.PublishXml("exchange", "key1", body);

        // Publish Batch (byte[], string, Json and/or Xml)
        await _publisher.PublishBatch(new List<PublishBatch> {
            new PublishBatch("exchange", "key1", MessageType.String, "Hello, World!"),
            new PublishBatch("exchange", "key2", MessageType.Json, body),
        });

        return "Message published!";
    }
}

5. In appsettings.json add the RabbitLight property:

{
  "RabbitLight": {
    "UserName": "guest",
    "Password": "guest",
    "VirtualHost": "/",
    "HostName": "127.0.0.1",
    "Port": 5672,
    "PortApi": 15672,
    "MinChannels": 10,
    "MaxChannels": 50,
    "ScallingThreshold": 500,
    "PrefetchCount": 10,
    "ChannelsPerConnection": 20,
    "RequeueDelay": "00:00:30",
    "MonitoringInterval": "00:01:00",
    "SkipVHostConfig": false,
    "SkipDeclarations": false
  }
}

Field	Description
RabbitMQ Config:	----------------------
UserName	Username to use when authenticating to the server.
Password	Password to use when authenticating to the server.
VirtualHost	Virtual host to access during this connection.
HostName	The host to connect to.
Port	The port to connect on.
AmqpProtocol	The amqp protocol to use (amqp or amqps).
ApiProtocol	The api protocol to use (http or https).
RabbitLight Config:	----------------------
PortApi	Port where RabbitMQ management UI plugin is available.
MinChannels	Minimum number of parallel channels for the whole application.
MaxChannels	Maximum number of parallel channels for the whole application.
ScallingThreshold	Number of messages required to scale a new channel (e.g. 500 messages) or null to disable scalling.
PrefetchCount	Number of messages that will be cached by each channel at once.
ChannelsPerConnection	Number of channels per connection (RabbitMQ's IConnection).
RequeueDelay	Delay for when Nacking a message for requeue or null to instantaneous (can be overridden by OnError callback).
MonitoringInterval	Interval regarding channel monitoring tasks (health check, auto scalling and self healing)
SkipVHostConfig	Skip VHost creation and other configs
SkipDeclarations	Skip queue and exchange declaration and bind

Bonus: Console App

Using RabbittLight with a simple Console App

using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using RabbitLight.Config;
using RabbitLight.ConsoleApp.Consumers.Context;
using System;
using System.IO;
using System.Reflection;

class Program
{
    static void Main(string[] args)
    {
        // Build appsettings.json Configurations
        var env = Environment.GetEnvironmentVariable("ENVIRONMENT") ?? "Development";
        var configuration = new ConfigurationBuilder()
            .SetBasePath(Directory.GetParent(AppContext.BaseDirectory).FullName)
            .AddJsonFile("appsettings.json", false, false)
            .AddJsonFile($"appsettings.{env}.json", false, false)
            .Build();

        // Build Service Provider
        var serviceProvider = new ServiceCollection()
            .AddLogging(c => c.AddConsole())
            .BuildServiceProvider();

        // Create Context
        var context = new TestContext(serviceProvider, new ContextConfig
        {
            ConnConfig = ConnectionConfig.FromConfig(configuration.GetSection("RabbitLight")),
            Consumers = Assembly.GetEntryAssembly().GetTypes()
        });

        // Register ContextConsumers
        context.Register().Wait();

        // Publish a message
        context.Publisher.PublishString("exchange", "*", "Hello, World").Wait();

        // Prevent App From Closing
        Console.ReadLine();

        // Dispose Dependencies
        serviceProvider.Dispose();
    }
}