QsMessaging-develop 1.0.265

QsMessaging

QsMessaging is a .NET 8 library designed for sending and receiving messages between services or components of your application using RabbitMQ or Azure Service Bus. It supports horizontal scalability, allowing multiple instances of the same service to handle messages efficiently.
Available on NuGet for seamless integration:

Project website:
https://pavlo-0.github.io/QsMessaging/

A simple, scalable messaging solution for distributed systems.

Note: Azure Service Bus support is in an early, not fully tested or implemented state. The API is the same as for RabbitMQ, but some features may be missing or behave unexpectedly.

Installation

Install the package using the following command:

dotnet add package QsMessaging

Registering the Library

Registering the library is simple. Add the following two lines of code to your Program.cs:

// Add QsMessaging (use the default configuration)...
builder.Services.AddQsMessaging(options => { });

...
await host.UseQsMessaging();

Handler Discovery

By default, QsMessaging scans the entry assembly and the assembly that called AddQsMessaging. If your handlers live in separate class libraries, pass those assemblies explicitly:

builder.Services.AddQsMessaging(options =>
{
    options.AssembliesToScan.Add(typeof(MyMessageHandler).Assembly);
});

// or
builder.Services.AddQsMessaging(options => { }, typeof(MyMessageHandler).Assembly);

When scan assemblies are explicitly configured and no QsMessaging consumer handlers are found, registration fails fast with an InvalidOperationException.

Default Configuration

RabbitMQ (default transport)

• Host: localhost
• UserName: guest
• Password: guest
• Port: 5672

Custom RabbitMQ Configuration

using Polly;

builder.Services.AddQsMessaging(options =>
{
    options.HandlerResilience.MaxRetryAttempts = 1;
    options.HandlerResilience.Delay = TimeSpan.FromSeconds(1);
    options.Resilience.MaxRetryAttempts = 3;
    options.Resilience.Delay = TimeSpan.FromSeconds(1);
    options.Resilience.BackoffType = DelayBackoffType.Constant;
    options.Resilience.UseJitter = false;
    options.RabbitMQ.Host = "my-rabbitmq-host";
    options.RabbitMQ.UserName = "myuser";
    options.RabbitMQ.Password = "mypassword";
    options.RabbitMQ.Port = 5672;
});

Send Resilience And Missing Receivers

SendMessageAsync does not run a queue/subscription existence check before every message. That keeps the hot send path fast and avoids a management call per publish.

If the transport reports that a message cannot be delivered because the receiver entity is missing, QsMessaging retries the send with a Polly resilience pipeline and then logs a warning instead of throwing.

options.Resilience is shared by RabbitMQ and Azure Service Bus send paths.

Shared resilience options:

RabbitMQ behavior:

• Durable message publishing uses publisher confirms/return tracking and mandatory publishing.
• If RabbitMQ returns a normal message as unroutable, QsMessaging retries according to options.Resilience.
• After retries are exhausted, the send is swallowed and a warning is logged.

Azure Service Bus behavior:

• QsMessaging does not check subscriptions before every topic send.
• If Azure Service Bus throws ServiceBusException with MessagingEntityNotFound, QsMessaging retries according to options.Resilience.
• After retries are exhausted, the send is swallowed and a warning is logged.
• Azure Service Bus usually accepts a send to an existing topic even when the topic has no subscriptions, so that specific case cannot be detected after send without a management check.

Handler Resilience And Error Handlers

When a user handler throws, QsMessaging retries that handler before calling IQsMessagingConsumerErrorHandler.

Configure handler retry on the root options object:

using Polly;

builder.Services.AddQsMessaging(options =>
{
    options.HandlerResilience.MaxRetryAttempts = 1;
    options.HandlerResilience.Delay = TimeSpan.FromSeconds(1);
    options.HandlerResilience.BackoffType = DelayBackoffType.Constant;
    options.HandlerResilience.UseJitter = false;
});

HandlerResilience uses the same option shape as send resilience:

If a retry succeeds, QsMessaging does not call consumer error handlers. If all attempts fail, QsMessaging calls each registered IQsMessagingConsumerErrorHandler once with ErrorConsumerType.InHandlerProblem.

Transport Cleanup Helpers

For debug or local reset scenarios you can explicitly clean transport entities before starting consumers again:

await host.CleanUpTransportation();
await host.FullCleanUpTransportation();
await host.UseQsMessaging();

• CleanUpTransportation() removes entities that QsMessaging can derive from the current app contracts.
• FullCleanUpTransportation() removes everything visible in the configured transport scope.
• For RabbitMQ, full cleanup uses the Management HTTP API for the configured virtual host.
• RabbitMQ Management HTTP API calls use Microsoft.Extensions.Http.Resilience with the configured Resilience retry settings for transient HTTP failures.

