< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.Transports.MongoDB.MongoDbSubscriberService
Assembly: AsyncResponse.Transports.MongoDB
File(s): /_/src/Transports/AsyncResponse.Transports.MongoDB/MongoDbSubscriberServices.cs
Line coverage
100%
Covered lines: 91
Uncovered lines: 0
Coverable lines: 91
Total lines: 247
Line coverage: 100%
Branch coverage
100%
Covered branches: 22
Total branches: 22
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_Options()100%11100%
get_Logger()100%11100%
StartAsync(...)100%11100%
ExecuteAsync()100%22100%
RunSubscriberAsync()100%1414100%
ListenLoopAsync()100%22100%
WaitForSignalOrDelayAsync()100%44100%

File(s)

/_/src/Transports/AsyncResponse.Transports.MongoDB/MongoDbSubscriberServices.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using System.Threading.Channels;
 5
 6namespace AsyncResponse.Transports.MongoDB;
 7
 8/// <summary>
 9/// Base hosted service that consumes one MongoDB queue and routes documents to AsyncResponse ingress
 10/// with configured acknowledgement, redelivery, and dead-letter behavior.
 11/// </summary>
 12internal abstract class MongoDbSubscriberService : BackgroundService
 13{
 14    private readonly MongoDbTransportStore _store;
 42215    private readonly Channel<bool> _signals = Channel.CreateBounded<bool>(new BoundedChannelOptions(1)
 42216    {
 42217        SingleReader = true,
 42218        SingleWriter = false,
 42219        FullMode = BoundedChannelFullMode.DropWrite
 42220    });
 21
 42222    protected MongoDbSubscriberService(
 42223        IOptions<MongoDbAsyncResponseTransportOptions> options,
 42224        MongoDbTransportStore store,
 42225        ILogger logger)
 26    {
 42227        Options = options.Value;
 42228        MongoDbTransportOptionsValidator.ValidateCommon(Options);
 42229        _store = store;
 42230        Logger = logger;
 42231    }
 32
 982733    protected MongoDbAsyncResponseTransportOptions Options { get; }
 83034    protected ILogger Logger { get; }
 35
 36    protected abstract string Queue { get; }
 37    protected abstract MongoDbSubscriberOptions SubscriberOptions { get; }
 38    protected abstract MongoDbSubscriberRole Role { get; }
 39    protected abstract Task HandleMessageAsync(MongoDbTransportDelivery delivery, CancellationToken cancellationToken);
 40
 41    /// <inheritdoc />
 42    /// <summary>
 43    /// Validates subscriber options here rather than at the top of <c>ExecuteAsync</c>: since
 44    /// Microsoft.Extensions.Hosting.Abstractions 10.0.10, <c>BackgroundService.StartAsync</c> no
 45    /// longer runs <c>ExecuteAsync</c> inline, so a throw there surfaces only through the host's
 46    /// background-exception handling — or never, when a fast stop discards the queued work —
 47    /// instead of failing host startup synchronously.
 48    /// </summary>
 49    public override Task StartAsync(CancellationToken cancellationToken)
 50    {
 40651        MongoDbTransportOptionsValidator.ValidateSubscriber(Options, SubscriberOptions, Role.ToString());
 40452        return base.StartAsync(cancellationToken);
 53    }
 54
 55    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 56    {
 57        // ONE dispatcher for the service's lifetime, outliving every supervised attempt below; only
 58        // the host stopping disposes it. In early-ACK mode it owns the queued and running work
 59        // whose queue items the ACK already deleted, and its DisposeAsync IS the stop-time drain:
 60        // wait out BackgroundDrainTimeout, then cancel and dead-letter whatever is still queued.
 61        // Built inside the attempt, every poll fault — a claim timeout, a deadlock victim, a
 62        // failover: routine for a loop that polls the database several times a second — ran that
 63        // drain on a host that was NOT stopping: consumption paused for the whole budget, then
 64        // healthy already-ACKed work was dead-lettered as "drain budget lapsed" — or, when the
 65        // dead-letter write needed the same failing database, survived only as an Error log line.
 40466        await using var dispatcher = new MongoDbMessageDispatcher(
 40467            HandleMessageAsync,
 40468            Options,
 40469            SubscriberOptions,
 40470            Logger,
 40471            Role);
 72
 40473        await SubscriberSupervisor.RunAsync(
 40874            attemptToken => RunSubscriberAsync(dispatcher, attemptToken),
 40475            stoppingToken,
 876            failures => AsyncResponseRetry.Backoff(failures, Options.SubscriberRetryBaseDelay, Options.SubscriberRetryMa
 41077            (ex, delay) => Logger.LogWarning(ex, "MongoDB subscriber failed for queue {Queue} ({Role}); retrying in {Ret
 40278    }
 79
 80    private async Task RunSubscriberAsync(MongoDbMessageDispatcher dispatcher, CancellationToken stoppingToken)
 81    {
 41082        await _store.EnsureCreatedAsync(stoppingToken).ConfigureAwait(false);
 41083        using var signalCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
 84
 85        // The wake task starts inside the try so ANY escape — a throwing logger provider
 86        // included — runs the cancelling finally. Disposing signalCts does NOT cancel it, so an
 87        // escape before the finally would otherwise leave ListenLoopAsync parked on the change
 88        // stream, leaking a cursor per retry.
 41089        Task? listenTask = null;
 90        try
 91        {
 41092            listenTask = Options.UseChangeStreamWake
 39693                ? Task.Run(() => ListenLoopAsync(signalCts.Token), signalCts.Token)
 41094                : Task.CompletedTask;
 95
 41096            Logger.LogInformation(
 41097                "MongoDB subscriber started. Queue: {Queue}. Role: {Role}. AckMode: {AckMode}.",
 41098                Queue,
 41099                Role,
 410100                SubscriberOptions.AckMode);
 101
 1594102            while (!stoppingToken.IsCancellationRequested)
 103            {
 1466104                var claimed = 0;
 3764105                await foreach (var delivery in _store.ClaimBatchAsync(Queue, SubscriberOptions.BatchSize, Options.LockTi
 106                {
 416107                    claimed++;
 416108                    await dispatcher.HandleAsync(delivery, stoppingToken).ConfigureAwait(false);
 109                }
 110
 1400111                if (claimed > 0)
 112                    continue;
 113
 1197114                await WaitForSignalOrDelayAsync(stoppingToken).ConfigureAwait(false);
 115            }
 116        }
 117        finally
 118        {
 410119            await signalCts.CancelAsync().ConfigureAwait(false);
 410120            if (listenTask is not null)
 121            {
 122                try
 123                {
 410124                    await listenTask.WaitAsync(Options.ShutdownTimeout).ConfigureAwait(false);
 408125                }
 2126                catch (Exception ex) when (ex is OperationCanceledException or TimeoutException)
 127                {
 2128                }
 129            }
 130        }
 128131    }
 132
 133    private async Task ListenLoopAsync(CancellationToken cancellationToken)
 134    {
 135        // Retry with backoff, mirroring the channel-side listener: a transient watch failure
 136        // (network blip, replica-set stepdown) must not permanently degrade this subscriber from
 137        // push wake to poll-only latency for the rest of the process's uptime.
 396138        var failures = 0;
 402139        while (!cancellationToken.IsCancellationRequested)
 140        {
 141            try
 142            {
 400143                await _store.WatchQueueAsync(Queue, () =>
 400144                {
 351145                    _signals.Writer.TryWrite(true);
 351146                    return Task.CompletedTask;
 400147                }, cancellationToken).ConfigureAwait(false);
 2148                failures = 0;
 2149            }
 388150            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
 151            {
 388152                return;
 153            }
 8154            catch (Exception ex) when (MongoDbTransportStore.IsChangeStreamUnsupported(ex))
 155            {
 156                // Structural, not transient: a standalone server never grows change streams, so
 157                // retrying is pointless — the poll loop is the permanent delivery path here.
 2158                Logger.LogInformation(
 2159                    "MongoDB change streams are unavailable for queue {Queue} (the server is not a replica set); polling
 2160                    Queue,
 2161                    SubscriberOptions.EmptyPollDelay);
 2162                return;
 163            }
 6164            catch (Exception ex)
 165            {
 6166                failures++;
 6167                var delay = AsyncResponseRetry.Backoff(failures, Options.SubscriberRetryBaseDelay, Options.SubscriberRet
 6168                Logger.LogWarning(ex, "MongoDB change-stream wake for queue {Queue} failed; retrying in {RetryDelay} (po
 169                try
 170                {
 6171                    await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
 4172                }
 2173                catch (OperationCanceledException)
 174                {
 2175                    return;
 176                }
 177            }
 178        }
 394179    }
 180
 181    private async Task WaitForSignalOrDelayAsync(CancellationToken cancellationToken)
 182    {
 183        // The WhenAny loser is cancelled via the per-iteration linked source (mirroring the
 184        // channel-side CollectDispatchScopeAsync): an abandoned WaitToReadAsync would otherwise
 185        // stay parked in the channel's blocked-reader list until the next signal — one per empty
 186        // poll, accumulating without bound on an idle queue.
 1287187        using var iteration = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 1287188        var delay = Task.Delay(SubscriberOptions.EmptyPollDelay, iteration.Token);
 1287189        var signal = _signals.Reader.WaitToReadAsync(iteration.Token).AsTask();
 1287190        var completed = await Task.WhenAny(delay, signal).ConfigureAwait(false);
 1287191        iteration.Cancel();
 1287192        if (completed == signal)
 193        {
 503194            await signal.ConfigureAwait(false);
 578195            while (_signals.Reader.TryRead(out _))
 196            {
 197            }
 198        }
 1073199    }
 200}
 201
 202/// <summary>Consumes worker-job documents and executes them through the AsyncResponse ingress.</summary>
 203internal sealed class MongoDbWorkerSubscriber : MongoDbSubscriberService
 204{
 205    private readonly IAsyncResponseIngress _ingress;
 206
 207    public MongoDbWorkerSubscriber(
 208        IOptions<MongoDbAsyncResponseTransportOptions> options,
 209        MongoDbTransportStore store,
 210        IAsyncResponseIngress ingress,
 211        ILogger<MongoDbWorkerSubscriber> logger)
 212        : base(options, store, logger)
 213        => _ingress = ingress;
 214
 215    protected override string Queue => Options.WorkerQueue;
 216    protected override MongoDbSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 217    protected override MongoDbSubscriberRole Role => MongoDbSubscriberRole.Worker;
 218
 219    protected override Task HandleMessageAsync(MongoDbTransportDelivery delivery, CancellationToken cancellationToken)
 220        => _ingress.HandleWorkerMessageAsync(delivery.Payload);
 221}
 222
 223/// <summary>Consumes response documents and feeds them into the AsyncResponse ingress.</summary>
 224internal sealed class MongoDbResponseIngressSubscriber : MongoDbSubscriberService
 225{
 226    private readonly IAsyncResponseIngress _ingress;
 227
 228    public MongoDbResponseIngressSubscriber(
 229        IOptions<MongoDbAsyncResponseTransportOptions> options,
 230        MongoDbTransportStore store,
 231        IAsyncResponseIngress ingress,
 232        ILogger<MongoDbResponseIngressSubscriber> logger)
 233        : base(options, store, logger)
 234        => _ingress = ingress;
 235
 236    protected override string Queue => Options.ResponseQueue;
 237    protected override MongoDbSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 238    protected override MongoDbSubscriberRole Role => MongoDbSubscriberRole.ResponseIngress;
 239
 240    protected override Task HandleMessageAsync(MongoDbTransportDelivery delivery, CancellationToken cancellationToken)
 241    {
 242        var correlationId = !_ingress.IsOverInboundBudget(delivery.Payload)
 243            ? MongoDbCorrelationIdExtractor.Extract(delivery.Headers, delivery.Payload, Options)
 244            : null;
 245        return _ingress.HandleResponseMessageAsync(delivery.Payload, correlationId);
 246    }
 247}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.MongoDB.MongoDbAsyncResponseTransportOptions>,AsyncResponse.Transports.MongoDB.MongoDbTransportStore,Microsoft.Extensions.Logging.ILogger)
get_Options()
get_Logger()
StartAsync(System.Threading.CancellationToken)
ExecuteAsync()
RunSubscriberAsync()
ListenLoopAsync()
WaitForSignalOrDelayAsync()