Lyo.MessageQueue.RabbitMq 2.0.0

Lyo.MessageQueue.RabbitMq

IMqService implementation (RabbitMqService) on RabbitMQ.Client. Also wired as IRabbitMqService when RabbitMQ-specific methods are needed (exchanges) that sit outside the shared contract.

Features

• RabbitMqOptions singleton (registered via an explicit Action<RabbitMqOptions> or bound from configuration). Default section: RabbitMqOptions.SectionName = "RabbitMqOptions".
• IConnectionFactory singleton built from those options, with:
• Host, virtual host, port, and credentials from options.
• ClientProvidedName set to MachineName - ApplicationName (EnvironmentName) so the RabbitMQ management UI can identify the connection.
• ClientProperties filled from the connectionProperties dictionary you pass to the extension (container id, build sha, and similar keys).
• RabbitMqService wired as a singleton, exposed as all three types: itself, IRabbitMqService, and IMqService.

Examples

Add services

services.SetupRabbitMqServiceFromConfiguration(
    builder.Configuration,
    connectionProperties: new Dictionary<string, object?> { ["build_sha"] = buildSha });
// or
services.SetupRabbitMqService(
    connectionProperties: [],
    options =>
    {
        options.Host = "rabbit.internal";
        options.Port = 5672;
        options.VirtualHost = "/";
        options.AdminUrl = "http://rabbit.internal:15672";
        options.Username = "...";
        options.Password = "...";
    });

Registration

• RabbitMqOptions singleton (registered via an explicit Action<RabbitMqOptions> or bound from configuration). Default section: RabbitMqOptions.SectionName = "RabbitMqOptions".
• IConnectionFactory singleton built from those options, with:
• Host, virtual host, port, and credentials from options.
• ClientProvidedName set to MachineName - ApplicationName (EnvironmentName) so the RabbitMQ management UI can identify the connection.
• ClientProperties filled from the connectionProperties dictionary you pass to the extension (container id, build sha, and similar keys).
• RabbitMqService wired as a singleton, exposed as all three types: itself, IRabbitMqService, and IMqService.

RabbitMqOptions

Concurrency per queue

SubscribeToQueue looks up the queue's limit (QueueProcessingLimits[queue], falling back to ProcessingLimit; 0 = unlimited) and enforces it at three levels:

1. Broker prefetch. BasicQosAsync(0, limit, false) on the dedicated subscription channel, so the broker only delivers limit unacked messages to this consumer. Real backpressure: excess messages stay on the server and are available to other consumers.
2. Channel dispatch concurrency. The subscription channel is created with consumerDispatchConcurrency = limit, so the client actually runs up to limit handler invocations in parallel (the RabbitMQ.Client 7.x default is 1, i.e. strictly sequential).
3. In-process semaphore. A SemaphoreSlim(limit) guards handler execution as a final in-process guarantee.

"RabbitMqOptions": {
  "ProcessingLimit": 5,
  "QueueProcessingLimits": { "job.run.cs": 1, "job.run.reports": 10 }
}

flowchart LR
    broker[RabbitMQ broker] -->|"prefetch = per-queue limit"| channel[Dedicated channel]
    channel -->|"dispatch concurrency = limit"| handler[Handler xN]
    handler -->|ack / counted requeue| broker
    conn[Connection drop] -->|automatic recovery| broker

Recovering connections

When AutomaticRecovery is on (the default), the RabbitMQ client reconnects after a network drop, re-opens channels, re-declares topology, and restores consumers. The service keeps its consumer bookkeeping across the drop and logs/metrics the transition (mq.connection.lost then mq.connection.recovered). With recovery disabled, a lost connection clears all consumers. The process must reconnect and resubscribe itself. DisconnectAsync no longer poisons the instance: connect, disconnect, connect on the same RabbitMqService works. Disposal (DisposeAsync) is final.

Deferred delivery

SendToQueueDelayed(queueName, data, delay) (on IRabbitMqService, also surfaced through the IDelayedMqService capability interface in Lyo.MessageQueue) delivers a message after a delay with no broker plugin. The message is published to a companion wait queue {queue}.wait declared with x-dead-letter-exchange: "" / x-dead-letter-routing-key: {queue} and a per-message TTL equal to the delay. When the TTL fires, the broker dead-letters the message onto the real queue. Wait-queue declarations are cached per service instance. QueueWorkerBase uses this on its own for retry backoff when its RequeueDelay is set (see the Lyo.MessageQueue README).

