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

Information
Class: AsyncResponse.Transports.AzureServiceBus.AzureServiceBusOutboundMessage
Assembly: AsyncResponse.Transports.AzureServiceBus
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.AzureServiceBus/AzureServiceBusClientAdapters.cs
Line coverage
100%
Covered lines: 5
Uncovered lines: 0
Coverable lines: 5
Total lines: 167
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%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/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        foreach (var property in message.ApplicationProperties)
 72            serviceBusMessage.ApplicationProperties[property.Key] = property.Value;
 73
 74        return inner.SendMessageAsync(serviceBusMessage, cancellationToken);
 75    }
 76
 77    /// <summary>Closes the sender link.</summary>
 78    public Task CloseAsync(CancellationToken cancellationToken = default)
 79        => inner.CloseAsync(cancellationToken);
 80
 81    /// <summary>Releases resources held by this instance.</summary>
 82    public ValueTask DisposeAsync() => inner.DisposeAsync();
 83}
 84
 85internal interface IAzureServiceBusReceiver : IAsyncDisposable
 86{
 87    Task<IReadOnlyList<AzureServiceBusTransportDelivery>> ReceiveMessagesAsync(
 88        int maxMessages,
 89        TimeSpan maxWaitTime,
 90        CancellationToken cancellationToken = default);
 91
 92    Task CloseAsync(CancellationToken cancellationToken = default);
 93}
 94
 95internal sealed class AzureServiceBusReceiverAdapter(
 96    ServiceBusReceiver inner,
 97    string? queueOverride = null) : IAzureServiceBusReceiver
 98{
 99    /// <summary>Receives and wraps messages from Service Bus.</summary>
 100    public async Task<IReadOnlyList<AzureServiceBusTransportDelivery>> ReceiveMessagesAsync(
 101        int maxMessages,
 102        TimeSpan maxWaitTime,
 103        CancellationToken cancellationToken = default)
 104    {
 105        var messages = await inner.ReceiveMessagesAsync(maxMessages, maxWaitTime, cancellationToken).ConfigureAwait(fals
 106        if (messages.Count == 0)
 107            return [];
 108
 109        var queue = queueOverride ?? inner.EntityPath;
 110        var deliveries = new AzureServiceBusTransportDelivery[messages.Count];
 111        for (var i = 0; i < messages.Count; i++)
 112            deliveries[i] = CreateDelivery(queue, messages[i]);
 113
 114        return deliveries;
 115    }
 116
 117    private AzureServiceBusTransportDelivery CreateDelivery(
 118        string queue,
 119        ServiceBusReceivedMessage message)
 120        => new(
 121            queue,
 122            message.Body.ToString(),
 123            message.MessageId,
 124            message.CorrelationId,
 125            message.SequenceNumber,
 126            message.DeliveryCount,
 127            new Dictionary<string, object?>(message.ApplicationProperties, StringComparer.OrdinalIgnoreCase),
 128            () => new ValueTask(inner.CompleteMessageAsync(message, CancellationToken.None)),
 129            () => new ValueTask(inner.AbandonMessageAsync(message, cancellationToken: CancellationToken.None)),
 130            (reason, description) => new ValueTask(inner.DeadLetterMessageAsync(
 131                message,
 132                deadLetterReason: reason,
 133                deadLetterErrorDescription: description,
 134                cancellationToken: CancellationToken.None)),
 135            // Settlement deliberately ignores cancellation so an in-flight message still settles
 136            // during shutdown. Lock renewal is a background courtesy and honors the caller's token:
 137            // on a degraded namespace each renew otherwise burns the SDK's full retry budget, and
 138            // the renewal loop must be interruptible mid-call for the batch (and shutdown) to
 139            // complete promptly.
 140            cancellationToken => new ValueTask(inner.RenewMessageLockAsync(message, cancellationToken)));
 141
 142    /// <summary>Closes the receiver link.</summary>
 143    public Task CloseAsync(CancellationToken cancellationToken = default)
 144        => inner.CloseAsync(cancellationToken);
 145
 146    /// <summary>Releases resources held by this instance.</summary>
 147    public ValueTask DisposeAsync() => inner.DisposeAsync();
 148}
 149
 3150internal sealed record AzureServiceBusOutboundMessage(
 3151    string Body,
 3152    string MessageId,
 3153    string? CorrelationId,
 3154    IReadOnlyDictionary<string, object?> ApplicationProperties);
 155
 156internal sealed record AzureServiceBusTransportDelivery(
 157    string Queue,
 158    string Body,
 159    string MessageId,
 160    string? CorrelationId,
 161    long SequenceNumber,
 162    int DeliveryCount,
 163    IReadOnlyDictionary<string, object?> ApplicationProperties,
 164    Func<ValueTask> CompleteAsync,
 165    Func<ValueTask> AbandonAsync,
 166    Func<string, string?, ValueTask> DeadLetterAsync,
 167    Func<CancellationToken, ValueTask> RenewLockAsync);
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(string, string, string, System.Collections.Generic.IReadOnlyDictionary<string, object>)