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

Information
Class: AsyncResponse.Transports.AzureServiceBus.AzureServiceBusSubscriberService
Assembly: AsyncResponse.Transports.AzureServiceBus
File(s): /_/src/Transports/AsyncResponse.Transports.AzureServiceBus/AzureServiceBusSubscriberServices.cs
Line coverage
85%
Covered lines: 85
Uncovered lines: 15
Coverable lines: 100
Total lines: 278
Line coverage: 85%
Branch coverage
90%
Covered branches: 18
Total branches: 20
Branch coverage: 90%
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%11100%
RunSubscriberAsync()100%22100%
DispatchBatchAsync()92.85%181472%
RenewLocksLoopAsync()75%5461.9%
get_SettledCount()100%11100%
MarkSettled()100%11100%

File(s)

/_/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
 41011    protected AzureServiceBusSubscriberService(
 41012        IOptions<AzureServiceBusAsyncResponseOptions> options,
 41013        IAzureServiceBusClient client,
 41014        ILogger logger)
 15    {
 41016        Options = options.Value;
 41017        AzureServiceBusOptionsValidator.ValidateCommon(Options);
 41018        _client = client;
 41019        Logger = logger;
 41020    }
 21
 677522    protected AzureServiceBusAsyncResponseOptions Options { get; }
 81823    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    /// <summary>
 33    /// Validates subscriber options here rather than at the top of <c>ExecuteAsync</c>: since
 34    /// Microsoft.Extensions.Hosting.Abstractions 10.0.10, <c>BackgroundService.StartAsync</c> no
 35    /// longer runs <c>ExecuteAsync</c> inline, so a throw there surfaces only through the host's
 36    /// background-exception handling — or never, when a fast stop discards the queued work —
 37    /// instead of failing host startup synchronously.
 38    /// </summary>
 39    public override Task StartAsync(CancellationToken cancellationToken)
 40    {
 40241        _ = QueueName; // Resolving the name enforces its Required check at startup too.
 40242        AzureServiceBusMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 40043        return base.StartAsync(cancellationToken);
 44    }
 45
 46    protected override Task ExecuteAsync(CancellationToken stoppingToken)
 47    {
 40048        var queue = QueueName;
 40049        return SubscriberSupervisor.RunAsync(
 40650            ct => RunSubscriberAsync(queue, ct),
 40051            stoppingToken,
 852            failures => AsyncResponseRetry.Backoff(failures, Options.SubscriberRetryBaseDelay, Options.SubscriberRetryMa
 40653            (ex, retryDelay) => Logger.LogWarning(
 40654                ex,
 40655                "Azure Service Bus subscriber failed for queue {Queue} ({Role}); retrying in {RetryDelay}.",
 40656                queue,
 40657                SubscriberRole,
 40658                retryDelay));
 59    }
 60
 61    private async Task RunSubscriberAsync(string queue, CancellationToken stoppingToken)
 62    {
 40663        await using var receiver = _client.CreateReceiver(queue, SubscriberOptions);
 40664        await using var dispatcher = AzureServiceBusMessageDispatcher.Create(
 40665            HandleMessageAsync,
 40666            Options,
 40667            SubscriberOptions,
 40668            Logger,
 40669            queue,
 40670            SubscriberRole);
 71
 40672        Logger.LogInformation(
 40673            "Azure Service Bus subscriber started. Queue: {Queue}. Role: {Role}. AckMode: {AckMode}.",
 40674            queue,
 40675            SubscriberRole,
 40676            SubscriberOptions.AckMode);
 77
 78        try
 79        {
 87580            while (!stoppingToken.IsCancellationRequested)
 81            {
 82                // In early-ACK mode, receiving while the background queue is saturated would burn
 83                // DeliveryCount via queue-full abandons, so wait for free capacity and never request
 84                // more messages than the dispatcher can accept.
 80085                await dispatcher.WaitForCapacityAsync(stoppingToken).ConfigureAwait(false);
 79886                var maxMessages = Math.Min(Options.MaxMessagesPerReceive, dispatcher.FreeCapacity);
 87
 79888                var messages = await receiver.ReceiveMessagesAsync(
 79889                    maxMessages,
 79890                    Options.ReceiveWaitTime,
 79891                    stoppingToken).ConfigureAwait(false);
 92
 46993                await DispatchBatchAsync(dispatcher, messages, queue, stoppingToken).ConfigureAwait(false);
 94            }
 7595        }
 32596        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 97        {
 32598            using var shutdown = new CancellationTokenSource(Options.ShutdownTimeout);
 32599            await receiver.CloseAsync(shutdown.Token).ConfigureAwait(false);
 325100        }
 400101    }
 102
 103    private async Task DispatchBatchAsync(
 104        AzureServiceBusMessageDispatcher dispatcher,
 105        IReadOnlyList<AzureServiceBusTransportDelivery> messages,
 106        string queue,
 107        CancellationToken stoppingToken)
 108    {
 469109        if (messages.Count == 0)
 63110            return;
 111
 406112        if (SubscriberOptions.AckMode is not AzureServiceBusAckMode.AckAfterHandlerCompletes
 406113            || SubscriberOptions.LockRenewalInterval is not { } renewalInterval)
 114        {
 48115            foreach (var message in messages)
 12116                await dispatcher.HandleAsync(message, stoppingToken).ConfigureAwait(false);
 12117            return;
 118        }
 119
 120        // The batch is processed serially, so a slow handler lets the peek locks of the later (still
 121        // unsettled) messages expire and Service Bus redelivers them to a competing consumer while
 122        // they are still queued here — systematic duplicate processing. While the batch is in flight,
 123        // a background loop renews the lock of every unsettled message each interval.
 394124        var progress = new BatchProgress();
 394125        using var renewalCancellation = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
 394126        var renewalTask = RenewLocksLoopAsync(messages, progress, renewalInterval, queue, renewalCancellation.Token);
 127        try
 128        {
 1632129            foreach (var message in messages)
 130            {
 131                try
 132                {
 422133                    await dispatcher.HandleAsync(message, stoppingToken).ConfigureAwait(false);
 422134                }
 135                finally
 136                {
 422137                    progress.MarkSettled();
 138                }
 139            }
 140        }
 141        finally
 142        {
 394143            renewalCancellation.Cancel();
 144            try
 145            {
 146                // Cancellation exits the sweep between messages and interrupts the in-flight renew
 147                // call, so this normally completes immediately. The bound is the hard backstop: an
 148                // unbounded await here would let a degraded namespace hold the receive loop (and, at
 149                // shutdown, the whole host budget) hostage for the SDK retry budget per message.
 394150                await renewalTask.WaitAsync(Options.ShutdownTimeout).ConfigureAwait(false);
 394151            }
 0152            catch (TimeoutException)
 153            {
 0154                Logger.LogWarning(
 0155                    "Azure Service Bus lock renewal for {Queue} ({Role}) did not stop within the shutdown budget ({Shutd
 0156                    queue,
 0157                    SubscriberRole,
 0158                    Options.ShutdownTimeout);
 0159            }
 160        }
 469161    }
 162
 163    private async Task RenewLocksLoopAsync(
 164        IReadOnlyList<AzureServiceBusTransportDelivery> messages,
 165        BatchProgress progress,
 166        TimeSpan renewalInterval,
 167        string queue,
 168        CancellationToken cancellationToken)
 169    {
 170        try
 171        {
 6172            while (true)
 173            {
 400174                await Task.Delay(renewalInterval, cancellationToken).ConfigureAwait(false);
 175
 176                // Renew from the first unsettled message onward: that covers the message currently in
 177                // the handler plus everything still waiting its turn. A renewal racing a just-settled
 178                // message merely fails and is logged; redelivery keeps at-least-once intact.
 40179                for (var i = progress.SettledCount; i < messages.Count; i++)
 180                {
 181                    // The batch finished or the subscriber is stopping: exit quietly between messages
 182                    // instead of spending up to a full SDK retry budget on each remaining renew.
 14183                    if (cancellationToken.IsCancellationRequested)
 0184                        return;
 185
 14186                    var message = messages[i];
 187                    try
 188                    {
 14189                        await message.RenewLockAsync(cancellationToken).ConfigureAwait(false);
 12190                    }
 2191                    catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
 192                    {
 193                        // Our token interrupted the in-flight renew; not a renewal failure.
 2194                        return;
 195                    }
 0196                    catch (Exception ex)
 197                    {
 0198                        Logger.LogWarning(
 0199                            ex,
 0200                            "Failed to renew the lock of Azure Service Bus message {MessageId} on {Queue}; it may redeli
 0201                            message.MessageId,
 0202                            queue);
 0203                    }
 12204                }
 205            }
 206        }
 392207        catch (OperationCanceledException)
 208        {
 209            // The batch finished or the subscriber is stopping.
 392210        }
 394211    }
 212
 213    private sealed class BatchProgress
 214    {
 215        private int _settledCount;
 216
 8217        public int SettledCount => Volatile.Read(ref _settledCount);
 218
 422219        public void MarkSettled() => Interlocked.Increment(ref _settledCount);
 220    }
 221}
 222
 223internal sealed class AzureServiceBusWorkerSubscriber : AzureServiceBusSubscriberService
 224{
 225    private readonly IAsyncResponseIngress _ingress;
 226
 227    /// <summary>Creates a worker subscriber for the configured Service Bus worker queue.</summary>
 228    public AzureServiceBusWorkerSubscriber(
 229        IOptions<AzureServiceBusAsyncResponseOptions> options,
 230        IAzureServiceBusClient client,
 231        IAsyncResponseIngress ingress,
 232        ILogger<AzureServiceBusWorkerSubscriber> logger)
 233        : base(options, client, logger)
 234    {
 235        _ingress = ingress;
 236    }
 237
 238    protected override string QueueName
 239        => AzureServiceBusOptionsValidator.Required(Options.WorkerQueue, nameof(Options.WorkerQueue));
 240
 241    protected override AzureServiceBusSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 242    protected override AzureServiceBusSubscriberRole SubscriberRole => AzureServiceBusSubscriberRole.Worker;
 243
 244    /// <summary>Handles the delivered message.</summary>
 245    protected override Task HandleMessageAsync(AzureServiceBusTransportDelivery delivery, CancellationToken cancellation
 246        => _ingress.HandleWorkerMessageAsync(delivery.Body);
 247}
 248
 249internal sealed class AzureServiceBusResponseIngressSubscriber : AzureServiceBusSubscriberService
 250{
 251    private readonly IAsyncResponseIngress _ingress;
 252
 253    /// <summary>Creates a response subscriber for the configured Service Bus response queue.</summary>
 254    public AzureServiceBusResponseIngressSubscriber(
 255        IOptions<AzureServiceBusAsyncResponseOptions> options,
 256        IAzureServiceBusClient client,
 257        IAsyncResponseIngress ingress,
 258        ILogger<AzureServiceBusResponseIngressSubscriber> logger)
 259        : base(options, client, logger)
 260    {
 261        _ingress = ingress;
 262    }
 263
 264    protected override string QueueName
 265        => AzureServiceBusOptionsValidator.Required(Options.ResponseQueue, nameof(Options.ResponseQueue));
 266
 267    protected override AzureServiceBusSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 268    protected override AzureServiceBusSubscriberRole SubscriberRole => AzureServiceBusSubscriberRole.ResponseIngress;
 269
 270    /// <summary>Handles the delivered message.</summary>
 271    protected override Task HandleMessageAsync(AzureServiceBusTransportDelivery delivery, CancellationToken cancellation
 272    {
 273        var correlationId = !_ingress.IsOverInboundBudget(delivery.Body)
 274            ? AzureServiceBusCorrelationIdExtractor.Extract(delivery, delivery.Body, Options)
 275            : null;
 276        return _ingress.HandleResponseMessageAsync(delivery.Body, correlationId);
 277    }
 278}
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.AzureServiceBus.AzureServiceBusAsyncResponseOptions>,AsyncResponse.Transports.AzureServiceBus.IAzureServiceBusClient,Microsoft.Extensions.Logging.ILogger)
get_Options()
get_Logger()
StartAsync(System.Threading.CancellationToken)
ExecuteAsync(System.Threading.CancellationToken)
RunSubscriberAsync()
DispatchBatchAsync()
RenewLocksLoopAsync()
get_SettledCount()
MarkSettled()