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

Information
Class: AsyncResponse.Transports.Redis.RedisStreamDatabaseAdapter
Assembly: AsyncResponse.Transports.Redis
File(s): /_/src/Transports/AsyncResponse.Transports.Redis/RedisTransportClientAdapters.cs
Line coverage
100%
Covered lines: 70
Uncovered lines: 0
Coverable lines: 70
Total lines: 298
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
.ctor(...)100%11100%
StreamAddAsync(...)100%11100%
ToInt32MaxLength(...)100%22100%
StreamAddOnceAsync()100%66100%
StreamCreateConsumerGroupAsync(...)100%11100%
StreamReadGroupAsync(...)100%11100%
StreamAcknowledgeAsync(...)100%11100%
StreamPendingMessagesAsync(...)100%11100%
StreamClaimAsync(...)100%11100%
StreamClaimIdsOnlyAsync(...)100%11100%
WithCancellation()100%11100%

File(s)

/_/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    /// <summary>
 53    /// XCLAIM JUSTID: transfers ownership of <paramref name="messageIds"/> to
 54    /// <paramref name="consumerName"/> and resets their idle time WITHOUT bumping the PEL
 55    /// delivery count, returning the ids actually claimed (already-ACKed ids are absent). Used
 56    /// as the in-flight batch heartbeat. Carries a claim-nothing default implementation so
 57    /// out-of-package fakes that never run the subscriber read loop keep compiling.
 58    /// </summary>
 59    Task<RedisValue[]> StreamClaimIdsOnlyAsync(
 60        RedisKey stream,
 61        RedisValue groupName,
 62        RedisValue consumerName,
 63        long minIdleTimeInMilliseconds,
 64        RedisValue[] messageIds,
 65        CancellationToken cancellationToken)
 66        => Task.FromResult(Array.Empty<RedisValue>());
 67
 68    /// <summary>
 69    /// Idempotent XADD: appends <paramref name="values"/> only when <paramref name="dedupKey"/>
 70    /// is not yet claimed (the marker and the append commit atomically, marker expiring after
 71    /// <paramref name="dedupTtl"/>), returning <see cref="RedisValue.Null"/> when a previous
 72    /// attempt's append already committed. Publish retries ride on this: XADD has no natural
 73    /// identity (the entry id is server-generated), so a retry after an ambiguous timeout — the
 74    /// adapter abandons the in-flight command best-effort while the multiplexer keeps running
 75    /// it — appended the same worker job twice. Carries a non-idempotent pass-through default so
 76    /// out-of-package fakes keep compiling.
 77    /// </summary>
 78    Task<RedisValue> StreamAddOnceAsync(
 79        RedisKey stream,
 80        RedisKey dedupKey,
 81        TimeSpan dedupTtl,
 82        NameValueEntry[] values,
 83        long? maxLength,
 84        bool useApproximateMaxLength,
 85        CancellationToken cancellationToken)
 86        => StreamAddAsync(stream, values, maxLength, useApproximateMaxLength, cancellationToken);
 87}
 88
 59689internal sealed class RedisStreamDatabaseAdapter(IDatabase _database, TimeSpan _operationTimeout) : IRedisStreamDatabase
 90{
 91    // Redis MULTI/EXEC does not roll back a successful SET when XADD fails. Record the
 92    // success marker only AFTER XADD succeeds, in the same server-side operation.
 93    internal const string AppendOnceScript = """
 94        local previous = redis.call('GET', KEYS[2])
 95        if previous then
 96            if previous == '' then
 97                return redis.error_reply('Invalid worker publish success marker')
 98            end
 99            return false
 100        end
 101        local command = {KEYS[1]}
 102        if ARGV[2] ~= '' then
 103            table.insert(command, 'MAXLEN')
 104            table.insert(command, ARGV[3])
 105            table.insert(command, ARGV[2])
 106        end
 107        table.insert(command, '*')
 108        for i = 4, #ARGV do table.insert(command, ARGV[i]) end
 109        local id = redis.call('XADD', unpack(command))
 110        redis.call('SET', KEYS[2], id, 'PX', ARGV[1])
 111        return id
 112        """;
 113
 114    /// <summary>Runs the StreamAddAsync operation.</summary>
 115    public Task<RedisValue> StreamAddAsync(
 116        RedisKey stream,
 117        NameValueEntry[] values,
 118        long? maxLength,
 119        bool useApproximateMaxLength,
 120        CancellationToken cancellationToken)
 121        // Call the classic overload (int? maxLength, no trim-mode parameter) so publishing emits plain
 122        // `XADD … MAXLEN ~ N` with no Redis 8 KEEPREF/DELREF/ACKED token. That keeps the transport
 123        // portable across Redis 8+, Valkey, and Dragonfly by construction — independent of whether the
 124        // StackExchange.Redis version would otherwise fold KEEPREF into the wire form. On Redis 8+ the
 125        // server default is KEEPREF, so an entry trimmed while still pending becomes a tombstone on claim
 126        // and the subscriber dead-letters it via DiscardUnprocessableAsync rather than wedging; the older
 127        // trim behavior on pre-8 servers is equivalent for that path.
 7128        => WithCancellation(
 7129            _database.StreamAddAsync(
 7130                stream,
 7131                values,
 7132                messageId: (RedisValue?)null,
 7133                maxLength: ToInt32MaxLength(maxLength),
 7134                useApproximateMaxLength: useApproximateMaxLength,
 7135                flags: CommandFlags.None),
 7136            cancellationToken);
 137
 138    // Redis caps a stream's MAXLEN well below int.MaxValue in practice; clamp so the classic overload
 139    // (int? maxLength) is always reachable without overflow.
 140    private static int? ToInt32MaxLength(long? maxLength)
 419141        => maxLength is null ? null : (int?)Math.Min(maxLength.Value, int.MaxValue);
 142
 143    /// <summary>Runs the StreamAddOnceAsync operation.</summary>
 144    public async Task<RedisValue> StreamAddOnceAsync(
 145        RedisKey stream,
 146        RedisKey dedupKey,
 147        TimeSpan dedupTtl,
 148        NameValueEntry[] values,
 149        long? maxLength,
 150        bool useApproximateMaxLength,
 151        CancellationToken cancellationToken)
 152    {
 412153        ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(dedupTtl, TimeSpan.Zero);
 412154        var args = new RedisValue[3 + values.Length * 2];
 412155        args[0] = checked((long)Math.Ceiling(dedupTtl.TotalMilliseconds));
 412156        args[1] = ToInt32MaxLength(maxLength) is { } cap ? cap : RedisValue.EmptyString;
 412157        args[2] = useApproximateMaxLength ? "~" : "=";
 1836158        for (var i = 0; i < values.Length; i++)
 159        {
 506160            args[3 + i * 2] = values[i].Name;
 506161            args[4 + i * 2] = values[i].Value;
 162        }
 163
 412164        return (RedisValue)await WithCancellation(
 412165            _database.ScriptEvaluateAsync(AppendOnceScript, [stream, dedupKey], args),
 412166            cancellationToken).ConfigureAwait(false);
 410167    }
 168
 169    /// <summary>Runs the StreamCreateConsumerGroupAsync operation.</summary>
 170    public Task<bool> StreamCreateConsumerGroupAsync(
 171        RedisKey stream,
 172        RedisValue groupName,
 173        RedisValue position,
 174        bool createStream,
 175        CancellationToken cancellationToken)
 384176        => WithCancellation(
 384177            _database.StreamCreateConsumerGroupAsync(stream, groupName, position, createStream),
 384178            cancellationToken);
 179
 180    /// <summary>Runs the StreamReadGroupAsync operation.</summary>
 181    public Task<StreamEntry[]> StreamReadGroupAsync(
 182        RedisKey stream,
 183        RedisValue groupName,
 184        RedisValue consumerName,
 185        int count,
 186        CancellationToken cancellationToken)
 3792187        => WithCancellation(
 3792188            _database.StreamReadGroupAsync(
 3792189                stream,
 3792190                groupName,
 3792191                consumerName,
 3792192                position: StreamPosition.NewMessages,
 3792193                count: count,
 3792194                noAck: false),
 3792195            cancellationToken);
 196
 197    /// <summary>Runs the StreamAcknowledgeAsync operation.</summary>
 198    public Task<long> StreamAcknowledgeAsync(
 199        RedisKey stream,
 200        RedisValue groupName,
 201        RedisValue messageId,
 202        CancellationToken cancellationToken)
 409203        => WithCancellation(
 409204            _database.StreamAcknowledgeAsync(stream, groupName, messageId),
 409205            cancellationToken);
 206
 207    /// <summary>Runs the StreamPendingMessagesAsync operation.</summary>
 208    public Task<StreamPendingMessageInfo[]> StreamPendingMessagesAsync(
 209        RedisKey stream,
 210        RedisValue groupName,
 211        int count,
 212        RedisValue consumerName,
 213        RedisValue? minId,
 214        RedisValue? maxId,
 215        long minIdleTimeInMilliseconds,
 216        CancellationToken cancellationToken)
 392217        => WithCancellation(
 392218            _database.StreamPendingMessagesAsync(
 392219                stream,
 392220                groupName,
 392221                count,
 392222                consumerName,
 392223                minId,
 392224                maxId,
 392225                minIdleTimeInMilliseconds),
 392226            cancellationToken);
 227
 228    /// <summary>Runs the StreamClaimAsync operation.</summary>
 229    public Task<StreamEntry[]> StreamClaimAsync(
 230        RedisKey stream,
 231        RedisValue groupName,
 232        RedisValue consumerName,
 233        long minIdleTimeInMilliseconds,
 234        RedisValue[] messageIds,
 235        CancellationToken cancellationToken)
 5236        => WithCancellation(
 5237            _database.StreamClaimAsync(
 5238                stream,
 5239                groupName,
 5240                consumerName,
 5241                minIdleTimeInMilliseconds,
 5242                messageIds),
 5243            cancellationToken);
 244
 245    /// <summary>Runs the StreamClaimIdsOnlyAsync operation.</summary>
 246    public Task<RedisValue[]> StreamClaimIdsOnlyAsync(
 247        RedisKey stream,
 248        RedisValue groupName,
 249        RedisValue consumerName,
 250        long minIdleTimeInMilliseconds,
 251        RedisValue[] messageIds,
 252        CancellationToken cancellationToken)
 2253        => WithCancellation(
 2254            _database.StreamClaimIdsOnlyAsync(
 2255                stream,
 2256                groupName,
 2257                consumerName,
 2258                minIdleTimeInMilliseconds,
 2259                messageIds),
 2260            cancellationToken);
 261
 262    private async Task<T> WithCancellation<T>(Task<T> command, CancellationToken cancellationToken)
 263    {
 264        // StackExchange.Redis enforces its own sync/async command timeouts; this adds an upper bound that
 265        // also honors the caller's token (e.g. host shutdown). On timeout the in-flight command is
 266        // abandoned best-effort — the multiplexer keeps running it — and surfaced as a TimeoutException so
 267        // the retry paths treat it as transient, while a genuine caller cancellation stays an
 268        // OperationCanceledException and is not retried.
 5403269        using var timeout = new CancellationTokenSource(_operationTimeout);
 5403270        using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeout.Token);
 271        try
 272        {
 5403273            return await command.WaitAsync(linked.Token).ConfigureAwait(false);
 274        }
 20275        catch (OperationCanceledException) when (timeout.IsCancellationRequested && !cancellationToken.IsCancellationReq
 276        {
 2277            throw new TimeoutException($"The Redis command did not complete within {_operationTimeout}.");
 278        }
 5381279    }
 280}
 281
 282internal static class RedisTransportRetry
 283{
 284    /// <summary>Runs this background operation until cancellation is requested.</summary>
 285    public static Task<T> ExecuteAsync<T>(
 286        Func<CancellationToken, Task<T>> action,
 287        int maxAttempts,
 288        TimeSpan baseDelay,
 289        TimeSpan maxDelay,
 290        CancellationToken cancellationToken)
 291        => AsyncResponseRetry.ExecuteAsync(action, IsTransient, maxAttempts, baseDelay, maxDelay, cancellationToken);
 292
 293    /// <summary>Runs the IsTransient operation.</summary>
 294    public static bool IsTransient(Exception exception)
 295        => exception is RedisConnectionException
 296            or RedisTimeoutException
 297            or TimeoutException;
 298}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(StackExchange.Redis.IDatabase,System.TimeSpan)