RabbitMQ full cleanup configuration:

builder.Services.AddQsMessaging(options =>
{
    options.RabbitMQ.Host = "localhost";
    options.RabbitMQ.VirtualHost = "/";
    options.RabbitMQ.ManagementPort = 15672;
    options.RabbitMQ.ManagementScheme = "http";
});

Azure Service Bus Support (Early Preview)

Azure Service Bus support is available but is not fully tested or implemented. The public interface (IQsMessaging) is identical to RabbitMQ — no code changes are needed in your handlers or senders.

Registering with Azure Service Bus

Set options.Transport = QsMessagingTransport.AzureServiceBus and supply a connection string:

builder.Services.AddQsMessaging(options =>
{
    options.Transport = QsMessagingTransport.AzureServiceBus;
    options.AzureServiceBus.ConnectionString = "<your-connection-string>";
    options.Resilience.MaxRetryAttempts = 3;
    options.Resilience.Delay = TimeSpan.FromSeconds(1);
});

...
await host.UseQsMessaging();

Configuration for Cloud (Azure)

Use your Azure Service Bus namespace connection string directly:

builder.Services.AddQsMessaging(options =>
{
    options.Transport = QsMessagingTransport.AzureServiceBus;
    options.AzureServiceBus.ConnectionString =
        "Endpoint=sb://your-namespace.servicebus.windows.net/;" +
        "SharedAccessKeyName=RootManageSharedAccessKey;" +
        "SharedAccessKey=YOUR_KEY;";
});

Configuration for Emulator (Local Development)

The Azure Service Bus Emulator uses separate ports for AMQP (messaging) and the management API:

builder.Services.AddQsMessaging(options =>
{
    options.Transport = QsMessagingTransport.AzureServiceBus;
    options.AzureServiceBus.ConnectionString =
        "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;";
    options.AzureServiceBus.EmulatorAmqpPort = 5672;
    options.AzureServiceBus.EmulatorManagementPort = 5300;
});

Usage

Sending Messages

Contract

Define a message contract:

public class RegularMessageContract
{
    public required string MyTextMessage { get; set; }
}

Sending a Message

Inject IQsMessaging into your class:

public YourClass(IQsMessaging qsMessaging) {}

Then, use it to send a message:

await qsMessaging.SendMessageAsync(new RegularMessageContract { MyTextMessage = "My message." });

Handling Messages

To handle the message, create a handler:

public class RegularMessageContractHandler : IQsMessageHandler<RegularMessageContract>
{
    public Task<bool> Consumer(RegularMessageContract contractModel)
    {
        // Process the message here
        return Task.FromResult(true);
    }
}

All handlers discovered by QsMessaging are registered in DI as Transient. This means each message/request is handled by a fresh handler instance, and handlers have full support for constructor injection of your application services.

What Happens If A Handler Throws An Exception?

If your handler throws an exception, QsMessaging catches it, retries the handler according to options.HandlerResilience, and only then calls error handlers if all attempts fail.

• The exception does not crash the consumer loop.
• The handler is retried according to options.HandlerResilience.
• If a retry succeeds, consumer error handlers are not called.
• If all handler attempts fail, the exception is forwarded to your custom error handler(s) together with message metadata.

Current behavior by transport:

• RabbitMQ: consumers use automatic acknowledge mode, so the message is treated as acknowledged even if the handler fails.
• Azure Service Bus: after the handler pipeline finishes, QsMessaging completes the message, so it is not re-delivered automatically.

If you need dead-letter, alerting, or custom logging after handler retries are exhausted, implement IQsMessagingConsumerErrorHandler and handle the exception there.

Short Operational Notes

• Queue/exchange naming: RabbitMQ uses names like Qs:{FullTypeName}:ex for exchanges and Qs:{FullTypeName}:permanent for durable queues. Azure Service Bus uses Qs-Queue-{FullTypeName} and Qs-Topic-{FullTypeName}. Long names are hashed.
• Send retry: ordinary message sends retry only after the transport reports a missing/unroutable receiver. There is no per-message pre-check of queue or subscription existence.
• Handler retry / dead-letter: handlers are retried with HandlerResilience before error handlers run. There is currently no built-in dead-letter flow for failed message handlers.
• Azure Service Bus TTL: normal SendMessageAsync messages use a 14 day TTL. Events use a 60 second TTL.
• Multiple instances of one consumer: for IQsMessageHandler<T>, instances compete on one shared queue, so one message is processed by one instance. For IQsEventHandler<T>, each instance gets its own temporary queue/subscription, so every instance receives the event.
• Unhappy path: if a handler still throws after configured retries, the exception is sent to IQsMessagingConsumerErrorHandler.
• Request/response: default timeout is 50000 ms. If no response arrives in time, the request fails with TimeoutException. Correlation ID is generated automatically per request as a new Guid string and copied to the response. Cancellation token is passed into transport operations, but timeout is the main response wait guard. Duplicate responses are not specially deduplicated by the library; late responses are ignored after the request is removed from the local store.

