Please read the Introduction before reading this provider documentation.
The RabbitMQ transport provider uses RabbitMQ.Client client to connect to the RabbitMQ cluster via the AMQP protocol.
The RabbitMQ and AMQP protocol introduce couple of concepts:
- Exchange - entities to which producers send messages,
- Queue - mailboxes which consumers read messages from,
- Binding - are rules that exchanges use to route messages to queues.
AMQP Concepts provides a brilliant overview.
The RabbitMQ transport configuration is arranged via the .WithProviderRabbitMQ(cfg => {}) method on the message bus builder.
using SlimMessageBus.Host.RabbitMQ;
// Register SlimMessageBus in MSDI
services.AddSlimMessageBus((mbb) =>
{
// Use RabbitMQ transport provider
mbb.WithProviderRabbitMQ(cfg =>
{
// Connect using AMQP URI
cfg.ConnectionString = configuration["RabbitMQ:ConnectionString"];
// Alternatively, when not using AMQP URI:
// cfg.ConnectionFactory.HostName = "..."
// cfg.ConnectionFactory.VirtualHost = "..."
// cfg.ConnectionFactory.UserName = "..."
// cfg.ConnectionFactory.Password = "..."
// cfg.ConnectionFactory.Ssl.Enabled = true
// Fine tune the underlying RabbitMQ.Client:
// cfg.ConnectionFactory.ClientProvidedName = $"MyService_{Environment.MachineName}";
});
mbb.AddServicesFromAssemblyContaining<PingConsumer>();
mbb.AddJsonSerializer();
});The relevant elements of the cfg:
- The
ConnectionStringallows to set the AMQP URI. This property is a convenience wrapper on top ofConnectionFactory.Urifrom the underlying client library. The URI has the following form:amqps://<username>:<password>@<host>/<virtual-host>. - The
ConnectionFactoryallows to access other client settings. It can be used to setup other connection details in case the AMQP URI cannot be used or there is a need to fine tune the client. For more options see the underlying RabbitMQ driver docs.
Producers need to declare the exchange name and type the message should be delivered to. SMB will provision the specified exchange. Additionally, we can specify:
- the modifier that allows to assign message properties (
MessageId,ContentType, and headers), - the message key provider that is used in routing for relevant exchange types.
mbb.Produce<OrderEvent>(x => x
// Will declare an orders exchange of type Fanout
.Exchange("orders", exchangeType: ExchangeType.Fanout)
// Will use a routing key provider that for a given message will take it's Id field
.RoutingKeyProvider((m, p) => m.Id.ToString())
// Will use
.MessagePropertiesModifier((m, p) =>
{
p.MessageId = GetMessageId(m);
}));We can also set defaults for all producers on the bus level:
services.AddSlimMessageBus((mbb) =>
{
mbb.WithProviderRabbitMQ(cfg =>
{
// All exchanges declared on producers will be durable by default
cfg.UseExchangeDefaults(durable: true);
// All messages will get the ContentType message property assigned
cfg.UseMessagePropertiesModifier((m, p) =>
{
p.ContentType = MediaTypeNames.Application.Json;
});
});
mbb.AddJsonSerializer();
});Consumers need to specify the queue name from which the consumer should be reading from. SMB will provision the specified queue. Additionally,
- when the exchange name binding is specified then SMB will provision that binding with the broker,
- when dead letter exchange is specified then the queue will provisioned with the broker, and if the exchange type is specified it will also be provisioned.
mbb.Consume<PingMessage>(x => x
// Use the subscriber queue, do not auto delete
.Queue("subscriber", autoDelete: false)
//
.ExchangeBinding("ping")
// The queue declaration in RabbitMQ will have a reference to the dead letter exchange and the DL exchange will be created
.DeadLetterExchange("subscriber-dlq", exchangeType: ExchangeType: Direct)
.WithConsumer<PingConsumer>());We can specify defaults for all consumers on the bus level:
services.AddSlimMessageBus((mbb) =>
{
mbb.WithProviderRabbitMQ(cfg =>
{
cfg.UseDeadLetterExchangeDefaults(durable: false, autoDelete: false, exchangeType: ExchangeType.Direct, routingKey: string.Empty);
cfg.UseQueueDefaults(durable: false);
});
});When a consumer processes a message from a RabbitMQ queue, it needs to acknowledge that the message was processed. RabbitMQ supports three types of acknowledgments out which two are available in SMB:
- Ack - to indicate the message was successfully processed and it should be removed from the queue.
- Nack (negative ack) - when the message was processed but resulted in an error, while still it needs to be removed from the queue or retried (depending what the user chooses in the given use case).
In SMB we can set the acknowledgment mode for each consumer:
builder.Consume<PingMessage>(x => x
.Queue("subscriber", autoDelete: false)
.ExchangeBinding(topic)
// Set the acknowledgement mode, the ConfirmAfterMessageProcessingWhenNoManualConfirmMade is the default
.AcknowledgementMode(RabbitMqMessageAcknowledgementMode.ConfirmAfterMessageProcessingWhenNoManualConfirmMade)
.WithConsumer<PingConsumer>());Alternatively, a bus wide default can be specified for all consumers:
services.AddSlimMessageBus((mbb) =>
{
mbb.WithProviderRabbitMQ(cfg =>
{
// Set the acknowledgement mode, the ConfirmAfterMessageProcessingWhenNoManualConfirmMade is the default
cfg.AcknowledgementMode(RabbitMqMessageAcknowledgementMode.ConfirmAfterMessageProcessingWhenNoManualConfirmMade);
});
});See the RabbitMqMessageAcknowledgementMode has the available options.
By default (ConfirmAfterMessageProcessingWhenNoManualConfirmMade), messages are acknowledge (Ack) after the message processing finish with success.
If an exception where to happen the message is rejected (Nack) (or else whatever the custom error handler logic does).
In that default mode, the user can still Ack or Nack the message depending on the need inside of the consumer or interceptor using the Ack() / Nack() methods exposed on the ConsumerContext - this allows for manual acknowledgements. The default setting is optimal and safe. However, message retries could happen (at-least-once delivery).
The other acknowledgement modes will ack the message before processing, but are less safe as it will lead to at-most-once delivery.
By default the the transport implementation performs a negative ack (nack) in the AMQP protocol for any message that failed in the consumer. As a result the message will be marked as failed and routed to an dead letter exchange or discarded by the RabbitMQ broker.
The recommendation here is to either:
- configure a dead letter exchange on the consumer queue,
- or provide a custom error handler (retry the message couple of times, if failed send to a dead letter exchange).
The Dead Letter Exchanges is a feature of RabbitMQ that will forward failed messages from a particular queue to a special exchange.
In SMB on the consumer declaration we can specify which dead letter exchange should be used:
mbb.Consume<PingMessage>(x => x
.Queue("subscriber", autoDelete: false)
.ExchangeBinding(topic)
// The queue provisioned in RabbitMQ will have a reference to the dead letter exchange
.DeadLetterExchange("subscriber-dlq")
.WithConsumer<PingConsumer>());However, the subscriber-dlq will not be created by SMB in the sample. For it to be created the ExchangeType has to be specified, so that SMB knows what exchange type should it apply.
It can be specified on the consumer:
mbb.Consume<PingMessage>(x => x
.Queue("subscriber", autoDelete: false)
.ExchangeBinding(topic)
// The queue provisioned in RabbitMQ will have a reference to the dead letter exchange and the DL exchange will be provisioned
.DeadLetterExchange("subscriber-dlq", exchangeType: ExchangeType: Direct)
.WithConsumer<PingConsumer>());Alternatively, a bus wide default can be specified for all dead letter exchanges:
services.AddSlimMessageBus((mbb) =>
{
mbb.WithProviderRabbitMQ(cfg =>
{
// All the declared dead letter exchanges on the consumers will be of Direct type
cfg.UseDeadLetterExchangeDefaults(durable: false, autoDelete: false, exchangeType: ExchangeType.Direct, routingKey: string.Empty);
});
});Define a custom class implementation of IRabbitMqConsumerErrorHandler<>:
public class CustomRabbitMqConsumerErrorHandler<T> : IRabbitMqConsumerErrorHandler<T>
{
// Inject needed dependencies via construction
public async Task<bool> OnHandleError(T message, Func<Task> retry, IConsumerContext consumerContext, Exception exception)
{
// Check if this is consumer context for RabbitMQ
var isRabbitMqContext = consumerContext.GetTransportMessage() != null;
if (isRabbitMqContext)
{
if (exception is TransientException)
{
// Send negative acknowledge but ask the broker to retry
consumerContext.NackWithRequeue();
}
else
{
// Send negative acknowledge (if dead letter setup it will be routed to it)
consumerContext.Nack();
}
// Mark that the errored message was handled
return true;
}
return false;
}
}Then register the implementation in MSDI for all (or specified) message types.
// Register error handler in MSDI for any message type
services.AddTransient(typeof(IRabbitMqConsumerErrorHandler<>), typeof(CustomRabbitMqConsumerErrorHandler<>));When error handler is not found in the DI for the given message type, or it returns
false, then default error handling will be applied.
See also the common error handling.
By default each consumer in the service process will handle one message at the same time.
In order to increase the desired concurrency, set the ConsumerDispatchConcurrency to a value greater than 1.
This is a setting from the underlying RabbitMQ driver that SMB uses.
services.AddSlimMessageBus((mbb) =>
{
mbb.WithProviderRabbitMQ(cfg =>
{
cfg.ConnectionFactory.ConsumerDispatchConcurrency = 2; // default is 1
// ...
}
}Notice that increasing concurrency will cause more messages to be processed at the same time within one service instance, hence affecting order of consumption. In scenarios where order of consumption is important, you may want to keep concurrency levels set to 1.
Here is an example how to set-up request-response flow over RabbitMQ. The fanout exchange types was used, but other type could be used as well (altough we might have to provide the routing key provider on the producer side.)
services.AddSlimMessageBus((mbb) =>
{
// ...
mbb.Produce<EchoRequest>(x =>
{
// The requests should be send to "test-echo" exchange
x.Exchange("test-echo", exchangeType: ExchangeType.Fanout);
})
.Handle<EchoRequest, EchoResponse>(x => x
// Declare the queue for the handler
.Queue("echo-request-handler")
// Bind the queue to the "test-echo" exchange
.ExchangeBinding("test-echo")
// If the request handling fails, the failed messages will be routed to the DLQ exchange
.DeadLetterExchange("echo-request-handler-dlq")
.WithHandler<EchoRequestHandler>())
.ExpectRequestResponses(x =>
{
// Tell the handler to which exchange send the responses to
x.ReplyToExchange("test-echo-resp", ExchangeType.Fanout);
// Which queue to use to read responses from
x.Queue("test-echo-resp-queue");
// Bind to the reply to exchange
x.ExchangeBinding();
// Timeout if the response doesn't arrive within 60 seconds
x.DefaultTimeout(TimeSpan.FromSeconds(60));
});
});SMB automatically creates exchanges from producers, queues, dead letter exchanges and bindings from consumers.
However, if you need to layer on other topology elements (or peform cleanup) this could be achieved with UseTopologyInitializer():
services.AddSlimMessageBus((mbb) =>
{
mbb.WithProviderRabbitMQ(cfg =>
{
cfg.UseTopologyInitializer((channel, applyDefaultTopology) =>
{
// perform some cleanup if needed
channel.QueueDelete("subscriber-0", ifUnused: true, ifEmpty: false);
channel.QueueDelete("subscriber-1", ifUnused: true, ifEmpty: false);
channel.ExchangeDelete("test-ping", ifUnused: true);
channel.ExchangeDelete("subscriber-dlq", ifUnused: true);
// apply default SMB infered topology
applyDefaultTopology();
});
});
});Avoiding the call applyDefaultTopology() will suppress the SMB inferred topology creation.
This might be useful in case the SMB inferred topology is not desired or there are other custom needs.
- Default type exchanges are not yet supported
- Broker generated queues are not yet supported.
Open a github issue if you need a feature, have a suggestion for improvement, or want to contribute an enhancement.