StreamAddAsync(StackExchange.Redis.RedisKey,StackExchange.Redis.NameValueEntry[],System.Nullable`1<System.Int64>,System.Boolean,System.Threading.CancellationToken)
ToInt32MaxLength(System.Nullable`1<System.Int64>)
StreamAddOnceAsync()
StreamCreateConsumerGroupAsync(StackExchange.Redis.RedisKey,StackExchange.Redis.RedisValue,StackExchange.Redis.RedisValue,System.Boolean,System.Threading.CancellationToken)
StreamReadGroupAsync(StackExchange.Redis.RedisKey,StackExchange.Redis.RedisValue,StackExchange.Redis.RedisValue,System.Int32,System.Threading.CancellationToken)
StreamAcknowledgeAsync(StackExchange.Redis.RedisKey,StackExchange.Redis.RedisValue,StackExchange.Redis.RedisValue,System.Threading.CancellationToken)
StreamPendingMessagesAsync(StackExchange.Redis.RedisKey,StackExchange.Redis.RedisValue,System.Int32,StackExchange.Redis.RedisValue,System.Nullable`1<StackExchange.Redis.RedisValue>,System.Nullable`1<StackExchange.Redis.RedisValue>,System.Int64,System.Threading.CancellationToken)
StreamClaimAsync(StackExchange.Redis.RedisKey,StackExchange.Redis.RedisValue,StackExchange.Redis.RedisValue,System.Int64,StackExchange.Redis.RedisValue[],System.Threading.CancellationToken)
StreamClaimIdsOnlyAsync(StackExchange.Redis.RedisKey,StackExchange.Redis.RedisValue,StackExchange.Redis.RedisValue,System.Int64,StackExchange.Redis.RedisValue[],System.Threading.CancellationToken)
WithCancellation()