Processing directly from RabbitMQ queues

Nexus allows you to use RabbitMQ as an external pending storage, which means that Nexus reads and processes messages directly from RabbitMQ without first storing them in an SQL backed Nexus queue.

Use it by first installing CommerceMind.Nexus.RabbitMQ and then configuring the external storage:

builder.Services
    .AddNexus()
    .AddRabbitMqExternalPendingStorage(options =>
    {
        options.ConnectionString = new Uri("amqp://guest:guest@localhost/");
    });

And then on your queue:

[Queue("export_order", StorageMode = QueueStorageMode.ExternalPending)]
public class ExportOrderQueueMessage : IQueueMessageWithId
{ }

For each Nexus queue, v3 declares a durable fan-out exchange and two durable queues. With the defaults, export_order uses:

• Exchange nexus:external-pending:export_order.
• Processing queue nexus:export_order.
• Readiness queue nexus:export_order:readiness.

Nexus leases messages only from the processing queue. The readiness bridge consumes the readiness queue and acknowledges a notification only after activation is durable. Direct producers must publish to the exchange, not directly to the processing queue, so both paths receive the message. Existing producers that publish directly to nexus:export_order must be migrated before the queue job is changed to readiness-only mode.

Read more about external pending storage here.

Enqueueing from RabbitMQ queues

RabbitMQ is a great way to communicate between different applications. But sometimes you want to have more visibility into queue messages, when they're processed and if they fail. And to be able to retry in a controlled fashion when they fail.

Nexus lets you get the best of both worlds by automatically enqueueing from a RabbitMQ queue to a Nexus queue like the example below. In order to use this you need to install the NuGet package called CommerceMind.Nexus.RabbitMQ.

builder.Services
    .AddNexus()
    .AddQueues()
    .AddRabbitMq(options =>
    {
        options.UserName = "...";
        options.Password = "...";
        options.HostNames = ["localhost"];

        // Or:
        options.ConnectionString = new Uri("amqp://...");
    })
    .AddNexusRabbitMqQueues(options => 
    {
        // This allows you to control things like which status or when messages are processed
        options.EnqueueContextFactory = message => new EnqueueContext { Priority = message is MyImportantQueueMessage ? 10 : 0 }
    })
        .EnqueueFrom<RabbitQueueMessage>(new NexusRabbitMqQueue
        {
            Name = "rabbit_queue",
            Bindings = [new NexusRabbitMqQueueBinding
            {
                ExchangeName = "my_exchange",
                RoutingKey = ""
            }]
        });

Instance events

You can also use RabbitMQ to send signals between Nexus instances in a scaled out environment.

builder.Services
    .AddNexus()
    .AddQueues()
    .AddRabbitMq(options =>
    {
        options.UserName = "...";
        options.Password = "...";
        options.HostNames = ["localhost"];

        // Or:
        options.ConnectionString = new Uri("amqp://...");
    })
    .AddRabbitMqInstanceEvents(options => 
    {
        // Change this to something unique if you're using the same RabbitMQ instance for multiple
        // different Nexus applications
        options.ExchangeName = "nexus_instance_events";
    });