Lyo.MessageQueue 3.0.0

Lyo.MessageQueue

IMqService defines the queue and exchange contract. Schedulers, workers, and gateways compile against one interface and swap RabbitMQ or later brokers behind Lyo.MessageQueue.* implementations.

Implements Lyo.Health.IHealth so dashboards can ping broker connectivity alongside DB/cache checks.

IMqService

• ConnectAsync / DisconnectAsync open and close a session.
• IsConnected is a synchronous snapshot used as a guard.
• CreateQueue / CreateExchange declare topology. Brokers treat matching re-declares as no-ops.

Message envelopes (QueueMessageEnvelope<T>)

QueueMessageEnvelope<T> holds Payload, RequeueCount, MessageId, EnqueuedAt, TraceId, and Version alongside the payload. The internal QueueWorkerHelpers.DeserializeMessage<T> detects JSON shaped like { Payload, RequeueCount, … }. Bare DTO JSON is not accepted.

• Attach RequeueCount / identifiers / timestamps without wrapping every caller manually.

MessageProcessingExceptionHandling (IgnoreAndRemoveFromQueue, ThrowAndRemoveFromQueue, RequeueOnException) is the shared enum implementations expose for tuning how thrown exceptions in message handlers map onto ack/nack/requeue.

QueueWorkerBase

• Implements IHostedService + IDisposable + IHealth. StartAsync connects when needed and calls SubscribeToQueue. StopAsync cancels, then waits up to DrainTimeoutMs (default 30_000 ms) for in-flight messages, then returns.
• Parses messages with the envelope-aware DeserializeMessage helper.
• Executes your abstract DoWorkAsync(TRequest, CancellationToken) → Task<TResult>.
• Applies requeue heuristics: an optional Metadata["requeue"] bool on the result overrides the default !IsSuccess requeue rule.
• Supports maxRequeueCount + optional DLQ publish (dlqName); when the count is exceeded, original message bytes go to the DLQ if one is configured; otherwise the message is dropped at Error level.
• Optional retry backoff through the public RequeueDelay property: when set and the transport implements IDelayedMqService, each counted requeue is republished with a broker-side delay of RequeueDelay × attempt (linear backoff), so a failing message cannot burn through its retry budget in milliseconds. Transports without delay support republish immediately.

Retry path for envelopes

Each failure path acks the original delivery and republishes a counted copy. A bad message or a repeatedly-throwing DoWorkAsync cannot loop forever on broker redelivery:

flowchart LR
    msg[Message delivered] --> des{Deserialize\nautocorrect ladder}
    des -->|unrecoverable| poison[Ack + forward original bytes to DLQ]
    des -->|ok| work[DoWorkAsync]
    work -->|success| ack[Ack]
    work -->|failure / exception| cap{RequeueCount < max?}
    cap -->|yes| requeue["Ack + republish with RequeueCount+1\n(delayed by RequeueDelay × attempt when supported)"]
    requeue --> msg
    cap -->|no| dlq[Ack + route to DLQ or drop]

QueueWorkerOptions

Shared defaults come from DI registration paths (for example AddJobWorker / AddJobWorkerFromConfiguration, section name "QueueWorkerOptions"). The QueueWorkerBase constructor signature is unchanged:

• Tracks InFlightCount, publishes a queue-worker:{QueueName} health probe via CheckHealthAsync, and emits metrics via the injected IMetrics:
  • queue.worker.message.processing.duration (timer; tag queue)
  • queue.worker.messages.received / processed / requeued / deserialization.failed / dropped.max_requeue / dlq
  • queue.worker.started / start.failed / stopped
  • queue.worker.running (gauge; 1 while running, 0 after stop)
  • Error records on queue.worker.message.processing.error and queue.worker.message.deserialization.error

Lyo job and email workers use this hosted-consumer path.

Health and diagnostics data

• MqServiceHealth. Queues and Connections collections.
• MessageQueueInfo(Name, State?, Type?, Messages, MessagesReady, MessagesUnacknowledged, Consumers, AdditionalProperties). Snapshot of one queue. AMQP declare flags (durable, exclusive, auto_delete) and x-* arguments are RabbitMQ-only and sit in AdditionalProperties.
• MessageExchangeInfo(Name, Type?, Durable, AutoDelete, Internal, AdditionalProperties). Per-exchange snapshot.
• ConnectionInfo(User, UserProvidedName?, State, VHost). Snapshot of a connection.
• QueuePeekMessage(Payload, PayloadEncoding?, Exchange?, RoutingKey?, MessageCount?, Redelivered). Returned by PeekQueueMessages.

Operations

• Treat byte[] as opaque on the interface. Sign and compress at the app layer when payloads leave a trust zone.
• Idempotency. Requeue storms show up when handlers throw. Keep side effects idempotent or persist processing tokens.
• Health. Implementors should make IHealth report broker reachability. Do not report healthy while IsConnected() is false unless you want a lazy connect.

Implementations and UI

Related

• Lyo.Job.Scheduler. Often paired with queues for fan-out triggers.
• Lyo.Health. Shared health reporting.

Dependencies

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

• Lyo.Exceptions (direct, lyo)
• Lyo.Health (direct, lyo)
• Lyo.Json (direct, lyo)
• Lyo.Metrics (direct, lyo)
• Lyo.Result (direct, lyo)
• Microsoft.Extensions.Hosting.Abstractions 10.0.5 (direct, microsoft)
• Lyo.Common.Core (transitive, lyo)
• Lyo.DataTable.Models (transitive, lyo)
• Microsoft.Bcl.AsyncInterfaces 10.0.5 (transitive, microsoft, netstandard2.0)
• System.Memory 4.6.3 (transitive, microsoft, netstandard2.0)
• System.Text.Encoding.CodePages 10.0.5 (transitive, microsoft)
• System.Text.Json 10.0.5 (transitive, microsoft, netstandard2.0)