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

Information
Class: AsyncResponse.Transports.AzureServiceBus.AzureServiceBusResponseIngressSubscriber
Assembly: AsyncResponse.Transports.AzureServiceBus
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.AzureServiceBus/AzureServiceBusSubscriberServices.cs
Line coverage
100%
Covered lines: 8
Uncovered lines: 0
Coverable lines: 8
Total lines: 286
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_QueueName()100%11100%
get_SubscriberOptions()100%11100%
get_SubscriberRole()100%11100%
HandleMessageAsync(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.AzureServiceBus/AzureServiceBusSubscriberServices.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4
 5namespace AsyncResponse.Transports.AzureServiceBus;
 6
 7internal abstract class AzureServiceBusSubscriberService : BackgroundService
 8{
 9    private readonly IAzureServiceBusClient _client;
 10
 11    protected AzureServiceBusSubscriberService(
 12        IOptions<AzureServiceBusAsyncResponseOptions> options,
 13        IAzureServiceBusClient client,
 14        ILogger logger)
 15    {
 16        Options = options.Value;
 17        AzureServiceBusOptionsValidator.ValidateCommon(Options);
 18        _client = client;
 19        Logger = logger;
 20    }
 21
 22    protected AzureServiceBusAsyncResponseOptions Options { get; }
 23    protected ILogger Logger { get; }
 24
 25    protected abstract string QueueName { get; }
 26    protected abstract AzureServiceBusSubscriberOptions SubscriberOptions { get; }
 27    protected abstract AzureServiceBusSubscriberRole SubscriberRole { get; }
 28    /// <summary>Handles the delivered message.</summary>
 29    protected abstract Task HandleMessageAsync(AzureServiceBusTransportDelivery delivery, CancellationToken cancellation
 30
 31    /// <summary>Runs this background operation until cancellation is requested.</summary>
 32    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 33    {
 34        var queue = QueueName;
 35        AzureServiceBusMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 36        var failures = 0;
 37
 38        while (!stoppingToken.IsCancellationRequested)
 39        {
 40            try
 41            {
 42                await RunSubscriberAsync(queue, stoppingToken).ConfigureAwait(false);
 43                return;
 44            }
 45            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 46            {
 47                return;
 48            }
 49            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 50            {
 51                failures++;
 52                var retryDelay = RetryDelay(failures);
 53                Logger.LogWarning(
 54                    ex,
 55                    "Azure Service Bus subscriber failed for queue {Queue} ({Role}); retrying in {RetryDelay}.",
 56                    queue,
 57                    SubscriberRole,
 58                    retryDelay);
 59                await Task.Delay(retryDelay, stoppingToken).ConfigureAwait(false);
 60            }
 61        }
 62    }
 63
 64    private async Task RunSubscriberAsync(string queue, CancellationToken stoppingToken)
 65    {
 66        await using var receiver = _client.CreateReceiver(queue, SubscriberOptions);
 67        await using var dispatcher = AzureServiceBusMessageDispatcher.Create(
 68            HandleMessageAsync,
 69            Options,
 70            SubscriberOptions,
 71            Logger,
 72            queue,
 73            SubscriberRole);
 74
 75        Logger.LogInformation(
 76            "Azure Service Bus subscriber started. Queue: {Queue}. Role: {Role}. AckMode: {AckMode}.",
 77            queue,
 78            SubscriberRole,
 79            SubscriberOptions.AckMode);
 80
 81        try
 82        {
 83            while (!stoppingToken.IsCancellationRequested)
 84            {
 85                // In early-ACK mode, receiving while the background queue is saturated would burn
 86                // DeliveryCount via queue-full abandons, so wait for free capacity and never request
 87                // more messages than the dispatcher can accept.
 88                await dispatcher.WaitForCapacityAsync(stoppingToken).ConfigureAwait(false);
 89                var maxMessages = Math.Min(Options.MaxMessagesPerReceive, dispatcher.FreeCapacity);
 90
 91                var messages = await receiver.ReceiveMessagesAsync(
 92                    maxMessages,
 93                    Options.ReceiveWaitTime,
 94                    stoppingToken).ConfigureAwait(false);
 95
 96                await DispatchBatchAsync(dispatcher, messages, queue, stoppingToken).ConfigureAwait(false);
 97            }
 98        }
 99        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 100        {
 101            using var shutdown = new CancellationTokenSource(Options.ShutdownTimeout);
 102            await receiver.CloseAsync(shutdown.Token).ConfigureAwait(false);
 103        }
 104    }
 105
 106    private async Task DispatchBatchAsync(
 107        AzureServiceBusMessageDispatcher dispatcher,
 108        IReadOnlyList<AzureServiceBusTransportDelivery> messages,
 109        string queue,
 110        CancellationToken stoppingToken)
 111    {
 112        if (messages.Count == 0)
 113            return;
 114
 115        if (SubscriberOptions.AckMode is not AzureServiceBusAckMode.AckAfterHandlerCompletes
 116            || SubscriberOptions.LockRenewalInterval is not { } renewalInterval)
 117        {
 118            foreach (var message in messages)
 119                await dispatcher.HandleAsync(message, stoppingToken).ConfigureAwait(false);
 120            return;
 121        }
 122
 123        // The batch is processed serially, so a slow handler lets the peek locks of the later (still
 124        // unsettled) messages expire and Service Bus redelivers them to a competing consumer while
 125        // they are still queued here — systematic duplicate processing. While the batch is in flight,
 126        // a background loop renews the lock of every unsettled message each interval.
 127        var progress = new BatchProgress();
 128        using var renewalCancellation = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
 129        var renewalTask = RenewLocksLoopAsync(messages, progress, renewalInterval, queue, renewalCancellation.Token);
 130        try
 131        {
 132            foreach (var message in messages)
 133            {
 134                try
 135                {
 136                    await dispatcher.HandleAsync(message, stoppingToken).ConfigureAwait(false);
 137                }
 138                finally
 139                {
 140                    progress.MarkSettled();
 141                }
 142            }
 143        }
 144        finally
 145        {
 146            renewalCancellation.Cancel();
 147            try
 148            {
 149                // Cancellation exits the sweep between messages and interrupts the in-flight renew
 150                // call, so this normally completes immediately. The bound is the hard backstop: an
 151                // unbounded await here would let a degraded namespace hold the receive loop (and, at
 152                // shutdown, the whole host budget) hostage for the SDK retry budget per message.
 153                await renewalTask.WaitAsync(Options.ShutdownTimeout).ConfigureAwait(false);
 154            }
 155            catch (TimeoutException)
 156            {
 157                Logger.LogWarning(
 158                    "Azure Service Bus lock renewal for {Queue} ({Role}) did not stop within the shutdown budget ({Shutd
 159                    queue,
 160                    SubscriberRole,
 161                    Options.ShutdownTimeout);
 162            }
 163        }
 164    }
 165
 166    private async Task RenewLocksLoopAsync(
 167        IReadOnlyList<AzureServiceBusTransportDelivery> messages,
 168        BatchProgress progress,
 169        TimeSpan renewalInterval,
 170        string queue,
 171        CancellationToken cancellationToken)
 172    {
 173        try
 174        {
 175            while (true)
 176            {
 177                await Task.Delay(renewalInterval, cancellationToken).ConfigureAwait(false);
 178
 179                // Renew from the first unsettled message onward: that covers the message currently in
 180                // the handler plus everything still waiting its turn. A renewal racing a just-settled
 181                // message merely fails and is logged; redelivery keeps at-least-once intact.
 182                for (var i = progress.SettledCount; i < messages.Count; i++)
 183                {
 184                    // The batch finished or the subscriber is stopping: exit quietly between messages
 185                    // instead of spending up to a full SDK retry budget on each remaining renew.
 186                    if (cancellationToken.IsCancellationRequested)
 187                        return;
 188
 189                    var message = messages[i];
 190                    try
 191                    {
 192                        await message.RenewLockAsync(cancellationToken).ConfigureAwait(false);
 193                    }
 194                    catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
 195                    {
 196                        // Our token interrupted the in-flight renew; not a renewal failure.
 197                        return;
 198                    }
 199                    catch (Exception ex)
 200                    {
 201                        Logger.LogWarning(
 202                            ex,
 203                            "Failed to renew the lock of Azure Service Bus message {MessageId} on {Queue}; it may redeli
 204                            message.MessageId,
 205                            queue);
 206                    }
 207                }
 208            }
 209        }
 210        catch (OperationCanceledException)
 211        {
 212            // The batch finished or the subscriber is stopping.
 213        }
 214    }
 215
 216    private sealed class BatchProgress
 217    {
 218        private int _settledCount;
 219
 220        public int SettledCount => Volatile.Read(ref _settledCount);
 221
 222        public void MarkSettled() => Interlocked.Increment(ref _settledCount);
 223    }
 224
 225    private TimeSpan RetryDelay(int failures)
 226    {
 227        var exponent = Math.Max(0, failures - 1);
 228        var milliseconds = Options.SubscriberRetryBaseDelay.TotalMilliseconds * Math.Pow(2, exponent);
 229        return TimeSpan.FromMilliseconds(Math.Min(milliseconds, Options.SubscriberRetryMaxDelay.TotalMilliseconds));
 230    }
 231}
 232
 233internal sealed class AzureServiceBusWorkerSubscriber : AzureServiceBusSubscriberService
 234{
 235    private readonly IAsyncResponseIngress _ingress;
 236
 237    /// <summary>Creates a worker subscriber for the configured Service Bus worker queue.</summary>
 238    public AzureServiceBusWorkerSubscriber(
 239        IOptions<AzureServiceBusAsyncResponseOptions> options,
 240        IAzureServiceBusClient client,
 241        IAsyncResponseIngress ingress,
 242        ILogger<AzureServiceBusWorkerSubscriber> logger)
 243        : base(options, client, logger)
 244    {
 245        _ingress = ingress;
 246    }
 247
 248    protected override string QueueName
 249        => AzureServiceBusOptionsValidator.Required(Options.WorkerQueue, nameof(Options.WorkerQueue));
 250
 251    protected override AzureServiceBusSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 252    protected override AzureServiceBusSubscriberRole SubscriberRole => AzureServiceBusSubscriberRole.Worker;
 253
 254    /// <summary>Handles the delivered message.</summary>
 255    protected override Task HandleMessageAsync(AzureServiceBusTransportDelivery delivery, CancellationToken cancellation
 256        => _ingress.HandleWorkerMessageAsync(delivery.Body);
 257}
 258
 259internal sealed class AzureServiceBusResponseIngressSubscriber : AzureServiceBusSubscriberService
 260{
 261    private readonly IAsyncResponseIngress _ingress;
 262
 263    /// <summary>Creates a response subscriber for the configured Service Bus response queue.</summary>
 264    public AzureServiceBusResponseIngressSubscriber(
 265        IOptions<AzureServiceBusAsyncResponseOptions> options,
 266        IAzureServiceBusClient client,
 267        IAsyncResponseIngress ingress,
 268        ILogger<AzureServiceBusResponseIngressSubscriber> logger)
 3269        : base(options, client, logger)
 270    {
 3271        _ingress = ingress;
 3272    }
 273
 274    protected override string QueueName
 3275        => AzureServiceBusOptionsValidator.Required(Options.ResponseQueue, nameof(Options.ResponseQueue));
 276
 3277    protected override AzureServiceBusSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 3278    protected override AzureServiceBusSubscriberRole SubscriberRole => AzureServiceBusSubscriberRole.ResponseIngress;
 279
 280    /// <summary>Handles the delivered message.</summary>
 281    protected override Task HandleMessageAsync(AzureServiceBusTransportDelivery delivery, CancellationToken cancellation
 282    {
 3283        var correlationId = AzureServiceBusCorrelationIdExtractor.Extract(delivery, delivery.Body, Options);
 3284        return _ingress.HandleResponseMessageAsync(delivery.Body, correlationId);
 285    }
 286}
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.AzureServiceBus.AzureServiceBusAsyncResponseOptions>, AsyncResponse.Transports.AzureServiceBus.IAzureServiceBusClient, AsyncResponse.IAsyncResponseIngress, Microsoft.Extensions.Logging.ILogger<AsyncResponse.Transports.AzureServiceBus.AzureServiceBusResponseIngressSubscriber>)
get_QueueName()
get_SubscriberOptions()
get_SubscriberRole()
HandleMessageAsync(AsyncResponse.Transports.AzureServiceBus.AzureServiceBusTransportDelivery, System.Threading.CancellationToken)