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

Information
Class: AsyncResponse.Transports.SQS.SqsSubscriberService
Assembly: AsyncResponse.Transports.SQS
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SQS/SqsSubscriberServices.cs
Line coverage
93%
Covered lines: 109
Uncovered lines: 7
Coverable lines: 116
Total lines: 300
Line coverage: 93.9%
Branch coverage
95%
Covered branches: 23
Total branches: 24
Branch coverage: 95.8%
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()50%22100%
RunSubscriberAsync()100%44100%
DispatchBatchAsync()100%1212100%
RenewVisibilityLoopAsync()100%8661.11%
.ctor(...)100%11100%
get_SettledCount()100%11100%
MarkSettled()100%11100%
SuppressRenewal(...)100%11100%
IsRenewalSuppressed(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SQS/SqsSubscriberServices.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4
 5namespace AsyncResponse.Transports.SQS;
 6
 7internal abstract class SqsSubscriberService : BackgroundService
 8{
 9    private readonly ISqsClient _client;
 10
 311    protected SqsSubscriberService(
 312        IOptions<SqsAsyncResponseOptions> options,
 313        ISqsClient client,
 314        ILogger logger)
 15    {
 316        Options = options.Value;
 317        SqsOptionsValidator.ValidateCommon(Options);
 318        _client = client;
 319        Logger = logger;
 320    }
 21
 22    protected SqsAsyncResponseOptions Options { get; }
 23    protected ILogger Logger { get; }
 24
 25    protected abstract string QueueName { get; }
 26    protected abstract SqsSubscriberOptions SubscriberOptions { get; }
 27    protected abstract SqsSubscriberRole SubscriberRole { get; }
 28    /// <summary>Handles the delivered message.</summary>
 29    protected abstract Task HandleMessageAsync(SqsTransportDelivery delivery, CancellationToken cancellationToken);
 30
 31    /// <summary>Runs this background operation until cancellation is requested.</summary>
 32    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 33    {
 334        var queue = QueueName;
 335        SqsMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 336        var failures = 0;
 37
 338        while (!stoppingToken.IsCancellationRequested)
 39        {
 40            try
 41            {
 342                await RunSubscriberAsync(queue, stoppingToken).ConfigureAwait(false);
 243                return;
 44            }
 345            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 46            {
 347                return;
 48            }
 349            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 50            {
 351                failures++;
 352                var retryDelay = AsyncResponseRetry.Backoff(
 353                    failures,
 354                    Options.SubscriberRetryBaseDelay,
 355                    Options.SubscriberRetryMaxDelay);
 356                Logger.LogWarning(
 357                    ex,
 358                    "SQS subscriber failed for queue {Queue} ({Role}); retrying in {RetryDelay}.",
 359                    queue,
 360                    SubscriberRole,
 361                    retryDelay);
 362                await Task.Delay(retryDelay, stoppingToken).ConfigureAwait(false);
 63            }
 64        }
 365    }
 66
 67    private async Task RunSubscriberAsync(string queue, CancellationToken stoppingToken)
 68    {
 69        // A queue configured by name resolves through GetQueueUrl; failures here (queue not yet
 70        // provisioned, endpoint still starting) surface to the retry loop above.
 371        var queueUrl = SqsQueueAddress.IsUrl(queue)
 372            ? queue
 373            : await _client.GetQueueUrlAsync(queue, stoppingToken).ConfigureAwait(false);
 74
 375        await using var dispatcher = SqsMessageDispatcher.Create(
 376            HandleMessageAsync,
 377            Options,
 378            SubscriberOptions,
 379            Logger,
 380            queue,
 381            SubscriberRole);
 82
 383        Logger.LogInformation(
 384            "SQS subscriber started. Queue: {Queue}. Role: {Role}. AckMode: {AckMode}.",
 385            queue,
 386            SubscriberRole,
 387            SubscriberOptions.AckMode);
 88
 389        while (!stoppingToken.IsCancellationRequested)
 90        {
 91            // In early-ACK mode, receiving while the background queue is saturated would burn the
 92            // queue's redrive policy (SQS counts every receive), so wait for free capacity and never
 93            // request more messages than the dispatcher can accept.
 394            await dispatcher.WaitForCapacityAsync(stoppingToken).ConfigureAwait(false);
 395            var maxMessages = Math.Min(Options.MaxMessagesPerReceive, dispatcher.FreeCapacity);
 96
 397            var deliveries = await _client.ReceiveMessagesAsync(
 398                new SqsReceiveRequest(
 399                    queueUrl,
 3100                    maxMessages,
 3101                    Options.ReceiveWaitTime,
 3102                    SubscriberOptions.VisibilityTimeout),
 3103                stoppingToken).ConfigureAwait(false);
 104
 3105            await DispatchBatchAsync(dispatcher, deliveries, queue, stoppingToken).ConfigureAwait(false);
 106        }
 2107    }
 108
 109    private async Task DispatchBatchAsync(
 110        SqsMessageDispatcher dispatcher,
 111        IReadOnlyList<SqsTransportDelivery> deliveries,
 112        string queue,
 113        CancellationToken stoppingToken)
 114    {
 3115        if (deliveries.Count == 0)
 3116            return;
 117
 3118        if (SubscriberOptions.AckMode is not SqsAckMode.AckAfterHandlerCompletes
 3119            || SubscriberOptions.VisibilityRenewalInterval is not { } renewalInterval
 3120            || SubscriberOptions.VisibilityTimeout is not { } visibilityTimeout)
 121        {
 3122            foreach (var delivery in deliveries)
 3123                await dispatcher.HandleAsync(delivery, stoppingToken).ConfigureAwait(false);
 3124            return;
 125        }
 126
 127        // The batch is processed serially, so a slow handler lets the visibility timeout of the later
 128        // (still unprocessed) messages lapse and a competing consumer processes them a second time.
 129        // While the batch is in flight, a heartbeat resets every unsettled message's invisibility to
 130        // the configured visibility timeout.
 2131        var progress = new BatchProgress(deliveries.Count);
 2132        using var renewalCancellation = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
 2133        var renewalTask = RenewVisibilityLoopAsync(
 2134            deliveries,
 2135            progress,
 2136            renewalInterval,
 2137            visibilityTimeout,
 2138            queue,
 2139            renewalCancellation.Token);
 140        try
 141        {
 2142            for (var index = 0; index < deliveries.Count; index++)
 143            {
 2144                var delivery = deliveries[index];
 2145                var batchIndex = index;
 146                // The dispatcher's failure path shortens visibility to RedeliveryDelay while the
 147                // heartbeat still counts the message as unsettled (MarkSettled runs only after
 148                // HandleAsync returns). Routing the dispatcher's visibility changes through a
 149                // suppression mark — set before the change itself — keeps a racing heartbeat from
 150                // stretching that fast retry back out to the full visibility timeout.
 2151                var tracked = delivery with
 2152                {
 2153                    ChangeVisibilityAsync = timeout =>
 2154                    {
 2155                        progress.SuppressRenewal(batchIndex);
 2156                        return delivery.ChangeVisibilityAsync(timeout);
 2157                    }
 2158                };
 159                try
 160                {
 2161                    await dispatcher.HandleAsync(tracked, stoppingToken).ConfigureAwait(false);
 2162                }
 163                finally
 164                {
 2165                    progress.MarkSettled();
 166                }
 2167            }
 168        }
 169        finally
 170        {
 2171            renewalCancellation.Cancel();
 2172            await renewalTask.ConfigureAwait(false);
 173        }
 3174    }
 175
 176    private async Task RenewVisibilityLoopAsync(
 177        IReadOnlyList<SqsTransportDelivery> deliveries,
 178        BatchProgress progress,
 179        TimeSpan renewalInterval,
 180        TimeSpan visibilityTimeout,
 181        string queue,
 182        CancellationToken cancellationToken)
 183    {
 184        try
 185        {
 2186            while (true)
 187            {
 2188                await Task.Delay(renewalInterval, cancellationToken).ConfigureAwait(false);
 189
 190                // Renew from the first unsettled message onward: that covers the message currently in
 191                // the handler plus everything still waiting its turn. Two settle paths race this
 192                // sweep, and only one of them is harmless. A handled message was deleted, so a late
 193                // renewal merely fails and is logged — SQS redelivery keeps at-least-once intact. A
 194                // failed message was NOT deleted (its receipt handle stays live) and already carries
 195                // the failure path's shortened RedeliveryDelay, so a late renewal here would SUCCEED
 196                // and stretch that fast retry back out to the full visibility timeout — the
 197                // suppression mark and the per-message re-read of the settled prefix keep the sweep
 198                // away from it.
 2199                for (var i = progress.SettledCount; i < deliveries.Count; i++)
 200                {
 2201                    if (i < progress.SettledCount || progress.IsRenewalSuppressed(i))
 202                        continue;
 203
 2204                    var delivery = deliveries[i];
 205                    try
 206                    {
 2207                        await delivery.ChangeVisibilityAsync(visibilityTimeout).ConfigureAwait(false);
 2208                    }
 0209                    catch (Exception ex)
 210                    {
 0211                        Logger.LogWarning(
 0212                            ex,
 0213                            "Failed to renew visibility of SQS message {MessageId} on {Queue}; it may redeliver while st
 0214                            delivery.MessageId,
 0215                            queue);
 0216                    }
 2217                }
 218            }
 219        }
 2220        catch (OperationCanceledException)
 221        {
 222            // The batch finished or the subscriber is stopping.
 2223        }
 2224    }
 225
 226    private sealed class BatchProgress
 227    {
 228        // One slot per batch message. Settled and suppressed only ever transition false→true, so
 229        // monotonic volatile writes/reads are enough — no lock, and a stale read only delays a
 230        // skip by one sweep pass.
 231        private readonly bool[] _renewalSuppressed;
 232        private int _settledCount;
 233
 2234        public BatchProgress(int batchSize) => _renewalSuppressed = new bool[batchSize];
 235
 2236        public int SettledCount => Volatile.Read(ref _settledCount);
 237
 2238        public void MarkSettled() => Interlocked.Increment(ref _settledCount);
 239
 240        /// <summary>Marks the message at <paramref name="index"/> as owning its own visibility; the renewal sweep must 
 2241        public void SuppressRenewal(int index) => Volatile.Write(ref _renewalSuppressed[index], true);
 242
 2243        public bool IsRenewalSuppressed(int index) => Volatile.Read(ref _renewalSuppressed[index]);
 244    }
 245}
 246
 247internal sealed class SqsWorkerSubscriber : SqsSubscriberService
 248{
 249    private readonly IAsyncResponseIngress _ingress;
 250
 251    /// <summary>Creates a worker subscriber for the configured SQS worker queue.</summary>
 252    public SqsWorkerSubscriber(
 253        IOptions<SqsAsyncResponseOptions> options,
 254        ISqsClient client,
 255        IAsyncResponseIngress ingress,
 256        ILogger<SqsWorkerSubscriber> logger)
 257        : base(options, client, logger)
 258    {
 259        _ingress = ingress;
 260    }
 261
 262    protected override string QueueName
 263        => SqsOptionsValidator.Required(Options.WorkerQueue, nameof(Options.WorkerQueue));
 264
 265    protected override SqsSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 266    protected override SqsSubscriberRole SubscriberRole => SqsSubscriberRole.Worker;
 267
 268    /// <summary>Handles the delivered message.</summary>
 269    protected override Task HandleMessageAsync(SqsTransportDelivery delivery, CancellationToken cancellationToken)
 270        => _ingress.HandleWorkerMessageAsync(delivery.Body);
 271}
 272
 273internal sealed class SqsResponseIngressSubscriber : SqsSubscriberService
 274{
 275    private readonly IAsyncResponseIngress _ingress;
 276
 277    /// <summary>Creates a response subscriber for the configured SQS response queue.</summary>
 278    public SqsResponseIngressSubscriber(
 279        IOptions<SqsAsyncResponseOptions> options,
 280        ISqsClient client,
 281        IAsyncResponseIngress ingress,
 282        ILogger<SqsResponseIngressSubscriber> logger)
 283        : base(options, client, logger)
 284    {
 285        _ingress = ingress;
 286    }
 287
 288    protected override string QueueName
 289        => SqsOptionsValidator.Required(Options.ResponseQueue, nameof(Options.ResponseQueue));
 290
 291    protected override SqsSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 292    protected override SqsSubscriberRole SubscriberRole => SqsSubscriberRole.ResponseIngress;
 293
 294    /// <summary>Handles the delivered message.</summary>
 295    protected override Task HandleMessageAsync(SqsTransportDelivery delivery, CancellationToken cancellationToken)
 296    {
 297        var correlationId = SqsCorrelationIdExtractor.Extract(delivery, delivery.Body, Options);
 298        return _ingress.HandleResponseMessageAsync(delivery.Body, correlationId);
 299    }
 300}
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.SQS.SqsAsyncResponseOptions>, AsyncResponse.Transports.SQS.ISqsClient, Microsoft.Extensions.Logging.ILogger)
ExecuteAsync()
RunSubscriberAsync()
DispatchBatchAsync()
RenewVisibilityLoopAsync()
.ctor(int)
get_SettledCount()
MarkSettled()
SuppressRenewal(int)
IsRenewalSuppressed(int)