EzRabbitMQ 0.2.0-beta002

EzRabbitMQ - Easy to use rabbitMQ library

<img src="./docs/images/logo.png" width="200">

Library for .Net Core 5 simplifying the usage of rabbitMQ.

Create mailboxes to receive rabbitMQ messages and use a producer to send messages.

Features :

• Mailboxes options are validated to catch common mistakes.

• Optional AppInsight consumer/ message / producer tracing.

• Error handling with optional retry count.

• PollyRetryPolicies on rabbitMQ calls for client resiliency.

• Consumer options to easyly toggle on/off rabbitMQ features.

Get Started:

Install

// todo: link nuget.org

Nuget-Install EzRabbitMQ

Register services EzRabbitMQ services :

// your service provider
services.AddEzRabbitMQ();

// to configure EzRabbitMQ LogLevel from `Logging:LogLevel:EzRabbitMQ`
// your IHostBuilder
builder.UseEzRabbitMq();

Requirements:

• .Net Core 5
• C# 9 ready
• RabbitMQ Client 6.2.2

Usage:

Create and receive direct Message

For direct message type the routing key of a message must match the routing key binding to a queue.

IProducerService producerService; // use injection to get a IProducerService
IMailboxService mailboxService; // use injection to get a IMailboxService

record DataSample(string Text);

var consumerOptions = new ConsumerOptions();

using var mailbox = mailboxService.Direct<DataSample>("ROUTING KEY", "MAILBOX NAME", consumerOptions);

// event raised on message received
mailbox.OnMessageReceived += (sender, data) => {
    // data is IMessage<DataSample>
    Console.WriteLine($"data received: {data.Text}");
};

// use producer service to send Direct message to mailbox 
producerService.DirectSend("ROUTING KEY", new DataSample("Example"));

Create and receive topic Message

Topic message routing key must follow some rules.

It must be words separated using .(dot) and contains at least a #(hash) can replace one or more values or a *(star) can replace one value.

using var mailbox = mailboxService.Topic<DataSample>("root.#", "ALL_UNDER_ROOT", new ConsumerOptions());

producerService.TopicSend("ROUTING KEY", new DataSample("Example"));

Create and receive fanout Message

Fanout exchange type doesnt need routing key, all message are like broadcasted to the exchange.

using var mailbox = mailboxService.Fanout<DataSample>("FANOUT_LISTENER", new ConsumerOptions());

producerService.FanoutSend(new DataSample("Example"));

Create and receive headers Message

Exchange type headers mailbox takes a XMatch parameter to configure if all headers must match or if any header that match will be received.

Dictionary<string, string> headers = new()
{
    { "type", "jpg" },
    { "format", "chart" }
};
using var mailbox = mailboxService.Headers<DataSample>(headers, XMatch.any, "JPG_RECEIVER", new ConsumerOptions());

producerService.HeadersSend(new Dictionary<string, string>(){
    {"type": "jpg"}
}, new DataSample("Example"));

Create RPC client and server and send message

You can implement a RpcServer that will receive request from a RpcClient.

// your implementation or RpcServerBase
public class IncrementRpcServer: RpcServerBase, 
    IRpcServerHandle<RpcIncrementResponse, RpcIncrementRequest>
{
    public IncrementRpcServer(ILogger<IncrementRpcServer> logger, IMailboxOptions options, ISessionService session, IProducerService producerService, ConsumerOptions consumerOptions) 
    : base(logger, options, session, producerService, consumerOptions)
    {
    }

    public RpcIncrementResponse Handle(RpcIncrementRequest request)
    {
        Logger.LogInformation("rpc received");
        return new RpcIncrementResponse(request.CurrentValue +1);
    }
}

var rpcServer = _mailboxService.RpcServer<IncrementRpcServer>();

var rpcClient = _mailboxService.RpcClient();

var response = _rpcClient.Call<RpcIncrementResponse>(new RpcIncrementRequest(1));
// response.NewValue = 2

Create and receive multiple direct messages

