| | | 1 | | using StackExchange.Redis; |
| | | 2 | | |
| | | 3 | | namespace AsyncResponse.Transports.Redis; |
| | | 4 | | |
| | | 5 | | internal 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 | | |
| | 3 | 53 | | internal 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. |
| | 3 | 69 | | => WithCancellation( |
| | 3 | 70 | | _database.StreamAddAsync( |
| | 3 | 71 | | stream, |
| | 3 | 72 | | values, |
| | 3 | 73 | | messageId: (RedisValue?)null, |
| | 3 | 74 | | maxLength: ToInt32MaxLength(maxLength), |
| | 3 | 75 | | useApproximateMaxLength: useApproximateMaxLength, |
| | 3 | 76 | | flags: CommandFlags.None), |
| | 3 | 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) |
| | 3 | 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) |
| | 3 | 91 | | => WithCancellation( |
| | 3 | 92 | | _database.StreamCreateConsumerGroupAsync(stream, groupName, position, createStream), |
| | 3 | 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) |
| | 3 | 102 | | => WithCancellation( |
| | 3 | 103 | | _database.StreamReadGroupAsync( |
| | 3 | 104 | | stream, |
| | 3 | 105 | | groupName, |
| | 3 | 106 | | consumerName, |
| | 3 | 107 | | position: StreamPosition.NewMessages, |
| | 3 | 108 | | count: count, |
| | 3 | 109 | | noAck: false), |
| | 3 | 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) |
| | 3 | 118 | | => WithCancellation( |
| | 3 | 119 | | _database.StreamAcknowledgeAsync(stream, groupName, messageId), |
| | 3 | 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) |
| | 3 | 132 | | => WithCancellation( |
| | 3 | 133 | | _database.StreamPendingMessagesAsync( |
| | 3 | 134 | | stream, |
| | 3 | 135 | | groupName, |
| | 3 | 136 | | count, |
| | 3 | 137 | | consumerName, |
| | 3 | 138 | | minId, |
| | 3 | 139 | | maxId, |
| | 3 | 140 | | minIdleTimeInMilliseconds), |
| | 3 | 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) |
| | 3 | 151 | | => WithCancellation( |
| | 3 | 152 | | _database.StreamClaimAsync( |
| | 3 | 153 | | stream, |
| | 3 | 154 | | groupName, |
| | 3 | 155 | | consumerName, |
| | 3 | 156 | | minIdleTimeInMilliseconds, |
| | 3 | 157 | | messageIds), |
| | 3 | 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. |
| | 3 | 167 | | using var timeout = new CancellationTokenSource(_operationTimeout); |
| | 3 | 168 | | using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeout.Token); |
| | | 169 | | try |
| | | 170 | | { |
| | 3 | 171 | | return await command.WaitAsync(linked.Token).ConfigureAwait(false); |
| | | 172 | | } |
| | 3 | 173 | | catch (OperationCanceledException) when (timeout.IsCancellationRequested && !cancellationToken.IsCancellationReq |
| | | 174 | | { |
| | 3 | 175 | | throw new TimeoutException($"The Redis command did not complete within {_operationTimeout}."); |
| | | 176 | | } |
| | 3 | 177 | | } |
| | | 178 | | } |
| | | 179 | | |
| | | 180 | | internal 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 | | } |