ND.FW.QueueManagement 1.0.0

ND.FW.QueueManagement

Provider-agnostic message publish/consume abstraction for Novacis Digital services.

Business code depends only on IQueuePublisher, IQueueConsumer, and IQueueProviderFactory — never on ActiveMQ or Azure Service Bus types directly. The active provider is selected via configuration (Messaging:Provider), so switching providers is a config change, not a code change.

Supported providers:

• ActiveMQ (via MassTransit) — primary
• Azure Service Bus — supported, credential-resolved through ND.FW.ClientManagement

Install

dotnet add package ND.FW.QueueManagement

Depends on ND.FW.ClientManagement (for Service Bus credential resolution). NuGet restores it transitively.

Register

using QueueManagement.Extensions;

// Registers both provider implementations + IQueueProviderFactory,
// bound from configuration sections "ActiveMQ" and "ServiceBus"
services.AddQueueConsumer(configuration);

// Starts the background worker that drives message consumption
services.AddQueueConsumerWorker();

Or register a single provider explicitly if you don't need the factory:

services.AddActiveMQ(options =>
{
    options.BrokerUri = "activemq:tcp://broker:61616";
    options.QueueName = "intake-events";
});

// — or —

services.AddAzureServiceBus(options =>
{
    options.FullyQualifiedNamespace = "mynamespace.servicebus.windows.net";
    options.QueueName = "intake-events";
    options.CredentialType = CredentialType.ManagedIdentity; // from ND.FW.ClientManagement
});

Publishing

public class MyPublisher
{
    private readonly IQueuePublisher _publisher;

    public MyPublisher(IQueueProviderFactory factory)
    {
        _publisher = factory.GetPublisher("servicebus"); // or "activemq"
    }

    public Task PublishAsync(MyEvent evt) =>
        _publisher.PublishAsync(evt, "intake-events");
}

PublishAsync<T> serialises T to JSON and injects trace correlation headers. PublishBatchAsync<T> is available for high-throughput scenarios.

Consuming

public class MyMessageHandler : IMessageHandler
{
    public async Task HandleAsync(QueueMessageContext context, CancellationToken ct)
    {
        // deserialize context.Body, do work...
        await context.CompleteAsync();      // ack — removes from queue
        // await context.AbandonAsync();    // negative-ack — redeliver
        // await context.DeadLetterAsync("reason"); // move to DLQ
    }
}

QueueConsumerWorker (registered via AddQueueConsumerWorker()) resolves all IMessageHandler registrations and invokes them in registration order for each incoming message. Register your handler as scoped:

services.AddScoped<IMessageHandler, MyMessageHandler>();

API

QueueMessageContext

public sealed record QueueMessageContext
{
    string MessageId;
    string Body;
    IReadOnlyDictionary<string, object> Properties;
    Func<Task> CompleteAsync;
    Func<Task> AbandonAsync;
    Func<string?, Task> DeadLetterAsync;
}

Configuration

{
  "ActiveMQ": {
    "BrokerUri": "tcp://broker:61616",
    "QueueName": "intake-events",
    "Username": "",
    "Password": ""
  },
  "ServiceBus": {
    "FullyQualifiedNamespace": "mynamespace.servicebus.windows.net",
    "QueueName": "intake-events",
    "MaxConcurrentCalls": 5,
    "CredentialType": "ManagedIdentity"
  }
}

Dependencies

• MassTransit / MassTransit.ActiveMQ 8.2.3
• Azure.Messaging.ServiceBus 7.18.0
• ND.FW.ClientManagement (>= 1.0.0) — required for Service Bus credential resolution. Must be published before this package.

Publish order

ND.FW.ClientManagement  →  ND.FW.QueueManagement

License

MIT