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

Information
Class: AsyncResponse.Transports.Redis.RedisWorkerTransport
Assembly: AsyncResponse.Transports.Redis
File(s): /_/src/Transports/AsyncResponse.Transports.Redis/RedisWorkerTransport.cs
Line coverage
100%
Covered lines: 68
Uncovered lines: 0
Coverable lines: 68
Total lines: 134
Line coverage: 100%
Branch coverage
100%
Covered branches: 12
Total branches: 12
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%
RedisTransportOptionsValidatorWithValue(...)100%11100%
ValidatePublishOptions(...)100%11100%
PublishAsync()100%1010100%
CreateMessageFields(...)100%22100%

File(s)

/_/src/Transports/AsyncResponse.Transports.Redis/RedisWorkerTransport.cs

#LineLine coverage
 1using Microsoft.Extensions.Options;
 2using StackExchange.Redis;
 3using System.Diagnostics;
 4using System.Text.Json;
 5
 6namespace AsyncResponse.Transports.Redis;
 7
 8/// <summary>
 9/// Publishes <see cref="WorkerJobEnvelope"/> messages to a Redis stream.
 10/// </summary>
 11/// <remarks>
 12/// Redis Streams provide durable queueing and consumer-group acknowledgement. Publishing uses XADD
 13/// with optional approximate trimming, and transient Redis failures are retried with bounded
 14/// exponential backoff before the exception is returned to the caller.
 15/// </remarks>
 16public sealed class RedisWorkerTransport : IWorkerTransport
 17{
 18    private readonly RedisAsyncResponseTransportOptions _options;
 19    private readonly IRedisStreamDatabase _database;
 20    private readonly RedisTransportKeySchema _keys;
 21    private readonly TimeSpan _publishDedupTtl;
 22
 23    /// <summary>Runs the RedisWorkerTransport operation.</summary>
 24    public RedisWorkerTransport(
 25        IOptions<RedisAsyncResponseTransportOptions> options,
 26        IConnectionMultiplexer multiplexer)
 19527        : this(
 19528            options,
 19529            new RedisStreamDatabaseAdapter(
 19530                multiplexer.GetDatabase(),
 19531                RedisTransportOptionsValidatorWithValue(options).OperationTimeout))
 32    {
 19533    }
 34
 21335    internal RedisWorkerTransport(
 21336        IOptions<RedisAsyncResponseTransportOptions> options,
 21337        IRedisStreamDatabase database)
 38    {
 21339        _options = options.Value;
 21340        RedisTransportOptionsValidator.ValidateCommon(_options);
 20941        ValidatePublishOptions(_options);
 20942        _database = database;
 20943        _keys = new RedisTransportKeySchema(_options);
 44
 45        // The dedup marker must outlive the whole retry window — attempts x (operation timeout +
 46        // max backoff) — so a late retry still finds it; 2x margin, then it is garbage and
 47        // expires. ~66s at the defaults.
 20948        var perAttempt = _options.OperationTimeout + _options.PublishRetryMaxDelay;
 20949        _publishDedupTtl = TimeSpan.FromTicks(perAttempt.Ticks * Math.Max(1, _options.PublishMaxAttempts) * 2);
 20950    }
 51
 52    private static RedisAsyncResponseTransportOptions RedisTransportOptionsValidatorWithValue(
 53        IOptions<RedisAsyncResponseTransportOptions> options)
 54    {
 19555        RedisTransportOptionsValidator.ValidateCommon(options.Value);
 19556        ValidatePublishOptions(options.Value);
 19557        return options.Value;
 58    }
 59
 60    private static void ValidatePublishOptions(RedisAsyncResponseTransportOptions options)
 61    {
 40462        _ = new RedisTransportKeySchema(options).WorkerStream;
 40463        _ = RedisTransportOptionsValidator.Required(options.PayloadField, nameof(options.PayloadField));
 40464    }
 65
 66    /// <summary>Publishes the supplied message.</summary>
 67    public async Task PublishAsync(WorkerJobEnvelope job, CancellationToken cancellationToken = default)
 68    {
 41969        ArgumentNullException.ThrowIfNull(job);
 70
 41771        using var activity = AsyncResponseDiagnostics.StartActivity(
 41772            "asyncresponse.worker.publish",
 41773            ActivityKind.Producer,
 41774            job.CorrelationId);
 41775        activity?.SetTag("asyncresponse.transport", "redis");
 41776        activity?.SetTag("messaging.system", "redis");
 41777        activity?.SetTag("messaging.destination.name", _keys.WorkerStream.ToString());
 41778        AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget);
 41779        AsyncResponseDiagnostics.SetWorker(activity, job.Call);
 80
 81        try
 82        {
 41783            var fields = CreateMessageFields(
 41784                AsyncResponseJson.Serialize(job),
 41785                job.CorrelationId,
 41786                _options);
 87            // Identity pinned OUTSIDE the retry loop (MongoDB/ASB/SQS parity): a retry after an
 88            // ambiguous timeout — the adapter abandons the in-flight XADD best-effort while the
 89            // multiplexer keeps running it — must not append the same worker job twice. The
 90            // server-side marker makes the retried append a no-op instead.
 41791            var publishId = Guid.NewGuid().ToString("N");
 41792            var dedupKey = _keys.WorkerPublishDedupKey(publishId);
 41793            var messageId = await RedisTransportRetry.ExecuteAsync(
 42394                token => _database.StreamAddOnceAsync(
 42395                    _keys.WorkerStream,
 42396                    dedupKey,
 42397                    _publishDedupTtl,
 42398                    fields,
 42399                    _options.StreamMaxLength,
 423100                    _options.UseApproximateStreamTrimming,
 423101                    token),
 417102                _options.PublishMaxAttempts,
 417103                _options.PublishRetryBaseDelay,
 417104                _options.PublishRetryMaxDelay,
 417105                cancellationToken).ConfigureAwait(false);
 106
 107            // Null: a previous attempt's append already committed — the job is on the stream once.
 415108            if (!messageId.IsNull)
 413109                activity?.SetTag("messaging.message.id", messageId.ToString());
 415110        }
 2111        catch (Exception ex)
 112        {
 2113            AsyncResponseDiagnostics.SetError(activity, ex);
 2114            throw;
 115        }
 415116    }
 117
 118    internal static NameValueEntry[] CreateMessageFields(
 119        string payloadJson,
 120        string? correlationId,
 121        RedisAsyncResponseTransportOptions options)
 122    {
 417123        var payloadField = RedisTransportOptionsValidator.Required(options.PayloadField, nameof(options.PayloadField));
 417124        var correlationField = RedisTransportOptionsValidator.Required(options.CorrelationIdField, nameof(options.Correl
 125
 417126        return string.IsNullOrWhiteSpace(correlationId)
 417127            ? [new NameValueEntry(payloadField, payloadJson)]
 417128            :
 417129            [
 417130                new NameValueEntry(payloadField, payloadJson),
 417131                new NameValueEntry(correlationField, correlationId)
 417132            ];
 133    }
 134}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions>,StackExchange.Redis.IConnectionMultiplexer)
.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions>,AsyncResponse.Transports.Redis.IRedisStreamDatabase)
RedisTransportOptionsValidatorWithValue(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions>)
ValidatePublishOptions(AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions)
PublishAsync()
CreateMessageFields(System.String,System.String,AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions)