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

Information
Class: AsyncResponse.Transports.SQS.SqsOutboundMessage
Assembly: AsyncResponse.Transports.SQS
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SQS/SqsClientAdapters.cs
Line coverage
100%
Covered lines: 7
Uncovered lines: 0
Coverable lines: 7
Total lines: 226
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.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
 19internal sealed class SqsClientAdapter(
 20    IAmazonSQS inner,
 21    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    {
 26        var response = await inner.GetQueueUrlAsync(queueName, cancellationToken).ConfigureAwait(false);
 27        return response.QueueUrl;
 28    }
 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    {
 36        var request = new CreateQueueRequest { QueueName = queueName };
 37        foreach (var attribute in attributes)
 38            (request.Attributes ??= []).Add(attribute.Key, attribute.Value);
 39
 40        var response = await inner.CreateQueueAsync(request, cancellationToken).ConfigureAwait(false);
 41        return response.QueueUrl;
 42    }
 43
 44    /// <summary>Reads the queue's ARN attribute.</summary>
 45    public async Task<string> GetQueueArnAsync(string queueUrl, CancellationToken cancellationToken = default)
 46    {
 47        var response = await inner.GetQueueAttributesAsync(
 48            new GetQueueAttributesRequest
 49            {
 50                QueueUrl = queueUrl,
 51                AttributeNames = [QueueAttributeName.QueueArn]
 52            },
 53            cancellationToken).ConfigureAwait(false);
 54        return response.QueueARN;
 55    }
 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)
 62        => inner.SetQueueAttributesAsync(
 63            new SetQueueAttributesRequest
 64            {
 65                QueueUrl = queueUrl,
 66                Attributes = attributes.ToDictionary(pair => pair.Key, pair => pair.Value, StringComparer.Ordinal)
 67            },
 68            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    {
 73        var request = new SendMessageRequest
 74        {
 75            QueueUrl = message.QueueUrl,
 76            MessageBody = message.Body,
 77            MessageGroupId = message.MessageGroupId,
 78            MessageDeduplicationId = message.MessageDeduplicationId
 79        };
 80
 81        foreach (var attribute in message.MessageAttributes)
 82        {
 83            (request.MessageAttributes ??= []).Add(attribute.Key, new MessageAttributeValue
 84            {
 85                DataType = "String",
 86                StringValue = attribute.Value
 87            });
 88        }
 89
 90        var response = await inner.SendMessageAsync(request, cancellationToken).ConfigureAwait(false);
 91        return response.MessageId;
 92    }
 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    {
 99        var receive = new ReceiveMessageRequest
 100        {
 101            QueueUrl = request.QueueUrl,
 102            MaxNumberOfMessages = request.MaxMessages,
 103            WaitTimeSeconds = (int)request.WaitTime.TotalSeconds,
 104            MessageSystemAttributeNames = [MessageSystemAttributeName.ApproximateReceiveCount],
 105            MessageAttributeNames = ["All"]
 106        };
 107        if (request.VisibilityTimeout is { } visibilityTimeout)
 108            receive.VisibilityTimeout = (int)visibilityTimeout.TotalSeconds;
 109
 110        var response = await inner.ReceiveMessageAsync(receive, cancellationToken).ConfigureAwait(false);
 111        // AWS SDK v4 leaves collections null when the response carries no items.
 112        if (response.Messages is not { Count: > 0 } messages)
 113            return [];
 114
 115        var deliveries = new SqsTransportDelivery[messages.Count];
 116        for (var i = 0; i < messages.Count; i++)
 117            deliveries[i] = CreateDelivery(request.QueueUrl, messages[i]);
 118
 119        return deliveries;
 120    }
 121
 122    private SqsTransportDelivery CreateDelivery(string queueUrl, Message message)
 123    {
 124        var receiveCount = 1;
 125        if (message.Attributes is { } systemAttributes
 126            && systemAttributes.TryGetValue(MessageSystemAttributeName.ApproximateReceiveCount, out var rawReceiveCount)
 127            && int.TryParse(rawReceiveCount, out var parsedReceiveCount))
 128        {
 129            receiveCount = parsedReceiveCount;
 130        }
 131
 132        var messageAttributes = new Dictionary<string, string>(StringComparer.Ordinal);
 133        if (message.MessageAttributes is { } attributes)
 134        {
 135            foreach (var attribute in attributes)
 136            {
 137                if (attribute.Value?.StringValue is { } value)
 138                    messageAttributes[attribute.Key] = value;
 139            }
 140        }
 141
 142        var receiptHandle = message.ReceiptHandle;
 143        return new SqsTransportDelivery(
 144            queueUrl,
 145            message.Body ?? string.Empty,
 146            message.MessageId ?? string.Empty,
 147            receiptHandle,
 148            receiveCount,
 149            messageAttributes,
 150            () => new ValueTask(inner.DeleteMessageAsync(queueUrl, receiptHandle, CancellationToken.None)),
 151            delay => new ValueTask(inner.ChangeMessageVisibilityAsync(
 152                queueUrl,
 153                receiptHandle,
 154                (int)delay.TotalSeconds,
 155                CancellationToken.None)));
 156    }
 157
 158    /// <summary>Releases resources held by this instance.</summary>
 159    public ValueTask DisposeAsync()
 160    {
 161        if (ownsClient)
 162            inner.Dispose();
 163        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
 3204internal sealed record SqsOutboundMessage(
 3205    string QueueUrl,
 3206    string Body,
 3207    string? CorrelationId,
 3208    string? MessageGroupId,
 3209    string? MessageDeduplicationId,
 3210    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(string, string, string, string, string, System.Collections.Generic.IReadOnlyDictionary<string, string>)