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

Information
Class: AsyncResponse.Transports.Redis.IRedisStreamDatabase
Assembly: AsyncResponse.Transports.Redis
File(s): /_/src/Transports/AsyncResponse.Transports.Redis/RedisTransportClientAdapters.cs
Line coverage
0%
Covered lines: 0
Uncovered lines: 2
Coverable lines: 2
Total lines: 298
Line coverage: 0%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
StreamClaimIdsOnlyAsync(...)100%210%
StreamAddOnceAsync(...)100%210%

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)
 066        => 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)
 086        => StreamAddAsync(stream, values, maxLength, useApproximateMaxLength, cancellationToken);
 87}
 88
 89internal 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.
 128        => WithCancellation(
 129            _database.StreamAddAsync(
 130                stream,
 131                values,
 132                messageId: (RedisValue?)null,
 133                maxLength: ToInt32MaxLength(maxLength),
 134                useApproximateMaxLength: useApproximateMaxLength,
 135                flags: CommandFlags.None),
 136            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)
 141        => 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    {
 153        ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(dedupTtl, TimeSpan.Zero);
 154        var args = new RedisValue[3 + values.Length * 2];
 155        args[0] = checked((long)Math.Ceiling(dedupTtl.TotalMilliseconds));
 156        args[1] = ToInt32MaxLength(maxLength) is { } cap ? cap : RedisValue.EmptyString;
 157        args[2] = useApproximateMaxLength ? "~" : "=";
 158        for (var i = 0; i < values.Length; i++)
 159        {
 160            args[3 + i * 2] = values[i].Name;
 161            args[4 + i * 2] = values[i].Value;
 162        }
 163
 164        return (RedisValue)await WithCancellation(
 165            _database.ScriptEvaluateAsync(AppendOnceScript, [stream, dedupKey], args),
 166            cancellationToken).ConfigureAwait(false);
 167    }
 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)
 176        => WithCancellation(
 177            _database.StreamCreateConsumerGroupAsync(stream, groupName, position, createStream),
 178            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)
 187        => WithCancellation(
 188            _database.StreamReadGroupAsync(
 189                stream,
 190                groupName,
 191                consumerName,
 192                position: StreamPosition.NewMessages,
 193                count: count,
 194                noAck: false),
 195            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)
 203        => WithCancellation(
 204            _database.StreamAcknowledgeAsync(stream, groupName, messageId),
 205            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)
 217        => WithCancellation(
 218            _database.StreamPendingMessagesAsync(
 219                stream,
 220                groupName,
 221                count,
 222                consumerName,
 223                minId,
 224                maxId,
 225                minIdleTimeInMilliseconds),
 226            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)
 236        => WithCancellation(
 237            _database.StreamClaimAsync(
 238                stream,
 239                groupName,
 240                consumerName,
 241                minIdleTimeInMilliseconds,
 242                messageIds),
 243            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)
 253        => WithCancellation(
 254            _database.StreamClaimIdsOnlyAsync(
 255                stream,
 256                groupName,
 257                consumerName,
 258                minIdleTimeInMilliseconds,
 259                messageIds),
 260            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.
 269        using var timeout = new CancellationTokenSource(_operationTimeout);
 270        using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeout.Token);
 271        try
 272        {
 273            return await command.WaitAsync(linked.Token).ConfigureAwait(false);
 274        }
 275        catch (OperationCanceledException) when (timeout.IsCancellationRequested && !cancellationToken.IsCancellationReq
 276        {
 277            throw new TimeoutException($"The Redis command did not complete within {_operationTimeout}.");
 278        }
 279    }
 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

StreamClaimIdsOnlyAsync(StackExchange.Redis.RedisKey,StackExchange.Redis.RedisValue,StackExchange.Redis.RedisValue,System.Int64,StackExchange.Redis.RedisValue[],System.Threading.CancellationToken)
StreamAddOnceAsync(StackExchange.Redis.RedisKey,StackExchange.Redis.RedisKey,System.TimeSpan,StackExchange.Redis.NameValueEntry[],System.Nullable`1<System.Int64>,System.Boolean,System.Threading.CancellationToken)