You can implement your own mailbox version, you will be able to implement multiple message handle to receive multiple message within the same context.

record WiseChildGiftRequest(string Name);
record NotWiseChildGiftRequest(string Name);

public class SantaMailbox : MailboxBase,
    IMailboxHandler<WiseChildGiftRequest>,
    IMailboxHandler<NotWiseChildGiftRequest>
{
    private readonly ILogger<TodoMailbox> _logger;

    public SantaMailbox(ILogger<TodoMailbox> logger, IMailboxOptions options, ISessionService session, ConsumerOptions consumerOptions) : base(logger, options, session, consumerOptions) { }

    public void OnMessageHandle(IMessage<WiseChildGiftRequest> message)
    {
        _logger.LogInformation("Received request from wise child");
    }

    public void OnMessageHandle(IMessage<NotWiseChildGiftRequest> message)
    {
        _logger.LogInformation("Received request from not wise child");
    }
}

// create instance of your mailbox
var santaMailbox = mailboxService.Create<SantaMailbox>("SANTA", "SANTA_MAILBOX", consumerOptions);

// send message to SANTA_MAILBOX
producerService.DirectSend("SANTA", new DataSample("Example"));

Configuration Options

You can configure several options using the configure callback in the startup.

Override Serializer/Deserializer

You can override and change the serializer / deserializer.

// your service provider
services.AddEzRabbitMQ(config =>
{
    config.SetSerializer(data => \* returns bytes array*\);
    config.SetDeserializer((bytes, type) => \* returns object from bytes array*\);
});

Set AppInsight InstrumentationKey

By setting a valid InstrumentationKey you will enable appInsight metrics and tracing.

// your service provider
services.AddEzRabbitMQ(config =>
{
     var ik = context.Configuration["ApplicationInsights:InstrumentationKey"];
    if (ik is not null) config.SetInstrumentationKey(ik);
});

Polly retry policy

You can override the retry policy.

// your service provider
services.AddEzRabbitMQ(config =>
{
    config.SetPollyPolicy(Policy
        .Handle<Exception>()
        .WaitAndRetryAsync(5, retryAttempt =>
            TimeSpan.FromSeconds(Math.Pow(2, retryAttempt))
        ));
});

RabbitMQ Connection

You can modify the current connectionFactory or you can create a connection and

// your service provider
services.AddEzRabbitMQ(config =>
{
    config.ConfigureConnection(c =>
    {
        c.Uri = new Uri("https://localhost:");
        return c;
    });

    // or you can create a new connection factory

    config.ConfigureConnection(_ => new ConnectionFactory());
});

ConsumerOptions

This options can add rabbitMQ arguments to enable / disable features.

ConsumerOptions default values:

RetryCount:

You can enable retry on consumer exception by using the consumerOptions object :

var options = new ConsumerOptions
{
    AutoAck = false, // AutoAck must be false to use RetryCount
    RetryCount = 3, // After 3 retry exception in the message will be reject,
};

Durable Queue :

To create a durable queue persistant after server restart :

var options = new ConsumerOptions
{
    QueueDurable = true, // set durable feature on queue (need queue recreation if changed)
    QueueAutoDelete = false. // set auto delete feature on queue (need queue recreation if changed)
};

Durable Exchange :

To create a durable exchange persistant after server restart :

var options = new ConsumerOptions
{
    ExchangeDurable = true, // set durable feature on queue (need queue recreation if changed)
    ExchangeAutoDelete = false. // set auto delete feature on queue (need queue recreation if changed)
};

Exclusive Queue

To allow only one consumer you can enable the Exclusive feature of the queue :

var options = new ConsumerOptions
{
    QueueExclusive  = true, // set exclusive feature on queue (need queue recreation if changed)
};

Limits

To set the prefetch message count limit :

var options = new ConsumerOptions
{
    PrefetchCount = 10, // set the prefetch amount of message read by this consumer
    PrefetchGlobal = false // channel global or by consumer limit
};

License

// here goes the license

Contribution guide

// short explain