| | | 1 | | using StackExchange.Redis; |
| | | 2 | | |
| | | 3 | | namespace AsyncResponse.Transports.Redis; |
| | | 4 | | |
| | | 5 | | /// <summary> |
| | | 6 | | /// Resolves Redis stream names from transport options. These names are deployment contracts: changing |
| | | 7 | | /// them while entries are in flight strands unprocessed worker or response messages in the old stream. |
| | | 8 | | /// </summary> |
| | 3333 | 9 | | internal sealed class RedisTransportKeySchema(RedisAsyncResponseTransportOptions _options) |
| | | 10 | | { |
| | 10370 | 11 | | public RedisKey WorkerStream => Resolve(_options.WorkerStream, "worker"); |
| | 6257 | 12 | | public RedisKey ResponseStream => Resolve(_options.ResponseStream, "response"); |
| | 1841 | 13 | | public RedisKey DeadLetterStream => Resolve(_options.DeadLetterStream, "deadletter"); |
| | | 14 | | |
| | | 15 | | /// <summary> |
| | | 16 | | /// Per-publish dedup marker for the idempotent worker XADD. Hash-tagged with the stream's own |
| | | 17 | | /// slot key so the marker and the stream share a cluster slot — the MULTI/EXEC that couples |
| | | 18 | | /// them would otherwise fail with CROSSSLOT on Redis Cluster. Wrapping the whole stream name |
| | | 19 | | /// in braces nested them when the name already carried a hash tag (the idiomatic |
| | | 20 | | /// <c>KeyPrefix = "{app}"</c> co-location), and Redis then read <c>{app</c> as the marker's |
| | | 21 | | /// tag — a different slot, and CROSSSLOT on every publish. |
| | | 22 | | /// </summary> |
| | | 23 | | public RedisKey WorkerPublishDedupKey(string publishId) |
| | 423 | 24 | | => $"{{{HashTagOf(((string?)WorkerStream)!)}}}:publish:{publishId}"; |
| | | 25 | | |
| | | 26 | | /// <summary> |
| | | 27 | | /// The key text Redis Cluster hashes for <paramref name="key"/>: the substring between the |
| | | 28 | | /// first <c>{</c> and the first <c>}</c> after it when that is non-empty, otherwise the whole |
| | | 29 | | /// key. A name whose braces do not form one such tag returns unchanged — see the validator, |
| | | 30 | | /// which rejects those, because no marker key could then share the stream's slot. |
| | | 31 | | /// </summary> |
| | | 32 | | internal static string HashTagOf(string key) |
| | | 33 | | { |
| | 431 | 34 | | var open = key.IndexOf('{'); |
| | 431 | 35 | | if (open < 0) |
| | 421 | 36 | | return key; |
| | 10 | 37 | | var close = key.IndexOf('}', open + 1); |
| | 10 | 38 | | return close > open + 1 ? key.Substring(open + 1, close - open - 1) : key; |
| | | 39 | | } |
| | | 40 | | |
| | | 41 | | private RedisKey Resolve(string? configured, string role) |
| | 18468 | 42 | | => !string.IsNullOrWhiteSpace(configured) |
| | 18468 | 43 | | ? configured |
| | 18468 | 44 | | : $"{_options.KeyPrefix}:transport:{role}"; |
| | | 45 | | } |