| | | 1 | | namespace AsyncResponse.Transports.Redis; |
| | | 2 | | |
| | | 3 | | internal static class RedisTransportOptionsValidator |
| | | 4 | | { |
| | | 5 | | /// <summary>Validates the supplied options.</summary> |
| | | 6 | | public static string Required(string? value, string name) |
| | 12116 | 7 | | => !string.IsNullOrWhiteSpace(value) |
| | 12116 | 8 | | ? value |
| | 12116 | 9 | | : throw new InvalidOperationException($"{nameof(RedisAsyncResponseTransportOptions)}.{name} must be configur |
| | | 10 | | |
| | | 11 | | /// <summary>Validates the supplied options.</summary> |
| | | 12 | | public static void PositiveOrNull(long? value, string name) |
| | | 13 | | { |
| | 3566 | 14 | | if (value is <= 0) |
| | 4 | 15 | | throw new InvalidOperationException($"{nameof(RedisAsyncResponseTransportOptions)}.{name} must be positive w |
| | 3562 | 16 | | } |
| | | 17 | | |
| | | 18 | | /// <summary>Validates the supplied options.</summary> |
| | | 19 | | public static void ValidateCommon(RedisAsyncResponseTransportOptions options) |
| | | 20 | | { |
| | 1814 | 21 | | _ = Required(options.KeyPrefix, nameof(options.KeyPrefix)); |
| | 1808 | 22 | | _ = Required(options.WorkerConsumerGroup, nameof(options.WorkerConsumerGroup)); |
| | 1806 | 23 | | _ = Required(options.ResponseConsumerGroup, nameof(options.ResponseConsumerGroup)); |
| | 1804 | 24 | | _ = Required(options.CorrelationIdField, nameof(options.CorrelationIdField)); |
| | 1802 | 25 | | _ = Required(options.PayloadField, nameof(options.PayloadField)); |
| | 1800 | 26 | | _ = Required(options.DefaultReplyTargetName, nameof(options.DefaultReplyTargetName)); |
| | | 27 | | |
| | | 28 | | // Worker and response subscribers must never share one stream: a Redis stream is not |
| | | 29 | | // partitioned between consumer groups — every entry is visible to every group — so a |
| | | 30 | | // shared stream would feed worker jobs to the response ingress and responses to the |
| | | 31 | | // worker dispatcher. Compare the resolved names so an explicit value colliding with the |
| | | 32 | | // other role's derived default is caught too. |
| | 1798 | 33 | | var schema = new RedisTransportKeySchema(options); |
| | 1798 | 34 | | if (StringComparer.Ordinal.Equals(schema.WorkerStream.ToString(), schema.ResponseStream.ToString())) |
| | | 35 | | { |
| | 0 | 36 | | throw new InvalidOperationException( |
| | 0 | 37 | | $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(options.WorkerStream)} and " + |
| | 0 | 38 | | $"{nameof(options.ResponseStream)} must resolve to distinct streams so worker and response " + |
| | 0 | 39 | | "subscribers do not consume each other's messages."); |
| | | 40 | | } |
| | | 41 | | |
| | | 42 | | // Parity with the Kafka/NATS validators: the dead-letter stream must not be a live one. |
| | | 43 | | // Streams fan out to every consumer group, so a dead-letter XADD into the worker stream is |
| | | 44 | | // read back as a brand-new entry with Attempt=1 — fail, dead-letter, re-read, an unbounded |
| | | 45 | | // loop re-running the handler's side effects; aimed at the response stream, poison worker |
| | | 46 | | // envelopes complete live waiters. Compare the resolved names so an explicit value |
| | | 47 | | // colliding with a derived default is caught too. |
| | 1798 | 48 | | var deadLetterStream = schema.DeadLetterStream.ToString(); |
| | 1798 | 49 | | if (StringComparer.Ordinal.Equals(deadLetterStream, schema.WorkerStream.ToString()) |
| | 1798 | 50 | | || StringComparer.Ordinal.Equals(deadLetterStream, schema.ResponseStream.ToString())) |
| | | 51 | | { |
| | 6 | 52 | | throw new InvalidOperationException( |
| | 6 | 53 | | $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(options.DeadLetterStream)} must resolve to a stre |
| | 6 | 54 | | $"distinct from {nameof(options.WorkerStream)} and {nameof(options.ResponseStream)} " + |
| | 6 | 55 | | $"(it resolves to '{deadLetterStream}') so dead-lettered messages park instead of re-entering live consu |
| | | 56 | | } |
| | | 57 | | |
| | | 58 | | // The worker publish dedup marker must share the worker stream's cluster slot (a |
| | | 59 | | // MULTI/EXEC couples them). The schema reuses the stream's own hash tag when it carries a |
| | | 60 | | // well-formed one; a name whose braces do NOT form one has no marker key that can land in |
| | | 61 | | // its slot, so reject it here instead of failing every publish with CROSSSLOT. |
| | 1792 | 62 | | var workerStream = schema.WorkerStream.ToString(); |
| | 1792 | 63 | | if (workerStream.AsSpan().IndexOfAny('{', '}') >= 0 |
| | 1792 | 64 | | && string.Equals(RedisTransportKeySchema.HashTagOf(workerStream), workerStream, StringComparison.Ordinal)) |
| | | 65 | | { |
| | 4 | 66 | | throw new InvalidOperationException( |
| | 4 | 67 | | $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(options.WorkerStream)} (resolved to '{workerStrea |
| | 4 | 68 | | "braces that do not form one well-formed Redis hash tag ('{tag}' with a non-empty tag). The idempotent p |
| | 4 | 69 | | "must share the stream's cluster slot, which is only possible when the name has no braces or exactly one |
| | | 70 | | } |
| | | 71 | | |
| | | 72 | | // OperationTimeout arms a CancellationTokenSource per command; the retry delays feed |
| | | 73 | | // Task.Delay — all timer-armed, so all carry the .NET timer ceiling. |
| | 1788 | 74 | | AsyncResponseChannelOptions.EnsureTimerBacked(options.OperationTimeout, nameof(RedisAsyncResponseTransportOption |
| | 1784 | 75 | | AsyncResponseChannelOptions.EnsureTimerBacked(options.PublishRetryBaseDelay, nameof(RedisAsyncResponseTransportO |
| | 1784 | 76 | | AsyncResponseChannelOptions.EnsureTimerBacked(options.PublishRetryMaxDelay, nameof(RedisAsyncResponseTransportOp |
| | 1784 | 77 | | AsyncResponseChannelOptions.EnsureTimerBacked(options.SubscriberRetryBaseDelay, nameof(RedisAsyncResponseTranspo |
| | 1784 | 78 | | AsyncResponseChannelOptions.EnsureTimerBacked(options.SubscriberRetryMaxDelay, nameof(RedisAsyncResponseTranspor |
| | 1784 | 79 | | PositiveOrNull(options.StreamMaxLength, nameof(options.StreamMaxLength)); |
| | 1782 | 80 | | PositiveOrNull(options.DeadLetterStreamMaxLength, nameof(options.DeadLetterStreamMaxLength)); |
| | | 81 | | |
| | 1780 | 82 | | if (options.PublishMaxAttempts <= 0) |
| | 2 | 83 | | throw new InvalidOperationException($"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(options.PublishMa |
| | | 84 | | |
| | 1778 | 85 | | if (options.PublishRetryBaseDelay > options.PublishRetryMaxDelay) |
| | | 86 | | { |
| | 2 | 87 | | throw new InvalidOperationException( |
| | 2 | 88 | | $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(options.PublishRetryBaseDelay)} cannot exceed " + |
| | 2 | 89 | | $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(options.PublishRetryMaxDelay)}."); |
| | | 90 | | } |
| | | 91 | | |
| | 1776 | 92 | | if (options.SubscriberRetryBaseDelay > options.SubscriberRetryMaxDelay) |
| | | 93 | | { |
| | 2 | 94 | | throw new InvalidOperationException( |
| | 2 | 95 | | $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(options.SubscriberRetryBaseDelay)} cannot exceed |
| | 2 | 96 | | $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(options.SubscriberRetryMaxDelay)}."); |
| | | 97 | | } |
| | | 98 | | |
| | 1774 | 99 | | if (options.HostShutdownTimeout is { } hostShutdownTimeout && hostShutdownTimeout <= TimeSpan.Zero) |
| | 2 | 100 | | throw new InvalidOperationException($"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(options.HostShutd |
| | 1772 | 101 | | } |
| | | 102 | | } |