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

Information
Class: AsyncResponse.Transports.MongoDB.MongoDbWorkerSubscriber
Assembly: AsyncResponse.Transports.MongoDB
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.MongoDB/MongoDbSubscriberServices.cs
Line coverage
100%
Covered lines: 6
Uncovered lines: 0
Coverable lines: 6
Total lines: 203
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
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_Queue()100%11100%
get_SubscriberOptions()100%11100%
get_Role()100%11100%
HandleMessageAsync(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/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;
 15    private readonly Channel<bool> _signals = Channel.CreateBounded<bool>(new BoundedChannelOptions(1)
 16    {
 17        SingleReader = true,
 18        SingleWriter = false,
 19        FullMode = BoundedChannelFullMode.DropWrite
 20    });
 21
 22    protected MongoDbSubscriberService(
 23        IOptions<MongoDbAsyncResponseTransportOptions> options,
 24        MongoDbTransportStore store,
 25        ILogger logger)
 26    {
 27        Options = options.Value;
 28        MongoDbTransportOptionsValidator.ValidateCommon(Options);
 29        _store = store;
 30        Logger = logger;
 31    }
 32
 33    protected MongoDbAsyncResponseTransportOptions Options { get; }
 34    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    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 43    {
 44        MongoDbTransportOptionsValidator.ValidateSubscriber(Options, SubscriberOptions, Role.ToString());
 45
 46        var failures = 0;
 47        while (!stoppingToken.IsCancellationRequested)
 48        {
 49            try
 50            {
 51                await RunSubscriberAsync(stoppingToken).ConfigureAwait(false);
 52                return;
 53            }
 54            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 55            {
 56                return;
 57            }
 58            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 59            {
 60                failures++;
 61                var delay = AsyncResponseRetry.Backoff(failures, Options.SubscriberRetryBaseDelay, Options.SubscriberRet
 62                Logger.LogWarning(ex, "MongoDB subscriber failed for queue {Queue} ({Role}); retrying in {RetryDelay}.",
 63                await Task.Delay(delay, stoppingToken).ConfigureAwait(false);
 64            }
 65        }
 66    }
 67
 68    private async Task RunSubscriberAsync(CancellationToken stoppingToken)
 69    {
 70        await _store.EnsureCreatedAsync(stoppingToken).ConfigureAwait(false);
 71        using var signalCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
 72        var listenTask = Options.UseChangeStreamWake
 73            ? Task.Run(() => ListenLoopAsync(signalCts.Token), signalCts.Token)
 74            : Task.CompletedTask;
 75
 76        await using var dispatcher = new MongoDbMessageDispatcher(
 77            HandleMessageAsync,
 78            Options,
 79            SubscriberOptions,
 80            Logger,
 81            Role);
 82
 83        Logger.LogInformation(
 84            "MongoDB subscriber started. Queue: {Queue}. Role: {Role}. AckMode: {AckMode}.",
 85            Queue,
 86            Role,
 87            SubscriberOptions.AckMode);
 88
 89        try
 90        {
 91            while (!stoppingToken.IsCancellationRequested)
 92            {
 93                var claimed = 0;
 94                await foreach (var delivery in _store.ClaimBatchAsync(Queue, SubscriberOptions.BatchSize, Options.LockTi
 95                {
 96                    claimed++;
 97                    await dispatcher.HandleAsync(delivery, stoppingToken).ConfigureAwait(false);
 98                }
 99
 100                if (claimed > 0)
 101                    continue;
 102
 103                await WaitForSignalOrDelayAsync(stoppingToken).ConfigureAwait(false);
 104            }
 105        }
 106        finally
 107        {
 108            await signalCts.CancelAsync().ConfigureAwait(false);
 109            try
 110            {
 111                await listenTask.WaitAsync(Options.ShutdownTimeout).ConfigureAwait(false);
 112            }
 113            catch (Exception ex) when (ex is OperationCanceledException or TimeoutException)
 114            {
 115            }
 116        }
 117    }
 118
 119    private async Task ListenLoopAsync(CancellationToken cancellationToken)
 120    {
 121        try
 122        {
 123            await _store.WatchQueueAsync(Queue, () =>
 124            {
 125                _signals.Writer.TryWrite(true);
 126                return Task.CompletedTask;
 127            }, cancellationToken).ConfigureAwait(false);
 128        }
 129        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
 130        {
 131        }
 132        catch (Exception ex) when (MongoDbTransportStore.IsChangeStreamUnsupported(ex))
 133        {
 134            Logger.LogInformation(
 135                "MongoDB change streams are unavailable for queue {Queue} (the server is not a replica set); polling con
 136                Queue,
 137                SubscriberOptions.EmptyPollDelay);
 138        }
 139        catch (Exception ex)
 140        {
 141            Logger.LogDebug(ex, "MongoDB change-stream wake for queue {Queue} stopped; polling continues.", Queue);
 142        }
 143    }
 144
 145    private async Task WaitForSignalOrDelayAsync(CancellationToken cancellationToken)
 146    {
 147        var delay = Task.Delay(SubscriberOptions.EmptyPollDelay, cancellationToken);
 148        var signal = _signals.Reader.WaitToReadAsync(cancellationToken).AsTask();
 149        var completed = await Task.WhenAny(delay, signal).ConfigureAwait(false);
 150        if (completed == signal)
 151        {
 152            await signal.ConfigureAwait(false);
 153            while (_signals.Reader.TryRead(out _))
 154            {
 155            }
 156        }
 157    }
 158}
 159
 160/// <summary>Consumes worker-job documents and executes them through the AsyncResponse ingress.</summary>
 161internal sealed class MongoDbWorkerSubscriber : MongoDbSubscriberService
 162{
 163    private readonly IAsyncResponseIngress _ingress;
 164
 165    public MongoDbWorkerSubscriber(
 166        IOptions<MongoDbAsyncResponseTransportOptions> options,
 167        MongoDbTransportStore store,
 168        IAsyncResponseIngress ingress,
 169        ILogger<MongoDbWorkerSubscriber> logger)
 3170        : base(options, store, logger)
 3171        => _ingress = ingress;
 172
 3173    protected override string Queue => Options.WorkerQueue;
 3174    protected override MongoDbSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 3175    protected override MongoDbSubscriberRole Role => MongoDbSubscriberRole.Worker;
 176
 177    protected override Task HandleMessageAsync(MongoDbTransportDelivery delivery, CancellationToken cancellationToken)
 3178        => _ingress.HandleWorkerMessageAsync(delivery.Payload);
 179}
 180
 181/// <summary>Consumes response documents and feeds them into the AsyncResponse ingress.</summary>
 182internal sealed class MongoDbResponseIngressSubscriber : MongoDbSubscriberService
 183{
 184    private readonly IAsyncResponseIngress _ingress;
 185
 186    public MongoDbResponseIngressSubscriber(
 187        IOptions<MongoDbAsyncResponseTransportOptions> options,
 188        MongoDbTransportStore store,
 189        IAsyncResponseIngress ingress,
 190        ILogger<MongoDbResponseIngressSubscriber> logger)
 191        : base(options, store, logger)
 192        => _ingress = ingress;
 193
 194    protected override string Queue => Options.ResponseQueue;
 195    protected override MongoDbSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 196    protected override MongoDbSubscriberRole Role => MongoDbSubscriberRole.ResponseIngress;
 197
 198    protected override Task HandleMessageAsync(MongoDbTransportDelivery delivery, CancellationToken cancellationToken)
 199    {
 200        var correlationId = MongoDbCorrelationIdExtractor.Extract(delivery.Headers, delivery.Payload, Options);
 201        return _ingress.HandleResponseMessageAsync(delivery.Payload, correlationId);
 202    }
 203}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.MongoDB.MongoDbAsyncResponseTransportOptions>, AsyncResponse.Transports.MongoDB.MongoDbTransportStore, AsyncResponse.IAsyncResponseIngress, Microsoft.Extensions.Logging.ILogger<AsyncResponse.Transports.MongoDB.MongoDbWorkerSubscriber>)
get_Queue()
get_SubscriberOptions()
get_Role()
HandleMessageAsync(AsyncResponse.Transports.MongoDB.MongoDbTransportDelivery, System.Threading.CancellationToken)