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

Information
Class: AsyncResponse.Transports.AzureServiceBus.AzureServiceBusSubscriberService
Assembly: AsyncResponse.Transports.AzureServiceBus
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.AzureServiceBus/AzureServiceBusSubscriberServices.cs
Line coverage
85%
Covered lines: 91
Uncovered lines: 15
Coverable lines: 106
Total lines: 286
Line coverage: 85.8%
Branch coverage
88%
Covered branches: 16
Total branches: 18
Branch coverage: 88.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%2294.74%
RunSubscriberAsync()100%22100%
DispatchBatchAsync()100%111076%
RenewLocksLoopAsync()75%5461.9%
get_SettledCount()100%11100%
MarkSettled()100%11100%
RetryDelay(...)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
 311    protected AzureServiceBusSubscriberService(
 312        IOptions<AzureServiceBusAsyncResponseOptions> options,
 313        IAzureServiceBusClient client,
 314        ILogger logger)
 15    {
 316        Options = options.Value;
 317        AzureServiceBusOptionsValidator.ValidateCommon(Options);
 318        _client = client;
 319        Logger = logger;
 320    }
 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    {
 334        var queue = QueueName;
 335        AzureServiceBusMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 336        var failures = 0;
 37
 338        while (!stoppingToken.IsCancellationRequested)
 39        {
 40            try
 41            {
 342                await RunSubscriberAsync(queue, stoppingToken).ConfigureAwait(false);
 343                return;
 44            }
 245            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 46            {
 047                return;
 48            }
 249            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 50            {
 251                failures++;
 252                var retryDelay = RetryDelay(failures);
 253                Logger.LogWarning(
 254                    ex,
 255                    "Azure Service Bus subscriber failed for queue {Queue} ({Role}); retrying in {RetryDelay}.",
 256                    queue,
 257                    SubscriberRole,
 258                    retryDelay);
 259                await Task.Delay(retryDelay, stoppingToken).ConfigureAwait(false);
 60            }
 61        }
 362    }
 63
 64    private async Task RunSubscriberAsync(string queue, CancellationToken stoppingToken)
 65    {
 366        await using var receiver = _client.CreateReceiver(queue, SubscriberOptions);
 367        await using var dispatcher = AzureServiceBusMessageDispatcher.Create(
 368            HandleMessageAsync,
 369            Options,
 370            SubscriberOptions,
 371            Logger,
 372            queue,
 373            SubscriberRole);
 74
 375        Logger.LogInformation(
 376            "Azure Service Bus subscriber started. Queue: {Queue}. Role: {Role}. AckMode: {AckMode}.",
 377            queue,
 378            SubscriberRole,
 379            SubscriberOptions.AckMode);
 80
 81        try
 82        {
 383            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.
 388                await dispatcher.WaitForCapacityAsync(stoppingToken).ConfigureAwait(false);
 389                var maxMessages = Math.Min(Options.MaxMessagesPerReceive, dispatcher.FreeCapacity);
 90
 391                var messages = await receiver.ReceiveMessagesAsync(
 392                    maxMessages,
 393                    Options.ReceiveWaitTime,
 394                    stoppingToken).ConfigureAwait(false);
 95
 396                await DispatchBatchAsync(dispatcher, messages, queue, stoppingToken).ConfigureAwait(false);
 97            }
 398        }
 399        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 100        {
 3101            using var shutdown = new CancellationTokenSource(Options.ShutdownTimeout);
 3102            await receiver.CloseAsync(shutdown.Token).ConfigureAwait(false);
 3103        }
 3104    }
 105
 106    private async Task DispatchBatchAsync(
 107        AzureServiceBusMessageDispatcher dispatcher,
 108        IReadOnlyList<AzureServiceBusTransportDelivery> messages,
 109        string queue,
 110        CancellationToken stoppingToken)
 111    {
 3112        if (messages.Count == 0)
 2113            return;
 114
 3115        if (SubscriberOptions.AckMode is not AzureServiceBusAckMode.AckAfterHandlerCompletes
 3116            || SubscriberOptions.LockRenewalInterval is not { } renewalInterval)
 117        {
 3118            foreach (var message in messages)
 3119                await dispatcher.HandleAsync(message, stoppingToken).ConfigureAwait(false);
 3120            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.
 3127        var progress = new BatchProgress();
 3128        using var renewalCancellation = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
 3129        var renewalTask = RenewLocksLoopAsync(messages, progress, renewalInterval, queue, renewalCancellation.Token);
 130        try
 131        {
 3132            foreach (var message in messages)
 133            {
 134                try
 135                {
 3136                    await dispatcher.HandleAsync(message, stoppingToken).ConfigureAwait(false);
 3137                }
 138                finally
 139                {
 3140                    progress.MarkSettled();
 141                }
 142            }
 143        }
 144        finally
 145        {
 3146            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.
 3153                await renewalTask.WaitAsync(Options.ShutdownTimeout).ConfigureAwait(false);
 3154            }
 0155            catch (TimeoutException)
 156            {
 0157                Logger.LogWarning(
 0158                    "Azure Service Bus lock renewal for {Queue} ({Role}) did not stop within the shutdown budget ({Shutd
 0159                    queue,
 0160                    SubscriberRole,
 0161                    Options.ShutdownTimeout);
 1162            }
 163        }
 3164    }
 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        {
 3175            while (true)
 176            {
 3177                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.
 2182                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.
 2186                    if (cancellationToken.IsCancellationRequested)
 0187                        return;
 188
 2189                    var message = messages[i];
 190                    try
 191                    {
 2192                        await message.RenewLockAsync(cancellationToken).ConfigureAwait(false);
 2193                    }
 2194                    catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
 195                    {
 196                        // Our token interrupted the in-flight renew; not a renewal failure.
 2197                        return;
 198                    }
 0199                    catch (Exception ex)
 200                    {
 0201                        Logger.LogWarning(
 0202                            ex,
 0203                            "Failed to renew the lock of Azure Service Bus message {MessageId} on {Queue}; it may redeli
 0204                            message.MessageId,
 0205                            queue);
 0206                    }
 2207                }
 208            }
 209        }
 3210        catch (OperationCanceledException)
 211        {
 212            // The batch finished or the subscriber is stopping.
 3213        }
 3214    }
 215
 216    private sealed class BatchProgress
 217    {
 218        private int _settledCount;
 219
 3220        public int SettledCount => Volatile.Read(ref _settledCount);
 221
 3222        public void MarkSettled() => Interlocked.Increment(ref _settledCount);
 223    }
 224
 225    private TimeSpan RetryDelay(int failures)
 226    {
 2227        var exponent = Math.Max(0, failures - 1);
 2228        var milliseconds = Options.SubscriberRetryBaseDelay.TotalMilliseconds * Math.Pow(2, exponent);
 3229        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)
 269        : base(options, client, logger)
 270    {
 271        _ingress = ingress;
 272    }
 273
 274    protected override string QueueName
 275        => AzureServiceBusOptionsValidator.Required(Options.ResponseQueue, nameof(Options.ResponseQueue));
 276
 277    protected override AzureServiceBusSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 278    protected override AzureServiceBusSubscriberRole SubscriberRole => AzureServiceBusSubscriberRole.ResponseIngress;
 279
 280    /// <summary>Handles the delivered message.</summary>
 281    protected override Task HandleMessageAsync(AzureServiceBusTransportDelivery delivery, CancellationToken cancellation
 282    {
 283        var correlationId = AzureServiceBusCorrelationIdExtractor.Extract(delivery, delivery.Body, Options);
 284        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, Microsoft.Extensions.Logging.ILogger)
ExecuteAsync()
RunSubscriberAsync()
DispatchBatchAsync()
RenewLocksLoopAsync()
get_SettledCount()
MarkSettled()
RetryDelay(int)