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

Information
Class: AsyncResponse.Transports.Redis.RedisStreamDatabaseAdapter
Assembly: AsyncResponse.Transports.Redis
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.Redis/RedisTransportClientAdapters.cs
Line coverage
100%
Covered lines: 50
Uncovered lines: 0
Coverable lines: 50
Total lines: 196
Line coverage: 100%
Branch coverage
100%
Covered branches: 4
Total branches: 4
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%
StreamAddAsync(...)100%11100%
ToInt32MaxLength(...)100%22100%
StreamCreateConsumerGroupAsync(...)100%11100%
StreamReadGroupAsync(...)100%11100%
StreamAcknowledgeAsync(...)100%11100%
StreamPendingMessagesAsync(...)100%11100%
StreamClaimAsync(...)100%11100%
WithCancellation()100%22100%

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
 353internal 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.
 369        => WithCancellation(
 370            _database.StreamAddAsync(
 371                stream,
 372                values,
 373                messageId: (RedisValue?)null,
 374                maxLength: ToInt32MaxLength(maxLength),
 375                useApproximateMaxLength: useApproximateMaxLength,
 376                flags: CommandFlags.None),
 377            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)
 382        => 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)
 391        => WithCancellation(
 392            _database.StreamCreateConsumerGroupAsync(stream, groupName, position, createStream),
 393            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)
 3102        => WithCancellation(
 3103            _database.StreamReadGroupAsync(
 3104                stream,
 3105                groupName,
 3106                consumerName,
 3107                position: StreamPosition.NewMessages,
 3108                count: count,
 3109                noAck: false),
 3110            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)
 3118        => WithCancellation(
 3119            _database.StreamAcknowledgeAsync(stream, groupName, messageId),
 3120            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)
 3132        => WithCancellation(
 3133            _database.StreamPendingMessagesAsync(
 3134                stream,
 3135                groupName,
 3136                count,
 3137                consumerName,
 3138                minId,
 3139                maxId,
 3140                minIdleTimeInMilliseconds),
 3141            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)
 3151        => WithCancellation(
 3152            _database.StreamClaimAsync(
 3153                stream,
 3154                groupName,
 3155                consumerName,
 3156                minIdleTimeInMilliseconds,
 3157                messageIds),
 3158            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.
 3167        using var timeout = new CancellationTokenSource(_operationTimeout);
 3168        using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeout.Token);
 169        try
 170        {
 3171            return await command.WaitAsync(linked.Token).ConfigureAwait(false);
 172        }
 3173        catch (OperationCanceledException) when (timeout.IsCancellationRequested && !cancellationToken.IsCancellationReq
 174        {
 3175            throw new TimeoutException($"The Redis command did not complete within {_operationTimeout}.");
 176        }
 3177    }
 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)
 189        => AsyncResponseRetry.ExecuteAsync(action, IsTransient, maxAttempts, baseDelay, maxDelay, cancellationToken);
 190
 191    /// <summary>Runs the IsTransient operation.</summary>
 192    public static bool IsTransient(Exception exception)
 193        => exception is RedisConnectionException
 194            or RedisTimeoutException
 195            or TimeoutException;
 196}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(StackExchange.Redis.IDatabase, System.TimeSpan)
StreamAddAsync(StackExchange.Redis.RedisKey, StackExchange.Redis.NameValueEntry[], System.Nullable<long>, bool, System.Threading.CancellationToken)
ToInt32MaxLength(System.Nullable<long>)
StreamCreateConsumerGroupAsync(StackExchange.Redis.RedisKey, StackExchange.Redis.RedisValue, StackExchange.Redis.RedisValue, bool, System.Threading.CancellationToken)
StreamReadGroupAsync(StackExchange.Redis.RedisKey, StackExchange.Redis.RedisValue, StackExchange.Redis.RedisValue, int, System.Threading.CancellationToken)
StreamAcknowledgeAsync(StackExchange.Redis.RedisKey, StackExchange.Redis.RedisValue, StackExchange.Redis.RedisValue, System.Threading.CancellationToken)
StreamPendingMessagesAsync(StackExchange.Redis.RedisKey, StackExchange.Redis.RedisValue, int, StackExchange.Redis.RedisValue, System.Nullable<StackExchange.Redis.RedisValue>, System.Nullable<StackExchange.Redis.RedisValue>, long, System.Threading.CancellationToken)
StreamClaimAsync(StackExchange.Redis.RedisKey, StackExchange.Redis.RedisValue, StackExchange.Redis.RedisValue, long, StackExchange.Redis.RedisValue[], System.Threading.CancellationToken)
WithCancellation()