Named queues

Named queues let several independently processed queues share a CLR message type. Define the destinations when registering queues:

public class LanguageMessage : IQueueMessageWithId
{
    public string? Id { get; set; }
    public string? Text { get; set; }
}

builder.Services.AddNexus().AddQueues(options =>
{
    options.NamedQueues = [
        new NamedQueue<LanguageMessage> { Name = "language_sv", DisplayName = "Swedish" },
        new NamedQueue<LanguageMessage> { Name = "language_en", DisplayName = "English" }
    ];
});

Queue names identify persistent storage and must remain stable. Use letters, digits, and underscores. Virtual queues have no storage, so their names may also contain hyphens and periods. Nexus rejects names that collide with another queue's backing tables, including provider identifier truncation. Definitions support the usual per-queue retention, retry, health, category, storage-mode, external-provider, processing-delay, and activation-delay options. Explicit named definitions suppress automatic concrete queue discovery for their message type. Queue names are unique, ignoring case, across concrete and virtual queues.

Contributions from libraries

Libraries can call AddNexus().AddQueues() and AddJobs() independently, in either library order. Each method shares its options across calls. Append to NamedQueues, GeneratedQueues, GeneratedJobs, and assembly lists with Add or AddRange; assigning a new list replaces earlier definitions. Scalar settings use the last configured value, including job background-worker settings.

var nexus = builder.Services.AddNexus();

// Library A
nexus.AddQueues(options => options.NamedQueues.Add(
    new NamedQueue<LanguageMessage> { Name = "language_sv", DisplayName = "Swedish" }));
nexus.AddJobs(options => options.AssembliesToScanForJobs.Add(typeof(LanguageJob).Assembly));

// Library B may contribute more queues after Library A registered its jobs.
nexus.AddQueues(options => options.NamedQueues.Add(
    new NamedQueue<LanguageMessage> { Name = "language_en", DisplayName = "English" }));

nexus.Build();

Call Build() once after all contributions. Nexus registers the complete queue set before binding jobs and distributed processors, so an ordinary LanguageJob discovered by Library A also binds to Library B's queue. Each assembly scan uses the filter configured at that call; discovering a type again does not register it twice. Explicit named definitions suppress ordinary concrete discovery even when contributed later. Repeated calls without new definitions do not duplicate registrations. Adding duplicate queue or job names fails at Build(), ignoring case, as do conflicting processing implementations.

Generated job bindings still require their concrete queues to be declared before the contributing AddJobs() call. An explicit generated binding takes precedence over automatic binding of the same job implementation to that queue. Registering queues alone does not enable processing: call AddJobs() or AddQueueProcessors() explicitly. Call AddQueues() before AddQueueProcessors().

Enqueueing

Supply the queue name on every dispatch to a named queue:

await enqueuer.EnqueueAsync("language_sv", new LanguageMessage { Id = "123", Text = "Hej" });
await enqueuer.EnqueueAsync("language_en", messages);

This applies even when there is only one named queue for the type. EnqueueAsync<LanguageMessage>(message) cannot infer a named destination. Adding or removing a language must not change whether a type-only call is valid. An unknown name, incompatible payload, or ambiguous unnamed destination fails with an error.

For a queue-bound typed accessor, explicitly select the name through IQueueResolver:

var swedish = queues.GetEnqueuer<LanguageMessage>("language_sv");
await swedish.EnqueueAsync(message);
var reader = queues.GetReader<LanguageMessage>("language_sv");
var updater = queues.GetUpdater<LanguageMessage>("language_sv");

Once selected, the accessor's ordinary methods operate on that queue. Advanced custom processing can also enumerate IQueueReader or IQueueItemUpdater and select by QueueName. Unqualified typed DI resolution cannot choose among named destinations.

Validation and priority calculation remain type-based. The validator for LanguageMessage applies to every concrete queue receiving that type, unless SkipValidation is set. Explicit enqueue priority takes precedence over IMessagePriorityCalculator<LanguageMessage>. Applications can include routing information on their own payload if these hooks need it.

Message identity, idempotency, previous successfully processed versions, history, partial merges, and processing updates are local to each queue. The same message ID can exist independently in both queues.

