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

Information
Class: AsyncResponse.Transports.MongoDB.MongoDbSubscriberService
Assembly: AsyncResponse.Transports.MongoDB
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.MongoDB/MongoDbSubscriberServices.cs
Line coverage
100%
Covered lines: 82
Uncovered lines: 0
Coverable lines: 82
Total lines: 203
Line coverage: 100%
Branch coverage
100%
Covered branches: 12
Total branches: 12
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%
ExecuteAsync()100%22100%
RunSubscriberAsync()100%66100%
ListenLoopAsync()100%11100%
WaitForSignalOrDelayAsync()100%44100%

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;
 315    private readonly Channel<bool> _signals = Channel.CreateBounded<bool>(new BoundedChannelOptions(1)
 316    {
 317        SingleReader = true,
 318        SingleWriter = false,
 319        FullMode = BoundedChannelFullMode.DropWrite
 320    });
 21
 322    protected MongoDbSubscriberService(
 323        IOptions<MongoDbAsyncResponseTransportOptions> options,
 324        MongoDbTransportStore store,
 325        ILogger logger)
 26    {
 327        Options = options.Value;
 328        MongoDbTransportOptionsValidator.ValidateCommon(Options);
 329        _store = store;
 330        Logger = logger;
 331    }
 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    {
 344        MongoDbTransportOptionsValidator.ValidateSubscriber(Options, SubscriberOptions, Role.ToString());
 45
 346        var failures = 0;
 347        while (!stoppingToken.IsCancellationRequested)
 48        {
 49            try
 50            {
 351                await RunSubscriberAsync(stoppingToken).ConfigureAwait(false);
 352                return;
 53            }
 354            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 55            {
 356                return;
 57            }
 358            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 59            {
 260                failures++;
 261                var delay = AsyncResponseRetry.Backoff(failures, Options.SubscriberRetryBaseDelay, Options.SubscriberRet
 262                Logger.LogWarning(ex, "MongoDB subscriber failed for queue {Queue} ({Role}); retrying in {RetryDelay}.",
 263                await Task.Delay(delay, stoppingToken).ConfigureAwait(false);
 64            }
 65        }
 366    }
 67
 68    private async Task RunSubscriberAsync(CancellationToken stoppingToken)
 69    {
 370        await _store.EnsureCreatedAsync(stoppingToken).ConfigureAwait(false);
 371        using var signalCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
 372        var listenTask = Options.UseChangeStreamWake
 373            ? Task.Run(() => ListenLoopAsync(signalCts.Token), signalCts.Token)
 374            : Task.CompletedTask;
 75
 376        await using var dispatcher = new MongoDbMessageDispatcher(
 377            HandleMessageAsync,
 378            Options,
 379            SubscriberOptions,
 380            Logger,
 381            Role);
 82
 383        Logger.LogInformation(
 384            "MongoDB subscriber started. Queue: {Queue}. Role: {Role}. AckMode: {AckMode}.",
 385            Queue,
 386            Role,
 387            SubscriberOptions.AckMode);
 88
 89        try
 90        {
 391            while (!stoppingToken.IsCancellationRequested)
 92            {
 393                var claimed = 0;
 394                await foreach (var delivery in _store.ClaimBatchAsync(Queue, SubscriberOptions.BatchSize, Options.LockTi
 95                {
 396                    claimed++;
 397                    await dispatcher.HandleAsync(delivery, stoppingToken).ConfigureAwait(false);
 98                }
 99
 3100                if (claimed > 0)
 101                    continue;
 102
 3103                await WaitForSignalOrDelayAsync(stoppingToken).ConfigureAwait(false);
 104            }
 105        }
 106        finally
 107        {
 3108            await signalCts.CancelAsync().ConfigureAwait(false);
 109            try
 110            {
 3111                await listenTask.WaitAsync(Options.ShutdownTimeout).ConfigureAwait(false);
 3112            }
 2113            catch (Exception ex) when (ex is OperationCanceledException or TimeoutException)
 114            {
 3115            }
 116        }
 3117    }
 118
 119    private async Task ListenLoopAsync(CancellationToken cancellationToken)
 120    {
 121        try
 122        {
 3123            await _store.WatchQueueAsync(Queue, () =>
 3124            {
 3125                _signals.Writer.TryWrite(true);
 3126                return Task.CompletedTask;
 3127            }, cancellationToken).ConfigureAwait(false);
 2128        }
 3129        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
 130        {
 3131        }
 2132        catch (Exception ex) when (MongoDbTransportStore.IsChangeStreamUnsupported(ex))
 133        {
 2134            Logger.LogInformation(
 2135                "MongoDB change streams are unavailable for queue {Queue} (the server is not a replica set); polling con
 2136                Queue,
 2137                SubscriberOptions.EmptyPollDelay);
 2138        }
 2139        catch (Exception ex)
 140        {
 2141            Logger.LogDebug(ex, "MongoDB change-stream wake for queue {Queue} stopped; polling continues.", Queue);
 3142        }
 3143    }
 144
 145    private async Task WaitForSignalOrDelayAsync(CancellationToken cancellationToken)
 146    {
 3147        var delay = Task.Delay(SubscriberOptions.EmptyPollDelay, cancellationToken);
 3148        var signal = _signals.Reader.WaitToReadAsync(cancellationToken).AsTask();
 3149        var completed = await Task.WhenAny(delay, signal).ConfigureAwait(false);
 3150        if (completed == signal)
 151        {
 3152            await signal.ConfigureAwait(false);
 3153            while (_signals.Reader.TryRead(out _))
 154            {
 155            }
 156        }
 3157    }
 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)
 170        : base(options, store, logger)
 171        => _ingress = ingress;
 172
 173    protected override string Queue => Options.WorkerQueue;
 174    protected override MongoDbSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 175    protected override MongoDbSubscriberRole Role => MongoDbSubscriberRole.Worker;
 176
 177    protected override Task HandleMessageAsync(MongoDbTransportDelivery delivery, CancellationToken cancellationToken)
 178        => _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, Microsoft.Extensions.Logging.ILogger)
ExecuteAsync()
RunSubscriberAsync()
ListenLoopAsync()
WaitForSignalOrDelayAsync()