Example: Custom Error Handler

public class MessagingErrorHandler : IQsMessagingConsumerErrorHandler
{
    private readonly ILogger<MessagingErrorHandler> _logger;

    public MessagingErrorHandler(ILogger<MessagingErrorHandler> logger)
    {
        _logger = logger;
    }

    public Task HandleErrorAsync(Exception exception, ErrorConsumerDetail detail)
    {
        _logger.LogError(
            exception,
            "Handler failed. Queue or entity: {QueueName}, Handler: {HandlerType}, Payload type: {PayloadType}",
            detail.QueueName,
            detail.HandlerType,
            detail.GenericType);

        // Add your own logic here:
        // - save to database
        // - send alert
        // - push to dead-letter queue
        // - trigger retry workflow

        return Task.CompletedTask;
    }
}

Request/Response Pattern

You can also use the Request/Response pattern to send a request and await a response. This is useful when you need to communicate between services and expect a response.

Request/Response Contract

Define the request and response contracts:

public class MyRequestContract
{
    public required string RequestMessage { get; set; }
}

public class MyResponseContract
{
    public required string ResponseMessage { get; set; }
}

Sending a Request and Receiving a Response

To send a request and await a response, use the RequestResponse<TRequest, TResponse>:

public class MyService
{
    private readonly IQsMessaging _qsMessaging;

    public MyService(IQsMessaging qsMessaging)
    {
        _qsMessaging = qsMessaging;
    }

    public async Task<MyResponseContract> SendRequestAsync(MyRequestContract request)
    {
        var response = await _qsMessaging.SendRequestResponseAsync<MyRequestContract, MyResponseContract>(request);
        return response;
    }
}

Handling Requests

To handle requests, implement the IQsRequestResponseHandler<TRequest, TResponse> interface:

public class MyRequestHandler : IQsRequestResponseHandler<MyRequestContract, MyResponseContract>
{
    public Task<MyResponseContract> Handle(MyRequestContract request)
    {
        // Process the request and create a response
        return Task.FromResult(new MyResponseContract { ResponseMessage = "Response to: " + request.RequestMessage });
    }
}

Dependency Injection Examples

The examples below show how handlers can consume dependencies through constructor injection.

1) Message Handler with Injected Services

public interface IOrderProcessor
{
    Task ProcessAsync(CreateOrderMessage message);
}

public class CreateOrderMessage
{
    public required string OrderId { get; set; }
}

public class CreateOrderMessageHandler : IQsMessageHandler<CreateOrderMessage>
{
    private readonly IOrderProcessor _orderProcessor;
    private readonly ILogger<CreateOrderMessageHandler> _logger;

    public CreateOrderMessageHandler(
        IOrderProcessor orderProcessor,
        ILogger<CreateOrderMessageHandler> logger)
    {
        _orderProcessor = orderProcessor;
        _logger = logger;
    }

    public async Task<bool> Consumer(CreateOrderMessage contractModel)
    {
        _logger.LogInformation("Processing order {OrderId}", contractModel.OrderId);
        await _orderProcessor.ProcessAsync(contractModel);
        return true;
    }
}

2) Request/Response Handler with Injected Repository

public interface IUserRepository
{
    Task<UserDto?> GetByIdAsync(Guid id);
}

public class GetUserRequest
{
    public Guid UserId { get; set; }
}

public class GetUserResponse
{
    public string? Name { get; set; }
    public bool Found { get; set; }
}

public class GetUserHandler : IQsRequestResponseHandler<GetUserRequest, GetUserResponse>
{
    private readonly IUserRepository _userRepository;

    public GetUserHandler(IUserRepository userRepository)
    {
        _userRepository = userRepository;
    }

    public async Task<GetUserResponse> Handle(GetUserRequest request)
    {
        var user = await _userRepository.GetByIdAsync(request.UserId);
        return new GetUserResponse
        {
            Found = user is not null,
            Name = user?.Name
        };
    }
}

Documentation

For detailed documentation, visit the QsMessaging Wiki.

That's all, folks!