Processing directly from Redis

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

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

builder.Services
    .AddNexus()
    .AddRedisExternalPendingStorage(options =>
    {
        options.ConnectionString = "localhost:6379";
    });

And then on your queue:

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

For a queue named export_order, the default topology uses keys with the same Redis Cluster hash tag:

• nexus:queue:{export_order} is the ready-message sorted set.
• nexus:queue:{export_order}:delayed is the delayed-message sorted set.
• nexus:queue:{export_order}:readiness is the durable readiness stream.
• nexus:queue:{export_order}:idempotent stores idempotency markers.

Nexus enqueueing writes the message and readiness stream entry atomically. A direct producer must do the same with a Lua script; publishing only to the sorted set is not enough for a readiness-only queue job. The readiness consumer group is independent of message processing and acknowledges a stream entry only after queue activation is durable. See external pending storage for the envelope and script contract.

Redis can also hold Nexus's coalesced activation state instead of SQL:

builder.Services
    .AddNexus()
    .AddRedisQueueActivations(options =>
    {
        options.ConnectionString = "localhost:6379";
        options.KeyPrefix = "nexus";
    });

This is separate from Redis external pending storage: either feature can be used without the other.

When upgrading existing external-pending queues, drain or explicitly migrate the old unbraced Redis keys. V3 does not read them automatically. Include delayed messages and any identity/idempotency state in the cutover.

Read more about external pending storage here.

Enqueueing from Kafka topics

Nexus has built-in support to use the very popular Apache Kafka platform for both instance events as well as taking messages from a Kafka topic and enqueueing into a Nexus queue.

In order to use this you need to install the NuGet package for RabbitMQ called CommerceMind.Nexus.Kafka.

You set it up like this:

builder.Services
    .AddNexus()
    .AddQueues()
    .AddKafka(options =>
    {
        options.ConsumerConfig.BootstrapServers = "...";
        options.ConsumerConfig.GroupId = "...";
    })
    .AddNexusKafkaQueues(options => 
    {
        // This allows you to control things like which status or when messages are processed
        options.EnqueueContextFactory = message => new EnqueueContext { Priority = message.Id != null ? 10 : 0 }
    })
        .EnqueueFrom<KafkaQueueMessage>("kafka-topic");

With this examples all messages posted to the kafka-topic will get placed in the Nexus queue for KafkaQueueMessage.

Instance events

You can also use Kafka as the way to send signals between Nexus instances in a scaled out environment.

builder.Services
    .AddNexus()
    .AddQueues()
    .AddKafka(options =>
    {
        options.ConsumerConfig.BootstrapServers = "...";
        options.ConsumerConfig.GroupId = "...";

        options.ProducerConfig.BootstrapServers = "...";
    })
    .AddKafkaInstanceEvents(options => 
    {
        // Change this to something unique if you're using the same Kafka instance for multiple
        // different Nexus applications
        options.TopicName = "nexus_instance_events";
    });