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

Information
Class: AsyncResponse.Transports.SQS.SqsClientAdapter
Assembly: AsyncResponse.Transports.SQS
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SQS/SqsClientAdapters.cs
Line coverage
100%
Covered lines: 89
Uncovered lines: 0
Coverable lines: 89
Total lines: 226
Line coverage: 100%
Branch coverage
100%
Covered branches: 40
Total branches: 40
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%
GetQueueUrlAsync()100%11100%
CreateQueueAsync()100%44100%
GetQueueArnAsync()100%11100%
SetQueueAttributesAsync(...)100%44100%
SendMessageAsync()100%44100%
ReceiveMessagesAsync()100%88100%
CreateDelivery(...)100%1818100%
DisposeAsync()100%22100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SQS/SqsClientAdapters.cs

#LineLine coverage
 1using Amazon;
 2using Amazon.Runtime;
 3using Amazon.SQS;
 4using Amazon.SQS.Model;
 5using Microsoft.Extensions.DependencyInjection;
 6
 7namespace AsyncResponse.Transports.SQS;
 8
 9internal interface ISqsClient : IAsyncDisposable
 10{
 11    Task<string> GetQueueUrlAsync(string queueName, CancellationToken cancellationToken = default);
 12    Task<string> CreateQueueAsync(string queueName, IReadOnlyDictionary<string, string> attributes, CancellationToken ca
 13    Task<string> GetQueueArnAsync(string queueUrl, CancellationToken cancellationToken = default);
 14    Task SetQueueAttributesAsync(string queueUrl, IReadOnlyDictionary<string, string> attributes, CancellationToken canc
 15    Task<string> SendMessageAsync(SqsOutboundMessage message, CancellationToken cancellationToken = default);
 16    Task<IReadOnlyList<SqsTransportDelivery>> ReceiveMessagesAsync(SqsReceiveRequest request, CancellationToken cancella
 17}
 18
 319internal sealed class SqsClientAdapter(
 320    IAmazonSQS inner,
 321    bool ownsClient) : ISqsClient
 22{
 23    /// <summary>Resolves a queue name to its queue URL.</summary>
 24    public async Task<string> GetQueueUrlAsync(string queueName, CancellationToken cancellationToken = default)
 25    {
 326        var response = await inner.GetQueueUrlAsync(queueName, cancellationToken).ConfigureAwait(false);
 327        return response.QueueUrl;
 328    }
 29
 30    /// <summary>Creates the queue (idempotent for identical attributes) and returns its URL.</summary>
 31    public async Task<string> CreateQueueAsync(
 32        string queueName,
 33        IReadOnlyDictionary<string, string> attributes,
 34        CancellationToken cancellationToken = default)
 35    {
 336        var request = new CreateQueueRequest { QueueName = queueName };
 337        foreach (var attribute in attributes)
 338            (request.Attributes ??= []).Add(attribute.Key, attribute.Value);
 39
 340        var response = await inner.CreateQueueAsync(request, cancellationToken).ConfigureAwait(false);
 341        return response.QueueUrl;
 342    }
 43
 44    /// <summary>Reads the queue's ARN attribute.</summary>
 45    public async Task<string> GetQueueArnAsync(string queueUrl, CancellationToken cancellationToken = default)
 46    {
 347        var response = await inner.GetQueueAttributesAsync(
 348            new GetQueueAttributesRequest
 349            {
 350                QueueUrl = queueUrl,
 351                AttributeNames = [QueueAttributeName.QueueArn]
 352            },
 353            cancellationToken).ConfigureAwait(false);
 354        return response.QueueARN;
 355    }
 56
 57    /// <summary>Applies the supplied attributes to an existing queue.</summary>
 58    public Task SetQueueAttributesAsync(
 59        string queueUrl,
 60        IReadOnlyDictionary<string, string> attributes,
 61        CancellationToken cancellationToken = default)
 262        => inner.SetQueueAttributesAsync(
 263            new SetQueueAttributesRequest
 264            {
 265                QueueUrl = queueUrl,
 266                Attributes = attributes.ToDictionary(pair => pair.Key, pair => pair.Value, StringComparer.Ordinal)
 267            },
 268            cancellationToken);
 69
 70    /// <summary>Sends the supplied outbound message and returns the SQS-assigned message id.</summary>
 71    public async Task<string> SendMessageAsync(SqsOutboundMessage message, CancellationToken cancellationToken = default
 72    {
 373        var request = new SendMessageRequest
 374        {
 375            QueueUrl = message.QueueUrl,
 376            MessageBody = message.Body,
 377            MessageGroupId = message.MessageGroupId,
 378            MessageDeduplicationId = message.MessageDeduplicationId
 379        };
 80
 381        foreach (var attribute in message.MessageAttributes)
 82        {
 383            (request.MessageAttributes ??= []).Add(attribute.Key, new MessageAttributeValue
 384            {
 385                DataType = "String",
 386                StringValue = attribute.Value
 387            });
 88        }
 89
 390        var response = await inner.SendMessageAsync(request, cancellationToken).ConfigureAwait(false);
 391        return response.MessageId;
 392    }
 93
 94    /// <summary>Long-polls the queue and wraps the received messages as transport deliveries.</summary>
 95    public async Task<IReadOnlyList<SqsTransportDelivery>> ReceiveMessagesAsync(
 96        SqsReceiveRequest request,
 97        CancellationToken cancellationToken = default)
 98    {
 399        var receive = new ReceiveMessageRequest
 3100        {
 3101            QueueUrl = request.QueueUrl,
 3102            MaxNumberOfMessages = request.MaxMessages,
 3103            WaitTimeSeconds = (int)request.WaitTime.TotalSeconds,
 3104            MessageSystemAttributeNames = [MessageSystemAttributeName.ApproximateReceiveCount],
 3105            MessageAttributeNames = ["All"]
 3106        };
 3107        if (request.VisibilityTimeout is { } visibilityTimeout)
 2108            receive.VisibilityTimeout = (int)visibilityTimeout.TotalSeconds;
 109
 3110        var response = await inner.ReceiveMessageAsync(receive, cancellationToken).ConfigureAwait(false);
 111        // AWS SDK v4 leaves collections null when the response carries no items.
 3112        if (response.Messages is not { Count: > 0 } messages)
 3113            return [];
 114
 3115        var deliveries = new SqsTransportDelivery[messages.Count];
 3116        for (var i = 0; i < messages.Count; i++)
 3117            deliveries[i] = CreateDelivery(request.QueueUrl, messages[i]);
 118
 3119        return deliveries;
 3120    }
 121
 122    private SqsTransportDelivery CreateDelivery(string queueUrl, Message message)
 123    {
 3124        var receiveCount = 1;
 3125        if (message.Attributes is { } systemAttributes
 3126            && systemAttributes.TryGetValue(MessageSystemAttributeName.ApproximateReceiveCount, out var rawReceiveCount)
 3127            && int.TryParse(rawReceiveCount, out var parsedReceiveCount))
 128        {
 3129            receiveCount = parsedReceiveCount;
 130        }
 131
 3132        var messageAttributes = new Dictionary<string, string>(StringComparer.Ordinal);
 3133        if (message.MessageAttributes is { } attributes)
 134        {
 3135            foreach (var attribute in attributes)
 136            {
 3137                if (attribute.Value?.StringValue is { } value)
 3138                    messageAttributes[attribute.Key] = value;
 139            }
 140        }
 141
 3142        var receiptHandle = message.ReceiptHandle;
 3143        return new SqsTransportDelivery(
 3144            queueUrl,
 3145            message.Body ?? string.Empty,
 3146            message.MessageId ?? string.Empty,
 3147            receiptHandle,
 3148            receiveCount,
 3149            messageAttributes,
 3150            () => new ValueTask(inner.DeleteMessageAsync(queueUrl, receiptHandle, CancellationToken.None)),
 2151            delay => new ValueTask(inner.ChangeMessageVisibilityAsync(
 2152                queueUrl,
 2153                receiptHandle,
 2154                (int)delay.TotalSeconds,
 2155                CancellationToken.None)));
 156    }
 157
 158    /// <summary>Releases resources held by this instance.</summary>
 159    public ValueTask DisposeAsync()
 160    {
 3161        if (ownsClient)
 3162            inner.Dispose();
 3163        return ValueTask.CompletedTask;
 164    }
 165}
 166
 167internal static class SqsClientFactory
 168{
 169    /// <summary>Builds an SQS client from the transport options (endpoint, region, credentials).</summary>
 170    public static ISqsClient Create(SqsAsyncResponseOptions options)
 171    {
 172        var config = new AmazonSQSConfig();
 173        if (!string.IsNullOrWhiteSpace(options.ServiceUrl))
 174        {
 175            config.ServiceURL = options.ServiceUrl;
 176            // Custom endpoints (LocalStack, proxies) still need a signing region.
 177            config.AuthenticationRegion = options.Region ?? "us-east-1";
 178        }
 179        else if (!string.IsNullOrWhiteSpace(options.Region))
 180        {
 181            config.RegionEndpoint = RegionEndpoint.GetBySystemName(options.Region);
 182        }
 183
 184        var client = !string.IsNullOrWhiteSpace(options.AccessKey) && !string.IsNullOrWhiteSpace(options.SecretKey)
 185            ? new AmazonSQSClient(new BasicAWSCredentials(options.AccessKey, options.SecretKey), config)
 186            : new AmazonSQSClient(config);
 187        return new SqsClientAdapter(client, ownsClient: true);
 188    }
 189}
 190
 191internal static class SqsClientResolver
 192{
 193    /// <summary>Reuses an application-registered <see cref="IAmazonSQS"/> or builds one from the options.</summary>
 194    public static ISqsClient Create(IServiceProvider provider)
 195    {
 196        if (provider.GetService<IAmazonSQS>() is { } registeredClient)
 197            return new SqsClientAdapter(registeredClient, ownsClient: false);
 198
 199        var options = provider.GetRequiredService<Microsoft.Extensions.Options.IOptions<SqsAsyncResponseOptions>>().Valu
 200        return SqsClientFactory.Create(options);
 201    }
 202}
 203
 204internal sealed record SqsOutboundMessage(
 205    string QueueUrl,
 206    string Body,
 207    string? CorrelationId,
 208    string? MessageGroupId,
 209    string? MessageDeduplicationId,
 210    IReadOnlyDictionary<string, string> MessageAttributes);
 211
 212internal sealed record SqsReceiveRequest(
 213    string QueueUrl,
 214    int MaxMessages,
 215    TimeSpan WaitTime,
 216    TimeSpan? VisibilityTimeout);
 217
 218internal sealed record SqsTransportDelivery(
 219    string QueueUrl,
 220    string Body,
 221    string MessageId,
 222    string ReceiptHandle,
 223    int ReceiveCount,
 224    IReadOnlyDictionary<string, string> MessageAttributes,
 225    Func<ValueTask> DeleteAsync,
 226    Func<TimeSpan, ValueTask> ChangeVisibilityAsync);
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Amazon.SQS.IAmazonSQS, bool)
GetQueueUrlAsync()
CreateQueueAsync()
GetQueueArnAsync()
SetQueueAttributesAsync(string, System.Collections.Generic.IReadOnlyDictionary<string, string>, System.Threading.CancellationToken)
SendMessageAsync()
ReceiveMessagesAsync()
CreateDelivery(string, Amazon.SQS.Model.Message)
DisposeAsync()