Lyo.MessageQueue.RabbitMq 1.0.6

Lyo.MessageQueue.RabbitMq

IMqService implementation (RabbitMqService) on RabbitMQ.Client. Also registered as IRabbitMqService when you need RabbitMQ-specific methods (exchanges) that are not part of the shared contract.

Features

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

Examples

Register 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 name: RabbitMqOptions.SectionName = "RabbitMqOptions".
• IConnectionFactory singleton built from those options with:
• Host / virtual host / port / credentials from options.
• ClientProvidedName set to MachineName - ApplicationName (EnvironmentName) so the connection is identifiable in the RabbitMQ management UI.
• ClientProperties populated from the connectionProperties dictionary you pass to the extension (container id, build sha, and similar keys).
• RabbitMqService registered as a singleton, exposed under all three types: itself, IRabbitMqService, and IMqService.

RabbitMqOptions

Per-queue concurrency

SubscribeToQueue resolves 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

Connection recovery

With AutomaticRecovery 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.

Delayed messages

SendToQueueDelayed(queueName, data, delay) (on IRabbitMqService, also exposed 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 automatically for retry backoff when its RequeueDelay is set (see the Lyo.MessageQueue README).

Note: TTL expiry is FIFO per 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-level DLQ auto-wiring

CreateQueueWithDlq(queueName, durable, dlqName, arguments, ct) declares {queue}.dlq (durable) and 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 (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 telling you to delete and recreate the queue.

Queue statistics

GetQueueInfoAsync(queueName, ct) and GetAllQueuesInfoAsync(ct) (on IRabbitMqService) query the Management API (GET /api/queues/{vhost}[/{name}]) and return the shared MessageQueueInfo record. Message counts, ready/unacked, consumer count, state, plus publish/deliver rates in AdditionalProperties when the broker reports them. The RabbitMqWorkbench Blazor component shows these in its Stats tab. Statistics update on the management emission interval (~5s), so counts can lag slightly behind broker state.

Capabilities

Because Rabbit features evolve quickly (streams, quorum queues), anything advanced tends to go through CreateQueue's arguments bag. Inspect RabbitMqService for the defaults you rely on before upgrading RabbitMQ.Client.

Hosted services and workers

Pair with 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 tooling

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

Testing

Lyo.MessageQueue.RabbitMq.Tests runs 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.

See also

• Lyo.MessageQueue. The underlying contract, envelopes, and worker base.
• Lyo.Result. Worker results and the Metadata["requeue"] pattern.
• Lyo.Metrics. Counters and timers emitted by the service and QueueWorkerBase.
• Lyo.Testing.Containers. RabbitMQ testcontainer and fixture for integration tests.

Dependencies

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

• Lyo.Common (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.Health (transitive, lyo)
• Lyo.Result (transitive, lyo)
• Microsoft.Extensions.DependencyInjection.Abstractions 10.0.5 (transitive, microsoft)
• Microsoft.Extensions.Hosting.Abstractions 10.0.5 (transitive, microsoft)
• Microsoft.Extensions.Logging.Abstractions 10.0.5 (transitive, microsoft)
• Microsoft.Extensions.Options.ConfigurationExtensions 10.0.5 (transitive, microsoft)
• System.Memory 4.6.3 (transitive, microsoft, netstandard2.0)