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

Information
Class: AsyncResponse.Transports.AzureServiceBus.AzureServiceBusOutboundMessage
Assembly: AsyncResponse.Transports.AzureServiceBus
File(s): /_/src/Transports/AsyncResponse.Transports.AzureServiceBus/AzureServiceBusClientAdapters.cs
Line coverage
100%
Covered lines: 6
Uncovered lines: 0
Coverable lines: 6
Total lines: 225
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_Body()100%11100%
get_MessageId()100%11100%
get_CorrelationId()100%11100%
get_ApplicationProperties()100%11100%
get_ScheduledEnqueueTime()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
 101internal sealed class AzureServiceBusReceiverAdapter(
 102    ServiceBusReceiver inner,
 103    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    {
 111        var messages = await inner.ReceiveMessagesAsync(maxMessages, maxWaitTime, cancellationToken).ConfigureAwait(fals
 112        if (messages.Count == 0)
 113            return [];
 114
 115        var queue = queueOverride ?? inner.EntityPath;
 116        var deliveries = new List<AzureServiceBusTransportDelivery>(messages.Count);
 117        for (var i = 0; i < messages.Count; i++)
 118        {
 119            var message = messages[i];
 120            try
 121            {
 122                deliveries.Add(CreateDelivery(queue, message));
 123            }
 124            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.
 132                await DeadLetterUnprojectableAsync(message, ex).ConfigureAwait(false);
 133            }
 134        }
 135
 136        return deliveries;
 137    }
 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        {
 148            await inner.DeadLetterMessageAsync(
 149                message,
 150                deadLetterReason: "AsyncResponseUnsupportedBody",
 151                deadLetterErrorDescription: cause.GetType().Name,
 152                cancellationToken: CancellationToken.None).ConfigureAwait(false);
 153        }
 154        catch
 155        {
 156            // Swallowed deliberately: the caller is mid-batch and the alternative is losing the
 157            // deliveries already projected.
 158        }
 159    }
 160
 161    private AzureServiceBusTransportDelivery CreateDelivery(
 162        string queue,
 163        ServiceBusReceivedMessage message)
 164        => new(
 165            queue,
 166            message.Body.ToString(),
 167            message.MessageId,
 168            message.CorrelationId,
 169            message.SequenceNumber,
 170            message.DeliveryCount,
 171            CopyApplicationProperties(message.ApplicationProperties),
 172            () => new ValueTask(inner.CompleteMessageAsync(message, CancellationToken.None)),
 173            () => new ValueTask(inner.AbandonMessageAsync(message, cancellationToken: CancellationToken.None)),
 174            (reason, description) => new ValueTask(inner.DeadLetterMessageAsync(
 175                message,
 176                deadLetterReason: reason,
 177                deadLetterErrorDescription: description,
 178                cancellationToken: CancellationToken.None)),
 179            // Settlement deliberately ignores cancellation so an in-flight message still settles
 180            // during shutdown. Lock renewal is a background courtesy and honors the caller's token:
 181            // on a degraded namespace each renew otherwise burns the SDK's full retry budget, and
 182            // the renewal loop must be interruptible mid-call for the batch (and shutdown) to
 183            // complete promptly.
 184            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    {
 193        var properties = new Dictionary<string, object?>(applicationProperties.Count, StringComparer.OrdinalIgnoreCase);
 194        foreach (var property in applicationProperties)
 195            properties[property.Key] = property.Value;
 196        return properties;
 197    }
 198
 199    /// <summary>Closes the receiver link.</summary>
 200    public Task CloseAsync(CancellationToken cancellationToken = default)
 201        => inner.CloseAsync(cancellationToken);
 202
 203    /// <summary>Releases resources held by this instance.</summary>
 204    public ValueTask DisposeAsync() => inner.DisposeAsync();
 205}
 206
 431207internal sealed record AzureServiceBusOutboundMessage(
 411208    string Body,
 419209    string MessageId,
 417210    string? CorrelationId,
 413211    IReadOnlyDictionary<string, object?> ApplicationProperties,
 840212    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(System.String,System.String,System.String,System.Collections.Generic.IReadOnlyDictionary`2<System.String,System.Object>,System.Nullable`1<System.DateTimeOffset>)
get_Body()
get_MessageId()
get_CorrelationId()
get_ApplicationProperties()
get_ScheduledEnqueueTime()