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

Information
Class: AsyncResponse.Transports.SQS.SqsTransportDelivery
Assembly: AsyncResponse.Transports.SQS
File(s): /_/src/Transports/AsyncResponse.Transports.SQS/SqsClientAdapters.cs
Line coverage
100%
Covered lines: 9
Uncovered lines: 0
Coverable lines: 9
Total lines: 250
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_QueueUrl()100%11100%
get_Body()100%11100%
get_MessageId()100%11100%
get_ReceiptHandle()100%11100%
get_ReceiveCount()100%11100%
get_MessageAttributes()100%11100%
get_DeleteAsync()100%11100%
get_ChangeVisibilityAsync()100%11100%

File(s)

/_/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        // Native per-message delay (standard queues; 0–900s). Longer waits arrive chunked: the
 81        // envelope's NotBeforeUtc re-publish chain in the worker-job executor takes the next hop.
 82        if (message.DelaySeconds is { } delaySeconds)
 83            request.DelaySeconds = delaySeconds;
 84
 85        foreach (var attribute in message.MessageAttributes)
 86        {
 87            (request.MessageAttributes ??= []).Add(attribute.Key, new MessageAttributeValue
 88            {
 89                DataType = "String",
 90                StringValue = attribute.Value
 91            });
 92        }
 93
 94        var response = await inner.SendMessageAsync(request, cancellationToken).ConfigureAwait(false);
 95        return response.MessageId;
 96    }
 97
 98    /// <summary>Long-polls the queue and wraps the received messages as transport deliveries.</summary>
 99    public async Task<IReadOnlyList<SqsTransportDelivery>> ReceiveMessagesAsync(
 100        SqsReceiveRequest request,
 101        CancellationToken cancellationToken = default)
 102    {
 103        var receive = new ReceiveMessageRequest
 104        {
 105            QueueUrl = request.QueueUrl,
 106            MaxNumberOfMessages = request.MaxMessages,
 107            WaitTimeSeconds = WholeSeconds(request.WaitTime),
 108            MessageSystemAttributeNames = [MessageSystemAttributeName.ApproximateReceiveCount],
 109            MessageAttributeNames = ["All"]
 110        };
 111        if (request.VisibilityTimeout is { } visibilityTimeout)
 112            receive.VisibilityTimeout = WholeSeconds(visibilityTimeout);
 113
 114        var response = await inner.ReceiveMessageAsync(receive, cancellationToken).ConfigureAwait(false);
 115        // AWS SDK v4 leaves collections null when the response carries no items.
 116        if (response.Messages is not { Count: > 0 } messages)
 117            return [];
 118
 119        var deliveries = new SqsTransportDelivery[messages.Count];
 120        for (var i = 0; i < messages.Count; i++)
 121            deliveries[i] = CreateDelivery(request.QueueUrl, messages[i]);
 122
 123        return deliveries;
 124    }
 125
 126    private SqsTransportDelivery CreateDelivery(string queueUrl, Message message)
 127    {
 128        var receiveCount = 1;
 129        if (message.Attributes is { } systemAttributes
 130            && systemAttributes.TryGetValue(MessageSystemAttributeName.ApproximateReceiveCount, out var rawReceiveCount)
 131            && int.TryParse(rawReceiveCount, out var parsedReceiveCount))
 132        {
 133            receiveCount = parsedReceiveCount;
 134        }
 135
 136        // Ordinal on purpose, unlike the sibling transports' OrdinalIgnoreCase inbound header maps:
 137        // SQS treats attribute names case-sensitively, so "CorrelationId" and "correlationId" can
 138        // coexist as two distinct real attributes on one message — a case-folding map would alias
 139        // them and let one silently shadow the other. The outbound publish path is Ordinal too.
 140        var messageAttributes = new Dictionary<string, string>(StringComparer.Ordinal);
 141        if (message.MessageAttributes is { } attributes)
 142        {
 143            foreach (var attribute in attributes)
 144            {
 145                if (attribute.Value?.StringValue is { } value)
 146                    messageAttributes[attribute.Key] = value;
 147            }
 148        }
 149
 150        var receiptHandle = message.ReceiptHandle;
 151        return new SqsTransportDelivery(
 152            queueUrl,
 153            message.Body ?? string.Empty,
 154            message.MessageId ?? string.Empty,
 155            receiptHandle,
 156            receiveCount,
 157            messageAttributes,
 158            () => new ValueTask(inner.DeleteMessageAsync(queueUrl, receiptHandle, CancellationToken.None)),
 159            // The caller chooses the token: settlement paths pass None (settlement ignores
 160            // cancellation), while the visibility-renewal heartbeat passes its shutdown-linked
 161            // token so a degraded endpoint cannot hold the stop hostage for the SDK retry budget.
 162            (delay, cancellationToken) => new ValueTask(inner.ChangeMessageVisibilityAsync(
 163                queueUrl,
 164                receiptHandle,
 165                WholeSeconds(delay),
 166                cancellationToken)));
 167    }
 168
 169    /// <summary>
 170    /// Converts a duration to the whole seconds SQS speaks, rounding UP so a positive value never
 171    /// becomes zero. Truncation is not a rounding nicety here: a 500 ms visibility timeout floored
 172    /// to 0 makes the message visible again immediately, and a second consumer picks it up while
 173    /// the first is still handling it — the exactly-one-handler guarantee the timeout exists for.
 174    /// A redelivery delay floored to 0 is a hot retry loop for the same reason. Zero itself is
 175    /// preserved, because "make it visible now" is a legitimate request. This matches the delayed
 176    /// publish path, which already rounds up.
 177    /// </summary>
 178    private static int WholeSeconds(TimeSpan value)
 179        => value <= TimeSpan.Zero ? 0 : (int)Math.Min(Math.Ceiling(value.TotalSeconds), int.MaxValue);
 180
 181    /// <summary>Releases resources held by this instance.</summary>
 182    public ValueTask DisposeAsync()
 183    {
 184        if (ownsClient)
 185            inner.Dispose();
 186        return ValueTask.CompletedTask;
 187    }
 188}
 189
 190internal static class SqsClientFactory
 191{
 192    /// <summary>Builds an SQS client from the transport options (endpoint, region, credentials).</summary>
 193    public static ISqsClient Create(SqsAsyncResponseOptions options)
 194    {
 195        var config = new AmazonSQSConfig();
 196        if (!string.IsNullOrWhiteSpace(options.ServiceUrl))
 197        {
 198            config.ServiceURL = options.ServiceUrl;
 199            // Custom endpoints (LocalStack, proxies) still need a signing region.
 200            config.AuthenticationRegion = options.Region ?? "us-east-1";
 201        }
 202        else if (!string.IsNullOrWhiteSpace(options.Region))
 203        {
 204            config.RegionEndpoint = RegionEndpoint.GetBySystemName(options.Region);
 205        }
 206
 207        var client = !string.IsNullOrWhiteSpace(options.AccessKey) && !string.IsNullOrWhiteSpace(options.SecretKey)
 208            ? new AmazonSQSClient(new BasicAWSCredentials(options.AccessKey, options.SecretKey), config)
 209            : new AmazonSQSClient(config);
 210        return new SqsClientAdapter(client, ownsClient: true);
 211    }
 212}
 213
 214internal static class SqsClientResolver
 215{
 216    /// <summary>Reuses an application-registered <see cref="IAmazonSQS"/> or builds one from the options.</summary>
 217    public static ISqsClient Create(IServiceProvider provider)
 218    {
 219        if (provider.GetService<IAmazonSQS>() is { } registeredClient)
 220            return new SqsClientAdapter(registeredClient, ownsClient: false);
 221
 222        var options = provider.GetRequiredService<Microsoft.Extensions.Options.IOptions<SqsAsyncResponseOptions>>().Valu
 223        return SqsClientFactory.Create(options);
 224    }
 225}
 226
 227internal sealed record SqsOutboundMessage(
 228    string QueueUrl,
 229    string Body,
 230    string? CorrelationId,
 231    string? MessageGroupId,
 232    string? MessageDeduplicationId,
 233    IReadOnlyDictionary<string, string> MessageAttributes,
 234    int? DelaySeconds = null);
 235
 236internal sealed record SqsReceiveRequest(
 237    string QueueUrl,
 238    int MaxMessages,
 239    TimeSpan WaitTime,
 240    TimeSpan? VisibilityTimeout);
 241
 527242internal sealed record SqsTransportDelivery(
 2243    string QueueUrl,
 457244    string Body,
 104245    string MessageId,
 2246    string ReceiptHandle,
 28247    int ReceiveCount,
 525248    IReadOnlyDictionary<string, string> MessageAttributes,
 475249    Func<ValueTask> DeleteAsync,
 599250    Func<TimeSpan, CancellationToken, ValueTask> ChangeVisibilityAsync);
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.String,System.Int32,System.Collections.Generic.IReadOnlyDictionary`2<System.String,System.String>,System.Func`1<System.Threading.Tasks.ValueTask>,System.Func`3<System.TimeSpan,System.Threading.CancellationToken,System.Threading.Tasks.ValueTask>)
get_QueueUrl()
get_Body()
get_MessageId()
get_ReceiptHandle()
get_ReceiveCount()
get_MessageAttributes()
get_DeleteAsync()
get_ChangeVisibilityAsync()