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

Information
Class: AsyncResponse.Transports.PostgreSQL.PostgreSqlSubscriberService
Assembly: AsyncResponse.Transports.PostgreSQL
File(s): /_/src/Transports/AsyncResponse.Transports.PostgreSQL/PostgreSqlSubscriberServices.cs
Line coverage
92%
Covered lines: 77
Uncovered lines: 6
Coverable lines: 83
Total lines: 235
Line coverage: 92.7%
Branch coverage
95%
Covered branches: 19
Total branches: 20
Branch coverage: 95%
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%121290.9%
ListenLoopAsync()50%2272.22%
WaitForSignalOrDelayAsync()100%44100%

File(s)

/_/src/Transports/AsyncResponse.Transports.PostgreSQL/PostgreSqlSubscriberServices.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using System.Threading.Channels;
 5
 6namespace AsyncResponse.Transports.PostgreSQL;
 7
 8/// <summary>
 9/// Base hosted service that consumes one PostgreSQL queue and routes rows to AsyncResponse ingress
 10/// with configured acknowledgement, redelivery, and dead-letter behavior.
 11/// </summary>
 12internal abstract class PostgreSqlSubscriberService : BackgroundService
 13{
 14    private readonly PostgreSqlTransportStore _store;
 40015    private readonly Channel<bool> _signals = Channel.CreateBounded<bool>(new BoundedChannelOptions(1)
 40016    {
 40017        SingleReader = true,
 40018        SingleWriter = false,
 40019        FullMode = BoundedChannelFullMode.DropWrite
 40020    });
 21
 40022    protected PostgreSqlSubscriberService(
 40023        IOptions<PostgreSqlAsyncResponseTransportOptions> options,
 40024        PostgreSqlTransportStore store,
 40025        ILogger logger)
 26    {
 40027        Options = options.Value;
 40028        PostgreSqlTransportOptionsValidator.ValidateCommon(Options);
 40029        _store = store;
 40030        Logger = logger;
 40031    }
 32
 956633    protected PostgreSqlAsyncResponseTransportOptions Options { get; }
 77834    protected ILogger Logger { get; }
 35
 36    protected abstract string Queue { get; }
 37    protected abstract PostgreSqlSubscriberOptions SubscriberOptions { get; }
 38    protected abstract PostgreSqlSubscriberRole Role { get; }
 39    protected abstract Task HandleMessageAsync(PostgreSqlTransportDelivery 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    {
 38851        PostgreSqlTransportOptionsValidator.ValidateSubscriber(Options, SubscriberOptions, Role.ToString());
 38652        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.
 38666        await using var dispatcher = new PostgreSqlMessageDispatcher(
 38667            HandleMessageAsync,
 38668            Options,
 38669            SubscriberOptions,
 38670            Logger,
 38671            Role);
 72
 38673        await SubscriberSupervisor.RunAsync(
 38674            attemptToken => RunSubscriberAsync(dispatcher, attemptToken),
 38675            stoppingToken,
 076            failures => AsyncResponseRetry.Backoff(failures, Options.SubscriberRetryBaseDelay, Options.SubscriberRetryMa
 38677            (ex, delay) => Logger.LogWarning(ex, "PostgreSQL subscriber failed for queue {Queue} ({Role}); retrying in {
 38678    }
 79
 80    private async Task RunSubscriberAsync(PostgreSqlMessageDispatcher dispatcher, CancellationToken stoppingToken)
 81    {
 38882        await _store.EnsureCreatedAsync(stoppingToken).ConfigureAwait(false);
 38883        using var signalCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
 84
 85        // The LISTEN 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 in
 88        // connection.WaitAsync holding a pooled connection, one per retry until pool exhaustion.
 38889        Task? listenTask = null;
 90        try
 91        {
 77692            listenTask = Task.Run(() => ListenLoopAsync(signalCts.Token), signalCts.Token);
 93
 38894            Logger.LogInformation(
 38895                "PostgreSQL subscriber started. Queue: {Queue}. Role: {Role}. AckMode: {AckMode}.",
 38896                Queue,
 38897                Role,
 38898                SubscriberOptions.AckMode);
 99
 1988100            while (!stoppingToken.IsCancellationRequested)
 101            {
 1654102                var claimed = 0;
 4132103                await foreach (var delivery in _store.ClaimBatchAsync(Queue, SubscriberOptions.BatchSize, Options.LockTi
 104                {
 412105                    claimed++;
 412106                    await dispatcher.HandleAsync(delivery, stoppingToken).ConfigureAwait(false);
 107                }
 108
 1630109                if (claimed > 0)
 110                    continue;
 111
 1391112                await WaitForSignalOrDelayAsync(stoppingToken).ConfigureAwait(false);
 113            }
 114        }
 115        finally
 116        {
 388117            await signalCts.CancelAsync().ConfigureAwait(false);
 388118            if (listenTask is not null)
 119            {
 120                try
 121                {
 388122                    await listenTask.WaitAsync(Options.ShutdownTimeout).ConfigureAwait(false);
 388123                }
 0124                catch (Exception ex) when (ex is OperationCanceledException or TimeoutException)
 125                {
 0126                }
 127            }
 128        }
 334129    }
 130
 131    private async Task ListenLoopAsync(CancellationToken cancellationToken)
 132    {
 133        // Retry with backoff, mirroring the channel-side listener: a transient LISTEN failure
 134        // (network blip, failover) must not permanently degrade this subscriber from push wake
 135        // to poll-only latency for the rest of the process's uptime.
 388136        var failures = 0;
 388137        while (!cancellationToken.IsCancellationRequested)
 138        {
 139            try
 140            {
 388141                await _store.ExecuteListenAsync(() =>
 388142                {
 605143                    _signals.Writer.TryWrite(true);
 605144                    return Task.CompletedTask;
 388145                }, cancellationToken).ConfigureAwait(false);
 0146                failures = 0;
 0147            }
 386148            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
 149            {
 386150                return;
 151            }
 2152            catch (Exception ex)
 153            {
 2154                failures++;
 2155                var delay = AsyncResponseRetry.Backoff(failures, Options.SubscriberRetryBaseDelay, Options.SubscriberRet
 2156                Logger.LogWarning(ex, "PostgreSQL LISTEN helper for queue {Queue} failed; retrying in {RetryDelay} (poll
 157                try
 158                {
 2159                    await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
 0160                }
 2161                catch (OperationCanceledException)
 162                {
 2163                    return;
 164                }
 165            }
 166        }
 388167    }
 168
 169    private async Task WaitForSignalOrDelayAsync(CancellationToken cancellationToken)
 170    {
 171        // The WhenAny loser is cancelled via the per-iteration linked source (mirroring the
 172        // channel-side CollectDispatchScopeAsync): an abandoned WaitToReadAsync would otherwise
 173        // stay parked in the channel's blocked-reader list until the next signal — one per empty
 174        // poll, accumulating without bound on an idle queue.
 1481175        using var iteration = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 1481176        var delay = Task.Delay(SubscriberOptions.EmptyPollDelay, iteration.Token);
 1481177        var signal = _signals.Reader.WaitToReadAsync(iteration.Token).AsTask();
 1481178        var completed = await Task.WhenAny(delay, signal).ConfigureAwait(false);
 1481179        iteration.Cancel();
 1481180        if (completed == signal)
 181        {
 560182            await signal.ConfigureAwait(false);
 1064183            while (_signals.Reader.TryRead(out _))
 184            {
 185            }
 186        }
 1453187    }
 188}
 189
 190/// <summary>Consumes worker-job rows and executes them through the AsyncResponse ingress.</summary>
 191internal sealed class PostgreSqlWorkerSubscriber : PostgreSqlSubscriberService
 192{
 193    private readonly IAsyncResponseIngress _ingress;
 194
 195    public PostgreSqlWorkerSubscriber(
 196        IOptions<PostgreSqlAsyncResponseTransportOptions> options,
 197        PostgreSqlTransportStore store,
 198        IAsyncResponseIngress ingress,
 199        ILogger<PostgreSqlWorkerSubscriber> logger)
 200        : base(options, store, logger)
 201        => _ingress = ingress;
 202
 203    protected override string Queue => Options.WorkerQueue;
 204    protected override PostgreSqlSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 205    protected override PostgreSqlSubscriberRole Role => PostgreSqlSubscriberRole.Worker;
 206
 207    protected override Task HandleMessageAsync(PostgreSqlTransportDelivery delivery, CancellationToken cancellationToken
 208        => _ingress.HandleWorkerMessageAsync(delivery.Payload);
 209}
 210
 211/// <summary>Consumes response rows and feeds them into the AsyncResponse ingress.</summary>
 212internal sealed class PostgreSqlResponseIngressSubscriber : PostgreSqlSubscriberService
 213{
 214    private readonly IAsyncResponseIngress _ingress;
 215
 216    public PostgreSqlResponseIngressSubscriber(
 217        IOptions<PostgreSqlAsyncResponseTransportOptions> options,
 218        PostgreSqlTransportStore store,
 219        IAsyncResponseIngress ingress,
 220        ILogger<PostgreSqlResponseIngressSubscriber> logger)
 221        : base(options, store, logger)
 222        => _ingress = ingress;
 223
 224    protected override string Queue => Options.ResponseQueue;
 225    protected override PostgreSqlSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 226    protected override PostgreSqlSubscriberRole Role => PostgreSqlSubscriberRole.ResponseIngress;
 227
 228    protected override Task HandleMessageAsync(PostgreSqlTransportDelivery delivery, CancellationToken cancellationToken
 229    {
 230        var correlationId = !_ingress.IsOverInboundBudget(delivery.Payload)
 231            ? PostgreSqlCorrelationIdExtractor.Extract(delivery.Headers, delivery.Payload, Options)
 232            : null;
 233        return _ingress.HandleResponseMessageAsync(delivery.Payload, correlationId);
 234    }
 235}
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.PostgreSQL.PostgreSqlAsyncResponseTransportOptions>,AsyncResponse.Transports.PostgreSQL.PostgreSqlTransportStore,Microsoft.Extensions.Logging.ILogger)
get_Options()
get_Logger()
StartAsync(System.Threading.CancellationToken)
ExecuteAsync()
RunSubscriberAsync()
ListenLoopAsync()
WaitForSignalOrDelayAsync()