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

Information
Class: AsyncResponse.Transports.Redis.RedisTransportRetry
Assembly: AsyncResponse.Transports.Redis
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.Redis/RedisTransportClientAdapters.cs
Line coverage
100%
Covered lines: 4
Uncovered lines: 0
Coverable lines: 4
Total lines: 196
Line coverage: 100%
Branch coverage
100%
Covered branches: 8
Total branches: 8
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
ExecuteAsync<T>(...)100%22100%
IsTransient(...)100%66100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.Redis/RedisTransportClientAdapters.cs

#LineLine coverage
 1using StackExchange.Redis;
 2
 3namespace AsyncResponse.Transports.Redis;
 4
 5internal interface IRedisStreamDatabase
 6{
 7    Task<RedisValue> StreamAddAsync(
 8        RedisKey stream,
 9        NameValueEntry[] values,
 10        long? maxLength,
 11        bool useApproximateMaxLength,
 12        CancellationToken cancellationToken);
 13
 14    Task<bool> StreamCreateConsumerGroupAsync(
 15        RedisKey stream,
 16        RedisValue groupName,
 17        RedisValue position,
 18        bool createStream,
 19        CancellationToken cancellationToken);
 20
 21    Task<StreamEntry[]> StreamReadGroupAsync(
 22        RedisKey stream,
 23        RedisValue groupName,
 24        RedisValue consumerName,
 25        int count,
 26        CancellationToken cancellationToken);
 27
 28    Task<long> StreamAcknowledgeAsync(
 29        RedisKey stream,
 30        RedisValue groupName,
 31        RedisValue messageId,
 32        CancellationToken cancellationToken);
 33
 34    Task<StreamPendingMessageInfo[]> StreamPendingMessagesAsync(
 35        RedisKey stream,
 36        RedisValue groupName,
 37        int count,
 38        RedisValue consumerName,
 39        RedisValue? minId,
 40        RedisValue? maxId,
 41        long minIdleTimeInMilliseconds,
 42        CancellationToken cancellationToken);
 43
 44    Task<StreamEntry[]> StreamClaimAsync(
 45        RedisKey stream,
 46        RedisValue groupName,
 47        RedisValue consumerName,
 48        long minIdleTimeInMilliseconds,
 49        RedisValue[] messageIds,
 50        CancellationToken cancellationToken);
 51}
 52
 53internal sealed class RedisStreamDatabaseAdapter(IDatabase _database, TimeSpan _operationTimeout) : IRedisStreamDatabase
 54{
 55    /// <summary>Runs the StreamAddAsync operation.</summary>
 56    public Task<RedisValue> StreamAddAsync(
 57        RedisKey stream,
 58        NameValueEntry[] values,
 59        long? maxLength,
 60        bool useApproximateMaxLength,
 61        CancellationToken cancellationToken)
 62        // Call the classic overload (int? maxLength, no trim-mode parameter) so publishing emits plain
 63        // `XADD … MAXLEN ~ N` with no Redis 8 KEEPREF/DELREF/ACKED token. That keeps the transport
 64        // portable across Redis 8+, Valkey, and Dragonfly by construction — independent of whether the
 65        // StackExchange.Redis version would otherwise fold KEEPREF into the wire form. On Redis 8+ the
 66        // server default is KEEPREF, so an entry trimmed while still pending becomes a tombstone on claim
 67        // and the subscriber dead-letters it via DiscardUnprocessableAsync rather than wedging; the older
 68        // trim behavior on pre-8 servers is equivalent for that path.
 69        => WithCancellation(
 70            _database.StreamAddAsync(
 71                stream,
 72                values,
 73                messageId: (RedisValue?)null,
 74                maxLength: ToInt32MaxLength(maxLength),
 75                useApproximateMaxLength: useApproximateMaxLength,
 76                flags: CommandFlags.None),
 77            cancellationToken);
 78
 79    // Redis caps a stream's MAXLEN well below int.MaxValue in practice; clamp so the classic overload
 80    // (int? maxLength) is always reachable without overflow.
 81    private static int? ToInt32MaxLength(long? maxLength)
 82        => maxLength is null ? null : (int?)Math.Min(maxLength.Value, int.MaxValue);
 83
 84    /// <summary>Runs the StreamCreateConsumerGroupAsync operation.</summary>
 85    public Task<bool> StreamCreateConsumerGroupAsync(
 86        RedisKey stream,
 87        RedisValue groupName,
 88        RedisValue position,
 89        bool createStream,
 90        CancellationToken cancellationToken)
 91        => WithCancellation(
 92            _database.StreamCreateConsumerGroupAsync(stream, groupName, position, createStream),
 93            cancellationToken);
 94
 95    /// <summary>Runs the StreamReadGroupAsync operation.</summary>
 96    public Task<StreamEntry[]> StreamReadGroupAsync(
 97        RedisKey stream,
 98        RedisValue groupName,
 99        RedisValue consumerName,
 100        int count,
 101        CancellationToken cancellationToken)
 102        => WithCancellation(
 103            _database.StreamReadGroupAsync(
 104                stream,
 105                groupName,
 106                consumerName,
 107                position: StreamPosition.NewMessages,
 108                count: count,
 109                noAck: false),
 110            cancellationToken);
 111
 112    /// <summary>Runs the StreamAcknowledgeAsync operation.</summary>
 113    public Task<long> StreamAcknowledgeAsync(
 114        RedisKey stream,
 115        RedisValue groupName,
 116        RedisValue messageId,
 117        CancellationToken cancellationToken)
 118        => WithCancellation(
 119            _database.StreamAcknowledgeAsync(stream, groupName, messageId),
 120            cancellationToken);
 121
 122    /// <summary>Runs the StreamPendingMessagesAsync operation.</summary>
 123    public Task<StreamPendingMessageInfo[]> StreamPendingMessagesAsync(
 124        RedisKey stream,
 125        RedisValue groupName,
 126        int count,
 127        RedisValue consumerName,
 128        RedisValue? minId,
 129        RedisValue? maxId,
 130        long minIdleTimeInMilliseconds,
 131        CancellationToken cancellationToken)
 132        => WithCancellation(
 133            _database.StreamPendingMessagesAsync(
 134                stream,
 135                groupName,
 136                count,
 137                consumerName,
 138                minId,
 139                maxId,
 140                minIdleTimeInMilliseconds),
 141            cancellationToken);
 142
 143    /// <summary>Runs the StreamClaimAsync operation.</summary>
 144    public Task<StreamEntry[]> StreamClaimAsync(
 145        RedisKey stream,
 146        RedisValue groupName,
 147        RedisValue consumerName,
 148        long minIdleTimeInMilliseconds,
 149        RedisValue[] messageIds,
 150        CancellationToken cancellationToken)
 151        => WithCancellation(
 152            _database.StreamClaimAsync(
 153                stream,
 154                groupName,
 155                consumerName,
 156                minIdleTimeInMilliseconds,
 157                messageIds),
 158            cancellationToken);
 159
 160    private async Task<T> WithCancellation<T>(Task<T> command, CancellationToken cancellationToken)
 161    {
 162        // StackExchange.Redis enforces its own sync/async command timeouts; this adds an upper bound that
 163        // also honors the caller's token (e.g. host shutdown). On timeout the in-flight command is
 164        // abandoned best-effort — the multiplexer keeps running it — and surfaced as a TimeoutException so
 165        // the retry paths treat it as transient, while a genuine caller cancellation stays an
 166        // OperationCanceledException and is not retried.
 167        using var timeout = new CancellationTokenSource(_operationTimeout);
 168        using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeout.Token);
 169        try
 170        {
 171            return await command.WaitAsync(linked.Token).ConfigureAwait(false);
 172        }
 173        catch (OperationCanceledException) when (timeout.IsCancellationRequested && !cancellationToken.IsCancellationReq
 174        {
 175            throw new TimeoutException($"The Redis command did not complete within {_operationTimeout}.");
 176        }
 177    }
 178}
 179
 180internal static class RedisTransportRetry
 181{
 182    /// <summary>Runs this background operation until cancellation is requested.</summary>
 183    public static Task<T> ExecuteAsync<T>(
 184        Func<CancellationToken, Task<T>> action,
 185        int maxAttempts,
 186        TimeSpan baseDelay,
 187        TimeSpan maxDelay,
 188        CancellationToken cancellationToken)
 3189        => AsyncResponseRetry.ExecuteAsync(action, IsTransient, maxAttempts, baseDelay, maxDelay, cancellationToken);
 190
 191    /// <summary>Runs the IsTransient operation.</summary>
 192    public static bool IsTransient(Exception exception)
 3193        => exception is RedisConnectionException
 3194            or RedisTimeoutException
 3195            or TimeoutException;
 196}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

ExecuteAsync<T>(System.Func<System.Threading.CancellationToken, System.Threading.Tasks.Task<T>>, int, System.TimeSpan, System.TimeSpan, System.Threading.CancellationToken)
IsTransient(System.Exception)