Note: TTL expiry is FIFO on each wait queue. A long-delay message queued ahead of a short-delay one delays the latter. For the retry-backoff use case (delays in the same order of magnitude) this is fine.

Broker DLQ wiring

CreateQueueWithDlq(queueName, durable, dlqName, arguments, ct) declares {queue}.dlq (durable) plus the main queue with x-dead-letter-exchange: "" / x-dead-letter-routing-key: {queue}.dlq. This catches broker-side rejections the application never sees: nack without requeue, per-queue TTL expiry, and queue overflow. It complements (and does not replace) QueueWorkerBase's application-level DLQ routing. > Caveat: RabbitMQ cannot change arguments on an existing queue. Declaring over an existing queue with different arguments fails with PRECONDITION_FAILED. The helper logs a clear error that tells you to delete and recreate the queue.

Queue stats

GetQueueInfoAsync(queueName, ct) and GetAllQueuesInfoAsync(ct) (on IRabbitMqService) hit the Management API (GET /api/queues/{vhost}[/{name}]) and return the shared MessageQueueInfo record. Message counts, ready/unacked, consumer count, and state are first-class. AMQP flags (durable, exclusive, auto_delete), x-* arguments (DLQ, max-priority), and publish/deliver rates land in AdditionalProperties (see Constants.QueueInfoProperties). GetExchangeInfoAsync / GetAllExchangesInfoAsync do the same for exchanges (GET /api/exchanges/{vhost}[/{name}]). The RabbitMqWorkbench Blazor component lists queues and exchanges with these flags as chips. Statistics refresh on the management emission interval (~5s), so counts can lag slightly behind broker state.

Capabilities

Rabbit features evolve quickly, so (streams, quorum queues), advanced work tends to go through CreateQueue's arguments bag. Check RabbitMqService for the defaults you rely on before you upgrade RabbitMQ.Client.

Workers and hosted services

Use alongside QueueWorkerBase from Lyo.MessageQueue. Workers deserialize JSON payloads, reuse the envelope helpers, integrate DLQ / maxRequeueCount, and drain on IHostedService shutdown. Publish typed payloads with IMqService.SendToQueueWithEnvelopeAsync when a QueueWorkerBase consumer is on the other end. Schedulers (Lyo.Job.Scheduler) commonly publish triggers here while separate worker processes consume.

Blazor UI

Lyo.MessageQueue.RabbitMq.Web.Components adds UI on the same service registrations for internal dashboards.

Testing

Lyo.MessageQueue.RabbitMq.Tests exercises the service against a real broker using the RabbitMqTestContainer helper from Lyo.Testing.Containers (management-enabled image, Docker required): connect/disconnect/reconnect, queue lifecycle + peek, publish→subscribe roundtrips, requeue/ack semantics, per-queue concurrency enforcement, publisher confirms, delayed delivery, DLQ auto-wiring, and queue statistics.

Dependencies

Generated from ProjectReference / PackageReference (same model as docs/Lyo.ProjectGraph.html).

• Lyo.Common.Json (direct, lyo)
• Lyo.Common.Metadata (direct, lyo)
• Lyo.Exceptions (direct, lyo)
• Lyo.MessageQueue (direct, lyo)
• Lyo.Metrics (direct, lyo)
• Microsoft.Extensions.Configuration.Binder 10.0.5 (direct, microsoft)
• RabbitMQ.Client 7.2.1 (direct, third-party)
• System.Text.Json 10.0.5 (direct, microsoft, netstandard2.0)
• Lyo.Common.Core (transitive, lyo)
• Lyo.Health (transitive, lyo)
• Lyo.Result (transitive, lyo)
• Microsoft.Bcl.AsyncInterfaces 10.0.5 (transitive, microsoft, netstandard2.0)
• Microsoft.Extensions.Hosting.Abstractions 10.0.5 (transitive, microsoft)
• System.Memory 4.6.3 (transitive, microsoft, netstandard2.0)