Processing

One ordinary processor class handles every concrete queue registered for its message type:

public class LanguageProcessor : IQueueProcessor<LanguageMessage>
{
    public Task<ProcessResults> ProcessMessageAsync(
        LanguageMessage message,
        QueueProcessingContext context,
        CancellationToken cancellationToken)
    {
        var queueName = context.QueueName;
        return Task.FromResult(ProcessResults.Processed());
    }
}

Queue jobs work the same way. A job binding has its own name, schedule, pause state, and history for each queue; named bindings use {job-name}_{queue-name} by default. Existing single ordinary queue jobs keep their names. Explicit generated job definitions can preserve existing per-queue job names and metadata.

Execution is independent per queue. With four workers per instance and ten queues, each instance can run forty workers. Each queue has its own leases, retries, activation, and administrative state. A queue can have one queue job or one distributed processor. Existing storage-mode restrictions still apply.

Single-message jobs and processors can receive QueueProcessingContext. Existing implementations of the overload without context continue to work. Batches expose batch.QueueName and batch.Context; each batch belongs to one queue. Named queues do not require generic subclasses of the message or job implementation.

Virtual entry points

Keep the ordinary different-type transformer for the common case:

public TransformResult<ExportOrderMessage> Transform(OrderPlacedMessage message, EnqueueContext? context)
    => new(new ExportOrderMessage { Id = message.Id });

Select a named destination with one extra argument:

return new(message, queueName: "language_sv");

Or send to several named queues through the same result type:

return TransformResult<LanguageMessage>.ToMany([
    new("language_sv", message),
    new("language_en", message)
]);

Register an explicit virtual queue using NamedQueue<T> with IsVirtual = true and TransformerTypes = [typeof(MyTransformer)]. Its input may use the same CLR type as its concrete destinations. An ordinary [VirtualQueue] registration remains supported. The virtual queue has its own name and no storage or processing settings. Incoming enqueue context is inherited unless a result/destination replaces it. Empty results filter out a message. Forwarding to another virtual queue is supported; routing cycles fail with a path description.

Enqueue into the virtual queue's name to request routing, or into a concrete name to choose a destination directly. Multiple destinations do not imply an atomic transaction across providers. A retry after partial delivery may repeat destinations that already succeeded.

Runtime processing settings

The Processing tab in the new Admin UI lets you override the code defaults for one concrete queue. Distributed processors expose workers per instance, lease size, lease time, and batch size for batch processors. Queue jobs expose processing threads, lease time, and batch size for batch jobs. Empty fields restore the implementation's defaults; Use code defaults resets all overrides.

Overrides are stored in individual nullable columns on the queue metadata table and propagate to all instances. A null column uses the implementation's code default. Active queue jobs finish with their existing settings and use new values on the next run. Distributed processors finish their current work and rebuild their worker pools before leasing more work with new settings. A busy queue does not need to become empty. Lease-time changes affect future leases; existing lease expiry times remain unchanged.

The API exposes GET and POST /api/queues/{queueName}/processing-settings. Reads include defaults, saved overrides, effective values, and whether running processor instances still report older settings. POST replaces all saved overrides; omitted or null fields use code defaults, and posting {} resets every override. Counts and lease durations must be positive. Custom processing loops can read metadata explicitly; these controls apply automatically to built-in executors.

The legacy Admin UI retains its existing functionality and does not expose these new controls. See processing settings for the full override API and reset semantics.

Separate producers and broker ingress

An API-only producer can share the message CLR contract and queue names with the Nexus host and choose a name dynamically through IApiEnqueuer<LanguageMessage>:

await apiEnqueuer.EnqueueAsync("language_" + language, message);

Register API enqueuers for the contract assembly as usual. No per-name registration, local queue storage, processors, or local virtual transformers are required. The host validates the named destination. Enqueueing to a remote virtual queue runs its transformers on the host.

Azure Service Bus, RabbitMQ, and Kafka ingress registrations accept an optional nexusQueueName argument. It selects a concrete destination or a virtual entry point; message-dependent routing belongs in the virtual transformer. Existing ordinary type-based ingress remains supported. External-pending storage continues to isolate broker resources by the concrete queue name.