Lyo.MessageQueue.RabbitMq 1.0.1

Lyo.MessageQueue.RabbitMq

Concrete IMqService (RabbitMqService) using RabbitMQ.Client, also surfaced as IRabbitMqService when you need RabbitMQ-specific knobs (exchanges) that are not part of the shared abstraction.

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 (rich connection metadata — container id, build sha, etc.).
• 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 (rich connection metadata — container id, build sha, etc.).
• 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 transparently 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 → 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 surfaces these in its Stats tab. Statistics update on the management emission interval (~5s), so counts can lag slightly behind broker state.

Capabilities mapping

Because Rabbit features evolve quickly (streams, quorum queues), anything advanced tends to funnel 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 gracefully on IHostedService shutdown. The publish helper IMqService.SendToQueueWithEnvelopeAsync is the recommended way to publish typed payloads to a queue consumed by such a worker. Schedulers (Lyo.Job.Scheduler) commonly publish triggers here while separate worker processes consume.

Blazor tooling

Lyo.MessageQueue.RabbitMq.Web.Components layers a UI on top of 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 + 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)