Snail.Toolkit.RabbitMQ 1.0.0

Toolkit.RabbitMQ

Extension for the framework RabbitMQ.Client

Concepts

• Release immutability

  • The topology for each release must be strictly defined and not changed during its existence
    • Until application is running you can't:
      • Cancel consumers
      • Remove queues and exchanges
      • Change bindings

• Declarativeness and simplicity

  • Intuitive RabbitMQ-close fluent interfaces
  • Generic type checks
  • Connection, Exchange, Queue, Consumer and Producer declaration classes

Installation

dotnet add package Snail.Toolkit.RabbitMQ

Usage (examples)

services.AddRabbit(options =>
        options.UseConnectionFactory(factory => ...), 
    connections => connections
    .AddConnection(connection =>
    {
        // Enable type checks with .AddDirectExchange<TPayload>() extension
        var exchange = connection.AddDirectExchange("exchange")
            // .NoWait()
            // .Deleted()
            .AutoDelete()
            .Durable();

        var queue = connection.AddQueue<Payload>("queue")
            // .Exclusive()
            // .Deleted()
            // .NoWait()
            // .Lazy()
            // .MaxPriority(10)
            // .TimeToLive(10000)
            // .MaxMessageSize(1000)
            // .MaxMessageCount(1000)
            // .MessageTimeToLive(1000)
            // .RejectPublish()
            // .Quorum()
            // .SingleActiveConsumer()
            .AutoDelete()
            .Durable()
            // Type checks
            .BoundTo(exchange);

        // Type checks
        connection.AddConsumer(queue)
            // .Tagged("tag")
            // .Prefetch(1)
            // .Count(1)
            // .Exclusive()
            // .NoLocal()
            // .AutoAck()
            // .Requeue()
            // .Priority(2)
            .Count(2)
            .Requeue()
            .Subscribe<Payload, Consumer>();
            .Subscribe(...);

        // Type cheks
        connection.AddProducer<Payload>(exchange)
            // .Transactional()
            // .Mandatory()
            // .MessageTimeToLive(10000)
            .Persistent();
    }));
    
public class Consumer : IConsumer<Payload> 
{
    public ValueTask HandleAsync(Payload message, CancellationToken cancellationToken)
    {
        Console.WriteLine($"Received: {message}");
        return ValueTask.CompletedTask;
    }
}

// Get or inject
var producer = serviceProvider.GetRequiredService<IRabbitProducer>();

// Use
await producer.PublishAsync(new Payload(), overrides => ...);

Sample configuration appsettings.json for using Vault

{
  "RabbitMQ": {
    "HostName": "localhost",
    "UserName": "guest",
    "Password": "guest"
    "ClientProvidedName": "default"
  }
}

Supported

• Durable, internal, dead letter, bound and alternate exchanges
• Lazy, durable and exclusive queues
• Default, confirm and transactional channel modes
• Persistent producers
• Consumers priority
• Queue and message time to live
• Max message count and size limitations
• Batch produce
• App id declaration
• Reject-publish when queue is full
• Deleted operations
• NoWait operations

Queues

• Declare durable queue with Durable extension
• Declare exclusive queue with Exclusive extension
• Declare queue without waiting with NoWait extension
• Bind exchange to exchange with BoundTo extension
• Declare lazy queue with Lazy extension
• Set queue max message count with MaxMessageCount extension
• Set queue max message size in bytes with MaxMessageSize extension
• Set dead letter exchange with DeadLetterTo extension
• Set queue ttl with TimeToLive extension
• Set message ttl with MessageTimeToLive extension
• Set queue max priority with MaxPriority extension
• Explicitly delete queue with Deleted extension
• Delete queue automatically with AutoDelete extension
• Add custom arguments with Argument extension

Exchanges

• Declare durable exchange with Durable extension
• Declare exchange without waiting with NoWait extension
• Delete exchange automatically with AutoDelete extension
• Explicitly delete exchange with Deleted extension
• Bind exchange to exchange with BoundTo extension
• Declare alternate exchange with AlternateTo extension
• Add custom arguments with Argument extension
• Declare exchange with AddDirectExchange(...), AddFanoutExchange(...), AddHeadersExchange(...), AddTopicExchange(...) extensions

Consumers

• Tag consumers using Tagged extension
• Limit prefetch count with Prefetch extension
• Scale consumers by using Count extension
• Declare exclusive consume with Exclusive extension
• Forbid to consume own producers with NoLocal extension
• When no need to ack explicitly use AutoAck extension
• Requeue messages on fail with Requeue extension
• Set consumer priority with Priority extension
• Add custom arguments with Argument extension
• All consumers start in IHostedService

Producers

• Set routing key RoutingKey extension
• Set mandatory with Mandatory extension
• Set message priority with Priority extension
• Set message ttl with MessageTimeToLive extension
• Use channel transactional mode with Transactional extension
• Use message persistence with Persistent extension
• Configure headers with Header extension
• Configure properties with Property extension

Limitations

• No dynamic topology declaration by design, but you can use IRabbitConnectionProvider for that
• No correlation-id's
• No message-id's
• No BasicReturn and other events for now

Tips

• RoutingKey is QueueName/ExchangeName by default
• Do not use same connection for consumers and producers because of tcp backpressure

License

Snail.Toolkit.RabbitMQ is a free and open source project, released under the permissible MIT license.