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

Information
Class: AsyncResponse.Transports.AzureServiceBus.AzureServiceBusReceiverAdapter
Assembly: AsyncResponse.Transports.AzureServiceBus
File(s): /_/src/Transports/AsyncResponse.Transports.AzureServiceBus/AzureServiceBusClientAdapters.cs
Line coverage
100%
Covered lines: 52
Uncovered lines: 0
Coverable lines: 52
Total lines: 225
Line coverage: 100%
Branch coverage
100%
Covered branches: 8
Total branches: 8
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
ReceiveMessagesAsync()100%66100%
DeadLetterUnprojectableAsync()100%11100%
CreateDelivery(...)100%11100%
CopyApplicationProperties(...)100%22100%
CloseAsync(...)100%11100%
DisposeAsync()100%11100%

File(s)

/_/src/Transports/AsyncResponse.Transports.AzureServiceBus/AzureServiceBusClientAdapters.cs

#LineLine coverage
 1using Azure.Messaging.ServiceBus;
 2using Microsoft.Extensions.DependencyInjection;
 3
 4namespace AsyncResponse.Transports.AzureServiceBus;
 5
 6internal interface IAzureServiceBusClient : IAsyncDisposable
 7{
 8    IAzureServiceBusSender CreateSender(string queue);
 9    IAzureServiceBusReceiver CreateReceiver(string queue, AzureServiceBusSubscriberOptions subscriberOptions);
 10}
 11
 12internal sealed class AzureServiceBusClientAdapter(
 13    ServiceBusClient inner,
 14    bool ownsClient) : IAzureServiceBusClient
 15{
 16    /// <summary>Creates a sender for the requested queue.</summary>
 17    public IAzureServiceBusSender CreateSender(string queue)
 18        => new AzureServiceBusSenderAdapter(inner.CreateSender(queue));
 19
 20    /// <summary>Creates a peek-lock receiver for the requested queue.</summary>
 21    public IAzureServiceBusReceiver CreateReceiver(
 22        string queue,
 23        AzureServiceBusSubscriberOptions subscriberOptions)
 24        => new AzureServiceBusReceiverAdapter(inner.CreateReceiver(
 25            queue,
 26            new ServiceBusReceiverOptions
 27            {
 28                ReceiveMode = ServiceBusReceiveMode.PeekLock,
 29                PrefetchCount = subscriberOptions.PrefetchCount
 30            }));
 31
 32    /// <summary>Releases resources held by this instance.</summary>
 33    public async ValueTask DisposeAsync()
 34    {
 35        if (ownsClient)
 36            await inner.DisposeAsync().ConfigureAwait(false);
 37    }
 38}
 39
 40internal static class AzureServiceBusClientResolver
 41{
 42    public static IAzureServiceBusClient Create(IServiceProvider provider)
 43    {
 44        if (provider.GetService<ServiceBusClient>() is { } registeredClient)
 45            return new AzureServiceBusClientAdapter(registeredClient, ownsClient: false);
 46
 47        var options = provider.GetRequiredService<Microsoft.Extensions.Options.IOptions<AzureServiceBusAsyncResponseOpti
 48        var connectionString = AzureServiceBusOptionsValidator.Required(options.ConnectionString, nameof(options.Connect
 49        return new AzureServiceBusClientAdapter(new ServiceBusClient(connectionString), ownsClient: true);
 50    }
 51}
 52
 53internal interface IAzureServiceBusSender : IAsyncDisposable
 54{
 55    Task SendMessageAsync(AzureServiceBusOutboundMessage message, CancellationToken cancellationToken = default);
 56    Task CloseAsync(CancellationToken cancellationToken = default);
 57}
 58
 59internal sealed class AzureServiceBusSenderAdapter(ServiceBusSender inner) : IAzureServiceBusSender
 60{
 61    /// <summary>Sends the supplied outbound message.</summary>
 62    public Task SendMessageAsync(AzureServiceBusOutboundMessage message, CancellationToken cancellationToken = default)
 63    {
 64        var serviceBusMessage = new ServiceBusMessage(BinaryData.FromString(message.Body))
 65        {
 66            ContentType = "application/json",
 67            MessageId = message.MessageId,
 68            CorrelationId = message.CorrelationId
 69        };
 70
 71        // Native delayed delivery: the broker holds a scheduled message and enqueues it at the
 72        // requested instant — the message survives restarts on the broker, unlike any client-side
 73        // timer.
 74        if (message.ScheduledEnqueueTime is { } scheduledEnqueueTime)
 75            serviceBusMessage.ScheduledEnqueueTime = scheduledEnqueueTime;
 76
 77        foreach (var property in message.ApplicationProperties)
 78            serviceBusMessage.ApplicationProperties[property.Key] = property.Value;
 79
 80        return inner.SendMessageAsync(serviceBusMessage, cancellationToken);
 81    }
 82
 83    /// <summary>Closes the sender link.</summary>
 84    public Task CloseAsync(CancellationToken cancellationToken = default)
 85        => inner.CloseAsync(cancellationToken);
 86
 87    /// <summary>Releases resources held by this instance.</summary>
 88    public ValueTask DisposeAsync() => inner.DisposeAsync();
 89}
 90
 91internal interface IAzureServiceBusReceiver : IAsyncDisposable
 92{
 93    Task<IReadOnlyList<AzureServiceBusTransportDelivery>> ReceiveMessagesAsync(
 94        int maxMessages,
 95        TimeSpan maxWaitTime,
 96        CancellationToken cancellationToken = default);
 97
 98    Task CloseAsync(CancellationToken cancellationToken = default);
 99}
 100
 396101internal sealed class AzureServiceBusReceiverAdapter(
 396102    ServiceBusReceiver inner,
 396103    string? queueOverride = null) : IAzureServiceBusReceiver
 104{
 105    /// <summary>Receives and wraps messages from Service Bus.</summary>
 106    public async Task<IReadOnlyList<AzureServiceBusTransportDelivery>> ReceiveMessagesAsync(
 107        int maxMessages,
 108        TimeSpan maxWaitTime,
 109        CancellationToken cancellationToken = default)
 110    {
 769111        var messages = await inner.ReceiveMessagesAsync(maxMessages, maxWaitTime, cancellationToken).ConfigureAwait(fals
 457112        if (messages.Count == 0)
 63113            return [];
 114
 394115        var queue = queueOverride ?? inner.EntityPath;
 394116        var deliveries = new List<AzureServiceBusTransportDelivery>(messages.Count);
 1632117        for (var i = 0; i < messages.Count; i++)
 118        {
 422119            var message = messages[i];
 120            try
 121            {
 422122                deliveries.Add(CreateDelivery(queue, message));
 418123            }
 4124            catch (Exception ex) when (ex is not OperationCanceledException)
 125            {
 126                // Projecting the message must not be able to abort the RECEIVE. The Body getter
 127                // throws for an AMQP Value/Sequence body — what a JMS or raw-AMQP producer sends —
 128                // and the throw used to escape the whole batch: nothing was settled, all N locks
 129                // lapsed, all N DeliveryCounts advanced, and the poison message crashed the loop
 130                // again next cycle while its innocent batch-mates burned attempts toward the
 131                // entity's MaxDeliveryCount without ever running. Bury this one and keep the rest.
 4132                await DeadLetterUnprojectableAsync(message, ex).ConfigureAwait(false);
 133            }
 134        }
 135
 394136        return deliveries;
 457137    }
 138
 139    /// <summary>
 140    /// Buries a message this adapter cannot project (an unsupported AMQP body type). Best-effort:
 141    /// if the dead-letter itself fails the lock simply lapses and the broker redelivers, which is
 142    /// still strictly better than tearing down the receive loop.
 143    /// </summary>
 144    private async Task DeadLetterUnprojectableAsync(ServiceBusReceivedMessage message, Exception cause)
 145    {
 146        try
 147        {
 4148            await inner.DeadLetterMessageAsync(
 4149                message,
 4150                deadLetterReason: "AsyncResponseUnsupportedBody",
 4151                deadLetterErrorDescription: cause.GetType().Name,
 4152                cancellationToken: CancellationToken.None).ConfigureAwait(false);
 2153        }
 2154        catch
 155        {
 156            // Swallowed deliberately: the caller is mid-batch and the alternative is losing the
 157            // deliveries already projected.
 2158        }
 4159    }
 160
 161    private AzureServiceBusTransportDelivery CreateDelivery(
 162        string queue,
 163        ServiceBusReceivedMessage message)
 422164        => new(
 422165            queue,
 422166            message.Body.ToString(),
 422167            message.MessageId,
 422168            message.CorrelationId,
 422169            message.SequenceNumber,
 422170            message.DeliveryCount,
 422171            CopyApplicationProperties(message.ApplicationProperties),
 408172            () => new ValueTask(inner.CompleteMessageAsync(message, CancellationToken.None)),
 5173            () => new ValueTask(inner.AbandonMessageAsync(message, cancellationToken: CancellationToken.None)),
 3174            (reason, description) => new ValueTask(inner.DeadLetterMessageAsync(
 3175                message,
 3176                deadLetterReason: reason,
 3177                deadLetterErrorDescription: description,
 3178                cancellationToken: CancellationToken.None)),
 422179            // Settlement deliberately ignores cancellation so an in-flight message still settles
 422180            // during shutdown. Lock renewal is a background courtesy and honors the caller's token:
 422181            // on a degraded namespace each renew otherwise burns the SDK's full retry budget, and
 422182            // the renewal loop must be interruptible mid-call for the batch (and shutdown) to
 422183            // complete promptly.
 424184            cancellationToken => new ValueTask(inner.RenewMessageLockAsync(message, cancellationToken)));
 185
 186    // Indexer, not the copying constructor: AMQP application-property names are case-sensitive,
 187    // so a message legally carries keys differing only in case — the constructor's internal Add
 188    // would throw ArgumentException out of the receive path before any delivery in the batch is
 189    // settled, stalling the whole batch. Last-seen wins under the case-insensitive comparer the
 190    // lookups rely on (same shape as the SQS and Kafka adapters).
 191    private static Dictionary<string, object?> CopyApplicationProperties(IReadOnlyDictionary<string, object> application
 192    {
 418193        var properties = new Dictionary<string, object?>(applicationProperties.Count, StringComparer.OrdinalIgnoreCase);
 1044194        foreach (var property in applicationProperties)
 104195            properties[property.Key] = property.Value;
 418196        return properties;
 197    }
 198
 199    /// <summary>Closes the receiver link.</summary>
 200    public Task CloseAsync(CancellationToken cancellationToken = default)
 314201        => inner.CloseAsync(cancellationToken);
 202
 203    /// <summary>Releases resources held by this instance.</summary>
 388204    public ValueTask DisposeAsync() => inner.DisposeAsync();
 205}
 206
 207internal sealed record AzureServiceBusOutboundMessage(
 208    string Body,
 209    string MessageId,
 210    string? CorrelationId,
 211    IReadOnlyDictionary<string, object?> ApplicationProperties,
 212    DateTimeOffset? ScheduledEnqueueTime = null);
 213
 214internal sealed record AzureServiceBusTransportDelivery(
 215    string Queue,
 216    string Body,
 217    string MessageId,
 218    string? CorrelationId,
 219    long SequenceNumber,
 220    int DeliveryCount,
 221    IReadOnlyDictionary<string, object?> ApplicationProperties,
 222    Func<ValueTask> CompleteAsync,
 223    Func<ValueTask> AbandonAsync,
 224    Func<string, string?, ValueTask> DeadLetterAsync,
 225    Func<CancellationToken, ValueTask> RenewLockAsync);
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Azure.Messaging.ServiceBus.ServiceBusReceiver,System.String)
ReceiveMessagesAsync()
DeadLetterUnprojectableAsync()
CreateDelivery(System.String,Azure.Messaging.ServiceBus.ServiceBusReceivedMessage)
CopyApplicationProperties(System.Collections.Generic.IReadOnlyDictionary`2<System.String,System.Object>)
CloseAsync(System.Threading.CancellationToken)
DisposeAsync()