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

Information
Class: AsyncResponse.Transports.SQS.SqsWorkerTransport
Assembly: AsyncResponse.Transports.SQS
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SQS/SqsWorkerTransport.cs
Line coverage
100%
Covered lines: 80
Uncovered lines: 0
Coverable lines: 80
Total lines: 168
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%
.ctor(...)100%11100%
.ctor(...)100%11100%
GetQueueUrlAsync()100%66100%
PublishAsync()100%1818100%
SendWithRetryAsync()100%44100%
IsTransient(...)100%88100%
DisposeAsync()100%44100%

File(s)

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

#LineLine coverage
 1using Amazon.SQS;
 2using Microsoft.Extensions.Options;
 3using System.Diagnostics;
 4using System.Net;
 5using System.Text.Json;
 6
 7namespace AsyncResponse.Transports.SQS;
 8
 9/// <summary>Publishes <see cref="WorkerJobEnvelope"/> messages to an AWS SQS queue.</summary>
 10/// <remarks>
 11/// The worker queue URL is resolved lazily (a queue configured by name goes through
 12/// <c>GetQueueUrl</c> once) and cached for the lifetime of the transport; a transient resolution
 13/// failure on the first publish is not cached, so the next publish retries. When the worker queue
 14/// is a FIFO queue (name or URL ending in <c>.fifo</c>), the correlation id becomes the
 15/// <c>MessageGroupId</c> so one flow's jobs stay ordered, and every message carries a unique
 16/// <c>MessageDeduplicationId</c> so distinct jobs of the same flow are never deduplicated away.
 17/// </remarks>
 18public sealed class SqsWorkerTransport : IWorkerTransport, IAsyncDisposable
 19{
 20    private readonly SqsAsyncResponseOptions _options;
 21    private readonly ISqsClient _client;
 22    private readonly bool _disposeClient;
 23    private readonly bool _isFifoQueue;
 324    private readonly SemaphoreSlim _queueUrlGate = new(1, 1);
 25    private string? _queueUrl;
 26    private int _disposeGate;
 27    private bool _disposed;
 28
 29    /// <summary>Creates a worker transport backed by a client built from the configured options.</summary>
 30    public SqsWorkerTransport(IOptions<SqsAsyncResponseOptions> options)
 231        : this(options, SqsClientFactory.Create(options.Value), disposeClient: true)
 32    {
 233    }
 34
 35    internal SqsWorkerTransport(
 36        IOptions<SqsAsyncResponseOptions> options,
 37        ISqsClient client)
 338        : this(options, client, disposeClient: false)
 39    {
 340    }
 41
 342    private SqsWorkerTransport(
 343        IOptions<SqsAsyncResponseOptions> options,
 344        ISqsClient client,
 345        bool disposeClient)
 46    {
 347        _options = options.Value;
 348        SqsOptionsValidator.ValidateCommon(_options);
 349        _client = client;
 350        _disposeClient = disposeClient;
 351        _isFifoQueue = SqsQueueAddress.IsFifo(_options.WorkerQueue);
 352    }
 53
 54    private async Task<string> GetQueueUrlAsync(CancellationToken cancellationToken)
 55    {
 356        var queueUrl = Volatile.Read(ref _queueUrl);
 357        if (queueUrl is not null)
 358            return queueUrl;
 59
 360        await _queueUrlGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 61        try
 62        {
 363            ObjectDisposedException.ThrowIf(_disposed, this);
 364            if (_queueUrl is not null)
 265                return _queueUrl;
 66
 67            // Assign only after the await succeeds, so a faulted resolution is not cached and the
 68            // next publish retries instead of reusing a permanently failed lookup.
 369            var resolved = SqsQueueAddress.IsUrl(_options.WorkerQueue)
 370                ? _options.WorkerQueue
 371                : await _client.GetQueueUrlAsync(_options.WorkerQueue, cancellationToken).ConfigureAwait(false);
 372            _queueUrl = resolved;
 373            return resolved;
 74        }
 75        finally
 76        {
 377            _queueUrlGate.Release();
 78        }
 379    }
 80
 81    /// <summary>Publishes the supplied worker job.</summary>
 82    public async Task PublishAsync(WorkerJobEnvelope job, CancellationToken cancellationToken = default)
 83    {
 384        ArgumentNullException.ThrowIfNull(job);
 85
 386        using var activity = AsyncResponseDiagnostics.StartActivity(
 387            "asyncresponse.worker.publish",
 388            ActivityKind.Producer,
 389            job.CorrelationId);
 390        activity?.SetTag("asyncresponse.transport", "aws_sqs");
 391        activity?.SetTag("messaging.system", "aws_sqs");
 392        activity?.SetTag("messaging.destination.name", _options.WorkerQueue);
 393        AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget);
 394        AsyncResponseDiagnostics.SetWorker(activity, job.Call);
 95
 96        try
 97        {
 398            var messageAttributes = new Dictionary<string, string>(StringComparer.Ordinal);
 399            if (!string.IsNullOrWhiteSpace(job.CorrelationId))
 3100                messageAttributes[_options.CorrelationIdAttribute] = job.CorrelationId;
 101
 3102            var queueUrl = await GetQueueUrlAsync(cancellationToken).ConfigureAwait(false);
 3103            var message = new SqsOutboundMessage(
 3104                queueUrl,
 3105                AsyncResponseJson.Serialize(job),
 3106                string.IsNullOrWhiteSpace(job.CorrelationId) ? null : job.CorrelationId,
 3107                MessageGroupId: _isFifoQueue
 3108                    ? (string.IsNullOrWhiteSpace(job.CorrelationId) ? _options.FifoMessageGroupIdFallback : job.Correlat
 3109                    : null,
 3110                MessageDeduplicationId: _isFifoQueue ? Guid.NewGuid().ToString("N") : null,
 3111                messageAttributes);
 112
 3113            var messageId = await SendWithRetryAsync(message, cancellationToken).ConfigureAwait(false);
 3114            activity?.SetTag("messaging.message.id", messageId);
 3115        }
 2116        catch (Exception ex)
 117        {
 2118            AsyncResponseDiagnostics.SetError(activity, ex);
 2119            throw;
 120        }
 3121    }
 122
 123    private async Task<string> SendWithRetryAsync(
 124        SqsOutboundMessage message,
 125        CancellationToken cancellationToken)
 126    {
 3127        for (var attempt = 1; ; attempt++)
 128        {
 129            try
 130            {
 3131                return await _client.SendMessageAsync(message, cancellationToken).ConfigureAwait(false);
 132            }
 2133            catch (Exception ex) when (IsTransient(ex) && attempt < _options.PublishMaxAttempts && !cancellationToken.Is
 134            {
 2135                var delay = AsyncResponseRetry.Backoff(attempt, _options.PublishRetryBaseDelay, _options.PublishRetryMax
 2136                await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
 137            }
 138        }
 3139    }
 140
 141    /// <summary>Classifies AWS SQS send failures worth retrying at the transport level.</summary>
 142    internal static bool IsTransient(Exception exception)
 2143        => exception is AmazonSQSException sqsException
 2144            && (sqsException.Retryable is not null
 2145                || sqsException.StatusCode >= HttpStatusCode.InternalServerError
 2146                || string.Equals(sqsException.ErrorCode, "RequestThrottled", StringComparison.Ordinal)
 2147                || string.Equals(sqsException.ErrorCode, "ThrottlingException", StringComparison.Ordinal));
 148
 149    /// <summary>Releases resources held by this instance.</summary>
 150    public async ValueTask DisposeAsync()
 151    {
 3152        if (Interlocked.Exchange(ref _disposeGate, 1) != 0)
 3153            return;
 154
 3155        await _queueUrlGate.WaitAsync().ConfigureAwait(false);
 156        try
 157        {
 3158            _disposed = true;
 3159            if (_disposeClient)
 2160                await _client.DisposeAsync().ConfigureAwait(false);
 3161        }
 162        finally
 163        {
 3164            _queueUrlGate.Release();
 3165            _queueUrlGate.Dispose();
 166        }
 3167    }
 168}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.SQS.SqsAsyncResponseOptions>, AsyncResponse.Transports.SQS.ISqsClient, bool)
.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.SQS.SqsAsyncResponseOptions>)
.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.SQS.SqsAsyncResponseOptions>, AsyncResponse.Transports.SQS.ISqsClient)
GetQueueUrlAsync()
PublishAsync()
SendWithRetryAsync()
IsTransient(System.Exception)
DisposeAsync()