| | | 1 | | using System.Runtime.CompilerServices; |
| | | 2 | | using System.Text.Json; |
| | | 3 | | |
| | | 4 | | namespace AsyncResponse.Transports; |
| | | 5 | | |
| | | 6 | | // Shared source for every transport's correlation-id extractor (the 7 broker transports plus |
| | | 7 | | // DbCorrelationIdExtractor, which the 3 database transports funnel through): each csproj pulls |
| | | 8 | | // this file in via <Compile Include="..\Shared\CorrelationIdJsonPaths.cs" />, so it compiles INTO |
| | | 9 | | // each provider assembly. The provider-specific fast path (header/property/attribute lookup) stays |
| | | 10 | | // in the per-transport extractor; only the JSON-body path walk lives here. |
| | | 11 | | |
| | | 12 | | /// <summary> |
| | | 13 | | /// Locates the AsyncResponse correlation id inside a message body via configured dotted JSON paths. |
| | | 14 | | /// Configured paths are pre-split into segments once per distinct <c>CorrelationIdJsonPaths</c> |
| | | 15 | | /// array instead of on every delivered message — the array is startup configuration read by every |
| | | 16 | | /// provider's options, not per-message data. The body is walked over a <see cref="JsonDocument"/> |
| | | 17 | | /// rather than a mutable <see cref="System.Text.Json.Nodes.JsonNode"/> DOM: a <see |
| | | 18 | | /// cref="JsonDocument"/> rents its buffer from the array pool and returns it on <c>Dispose</c>, |
| | | 19 | | /// and <see cref="JsonElement.EnumerateObject"/> never allocates a wrapper node per property, where |
| | | 20 | | /// the DOM allocates one node per visited property and is never disposed by its caller. |
| | | 21 | | /// </summary> |
| | | 22 | | internal static class CorrelationIdJsonPaths |
| | | 23 | | { |
| | | 24 | | // Keyed by the options-held array's identity (never its content): the array is validated once |
| | | 25 | | // at subscriber startup and re-read on every delivered message afterward, so this amortizes |
| | | 26 | | // path.Split across the process lifetime of a subscriber instead of paying it per message. A |
| | | 27 | | // ConditionalWeakTable lets a replaced array (or one built by a short-lived options instance in |
| | | 28 | | // a test) fall out of the cache with the array itself instead of being pinned forever. |
| | 2 | 29 | | private static readonly ConditionalWeakTable<string[], string[][]> SplitPathCache = new(); |
| | | 30 | | |
| | | 31 | | /// <summary> |
| | | 32 | | /// Returns the first non-blank value found by walking <paramref name="jsonPaths"/> against |
| | | 33 | | /// <paramref name="messageJson"/> in order, or <see langword="null"/> when none resolve. |
| | | 34 | | /// </summary> |
| | | 35 | | public static string? Extract(string messageJson, string[]? jsonPaths) |
| | | 36 | | { |
| | 34 | 37 | | if (jsonPaths is null || jsonPaths.Length == 0 || string.IsNullOrWhiteSpace(messageJson)) |
| | 4 | 38 | | return null; |
| | | 39 | | |
| | 30 | 40 | | var splitPaths = GetSplitPaths(jsonPaths); |
| | | 41 | | |
| | | 42 | | JsonDocument document; |
| | | 43 | | try |
| | | 44 | | { |
| | 30 | 45 | | document = JsonDocument.Parse(messageJson); |
| | 28 | 46 | | } |
| | 2 | 47 | | catch (JsonException) |
| | | 48 | | { |
| | 2 | 49 | | return null; |
| | | 50 | | } |
| | | 51 | | |
| | 28 | 52 | | using (document) |
| | | 53 | | { |
| | 148 | 54 | | foreach (var segments in splitPaths) |
| | | 55 | | { |
| | 52 | 56 | | var value = TryReadPath(document.RootElement, segments); |
| | 52 | 57 | | if (!string.IsNullOrWhiteSpace(value)) |
| | 12 | 58 | | return value; |
| | | 59 | | } |
| | 16 | 60 | | } |
| | | 61 | | |
| | 16 | 62 | | return null; |
| | 14 | 63 | | } |
| | | 64 | | |
| | | 65 | | private static string[][] GetSplitPaths(string[] jsonPaths) |
| | 30 | 66 | | => SplitPathCache.GetValue(jsonPaths, static paths => |
| | 30 | 67 | | { |
| | 30 | 68 | | // A blank configured path is dropped here rather than cached as a no-op entry, |
| | 30 | 69 | | // mirroring the pre-shared walker's per-call IsNullOrWhiteSpace(path) short-circuit. |
| | 30 | 70 | | // A non-blank path of only dots/whitespace (e.g. ".") is NOT blank by that check, so |
| | 30 | 71 | | // it is kept even though it splits to zero segments — TryReadPath below walks zero |
| | 30 | 72 | | // steps and reads the message root itself, same as the pre-shared walker did. |
| | 30 | 73 | | var nonBlankCount = 0; |
| | 224 | 74 | | foreach (var path in paths) |
| | 30 | 75 | | { |
| | 82 | 76 | | if (!string.IsNullOrWhiteSpace(path)) |
| | 80 | 77 | | nonBlankCount++; |
| | 30 | 78 | | } |
| | 30 | 79 | | |
| | 30 | 80 | | var split = new string[nonBlankCount][]; |
| | 30 | 81 | | var index = 0; |
| | 224 | 82 | | foreach (var path in paths) |
| | 30 | 83 | | { |
| | 82 | 84 | | if (string.IsNullOrWhiteSpace(path)) |
| | 30 | 85 | | continue; |
| | 80 | 86 | | split[index++] = path.Split('.', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries) |
| | 30 | 87 | | } |
| | 30 | 88 | | |
| | 30 | 89 | | return split; |
| | 30 | 90 | | }); |
| | | 91 | | |
| | | 92 | | private static string? TryReadPath(JsonElement root, string[] segments) |
| | | 93 | | { |
| | 52 | 94 | | var current = root; |
| | | 95 | | |
| | | 96 | | // An embedded JSON string can appear at any segment, so the unwrap below can reparse |
| | | 97 | | // more than once per path; only the most recent reparse needs to stay alive; a value it |
| | | 98 | | // produced is copied out (GetString/GetRawText) before it is disposed. |
| | 52 | 99 | | JsonDocument? scratch = null; |
| | | 100 | | try |
| | | 101 | | { |
| | 198 | 102 | | foreach (var segment in segments) |
| | | 103 | | { |
| | 64 | 104 | | current = UnwrapJsonString(current, ref scratch); |
| | 62 | 105 | | if (current.ValueKind != JsonValueKind.Object) |
| | 14 | 106 | | return null; |
| | | 107 | | |
| | 48 | 108 | | if (!TryGetProperty(current, segment, out var next)) |
| | 16 | 109 | | return null; |
| | | 110 | | |
| | 30 | 111 | | current = next; |
| | | 112 | | } |
| | | 113 | | |
| | 18 | 114 | | current = UnwrapJsonString(current, ref scratch); |
| | 14 | 115 | | return current.ValueKind switch |
| | 14 | 116 | | { |
| | 10 | 117 | | JsonValueKind.String => current.GetString(), |
| | 2 | 118 | | JsonValueKind.Number or JsonValueKind.True or JsonValueKind.False => current.GetRawText(), |
| | 2 | 119 | | _ => null |
| | 14 | 120 | | }; |
| | | 121 | | } |
| | 8 | 122 | | catch (InvalidOperationException) |
| | | 123 | | { |
| | | 124 | | // An ESCAPED lone surrogate ("\ud800") is well-formed JSON — Parse accepts it — but it |
| | | 125 | | // has no UTF-16 string form, so transcoding it (JsonProperty.Name on any property of a |
| | | 126 | | // walked object, GetString on the value or on an embedded-JSON candidate) throws |
| | | 127 | | // InvalidOperationException, not JsonException. Same rule as the duplicate key below: |
| | | 128 | | // unresolvable, NOT a failure — an escape here made an unroutable inbound message a |
| | | 129 | | // handler failure, which on RabbitMQ's default MaxDeliveryAttempts = 0 requeues forever. |
| | 8 | 130 | | return null; |
| | | 131 | | } |
| | | 132 | | finally |
| | | 133 | | { |
| | 52 | 134 | | scratch?.Dispose(); |
| | 52 | 135 | | } |
| | 52 | 136 | | } |
| | | 137 | | |
| | | 138 | | // Mirrors JsonObject.TryGetPropertyValue then a case-insensitive scan: an exact (ordinal) match |
| | | 139 | | // wins when present, else the first case-insensitive match in document order. JsonObject |
| | | 140 | | // materializes its whole backing dictionary the instant any property is touched, so on this |
| | | 141 | | // runtime looking up ANY property of an object that contains an exact-duplicate key anywhere — |
| | | 142 | | // not only when the duplicate is the key being requested — throws ArgumentException. A |
| | | 143 | | // JsonElement walk materializes nothing, so the object is scanned once here to detect and |
| | | 144 | | // reproduce that rather than silently resolving to one of the duplicates. |
| | | 145 | | private static bool TryGetProperty(JsonElement obj, string name, out JsonElement value) |
| | | 146 | | { |
| | 48 | 147 | | var seen = RentSeenNames(); |
| | | 148 | | try |
| | | 149 | | { |
| | 48 | 150 | | return TryGetProperty(obj, name, seen, out value); |
| | | 151 | | } |
| | | 152 | | finally |
| | | 153 | | { |
| | 48 | 154 | | ReturnSeenNames(seen); |
| | 48 | 155 | | } |
| | 46 | 156 | | } |
| | | 157 | | |
| | | 158 | | // The duplicate-key scan needs a set of the object's property names, and used to allocate a |
| | | 159 | | // fresh one — set, buckets, and every growth step on the way to the object's size — for every |
| | | 160 | | // segment of every configured path of every delivered message. The walk is synchronous and |
| | | 161 | | // never re-enters itself, so one set per thread serves them all; a set grown by an unusually |
| | | 162 | | // wide object is dropped rather than pinned on the thread for good. |
| | | 163 | | [ThreadStatic] |
| | | 164 | | private static HashSet<string>? t_seenNames; |
| | | 165 | | |
| | | 166 | | private const int MaxRetainedSeenNames = 256; |
| | | 167 | | |
| | | 168 | | private static HashSet<string> RentSeenNames() |
| | | 169 | | { |
| | 48 | 170 | | var seen = t_seenNames ?? new HashSet<string>(StringComparer.Ordinal); |
| | 48 | 171 | | t_seenNames = null; |
| | 48 | 172 | | return seen; |
| | | 173 | | } |
| | | 174 | | |
| | | 175 | | private static void ReturnSeenNames(HashSet<string> seen) |
| | | 176 | | { |
| | 48 | 177 | | if (seen.Count > MaxRetainedSeenNames) |
| | 0 | 178 | | return; |
| | | 179 | | |
| | 48 | 180 | | seen.Clear(); |
| | 48 | 181 | | t_seenNames = seen; |
| | 48 | 182 | | } |
| | | 183 | | |
| | | 184 | | private static bool TryGetProperty(JsonElement obj, string name, HashSet<string> seen, out JsonElement value) |
| | | 185 | | { |
| | 48 | 186 | | var exactFound = false; |
| | 48 | 187 | | JsonElement exactValue = default; |
| | 48 | 188 | | var caseInsensitiveFound = false; |
| | 48 | 189 | | JsonElement caseInsensitiveValue = default; |
| | | 190 | | |
| | 200 | 191 | | foreach (var property in obj.EnumerateObject()) |
| | | 192 | | { |
| | | 193 | | // Read ONCE: JsonProperty.Name transcodes a new string on every call, and the three |
| | | 194 | | // reads below used to cost three strings per property of every object walked. |
| | 54 | 195 | | var propertyName = property.Name; |
| | 52 | 196 | | if (!seen.Add(propertyName)) |
| | | 197 | | { |
| | | 198 | | // Unresolvable, NOT a failure: extraction cannot choose between the duplicates, so |
| | | 199 | | // the id is simply not in this body. Throwing here made an unroutable inbound |
| | | 200 | | // message a handler failure, which the ingress explicitly refuses to do (see |
| | | 201 | | // AsyncResponseIngress: an id that cannot route is acknowledged, never redelivered, |
| | | 202 | | // because RabbitMQ's default MaxDeliveryAttempts = 0 has no cap). Returning false |
| | | 203 | | // lets the caller log-and-ack it instead of hot-looping at broker rate. |
| | | 204 | | // |
| | | 205 | | // The duplicate's NAME is deliberately not surfaced anywhere: it comes straight off |
| | | 206 | | // an untrusted body, and the "never logs a message body" guarantee covers it. |
| | 2 | 207 | | value = default; |
| | 2 | 208 | | return false; |
| | | 209 | | } |
| | | 210 | | |
| | 50 | 211 | | if (!exactFound && string.Equals(propertyName, name, StringComparison.Ordinal)) |
| | | 212 | | { |
| | 30 | 213 | | exactFound = true; |
| | 30 | 214 | | exactValue = property.Value; |
| | | 215 | | } |
| | 20 | 216 | | else if (!caseInsensitiveFound && string.Equals(propertyName, name, StringComparison.OrdinalIgnoreCase)) |
| | | 217 | | { |
| | 2 | 218 | | caseInsensitiveFound = true; |
| | 2 | 219 | | caseInsensitiveValue = property.Value; |
| | | 220 | | } |
| | | 221 | | } |
| | | 222 | | |
| | 44 | 223 | | if (exactFound) |
| | | 224 | | { |
| | 28 | 225 | | value = exactValue; |
| | 28 | 226 | | return true; |
| | | 227 | | } |
| | | 228 | | |
| | 16 | 229 | | if (caseInsensitiveFound) |
| | | 230 | | { |
| | 2 | 231 | | value = caseInsensitiveValue; |
| | 2 | 232 | | return true; |
| | | 233 | | } |
| | | 234 | | |
| | 14 | 235 | | value = default; |
| | 14 | 236 | | return false; |
| | 2 | 237 | | } |
| | | 238 | | |
| | | 239 | | // A JSON-string-encoded object/array (e.g. a broker envelope carrying its payload as an escaped |
| | | 240 | | // string) is parsed and descended into transparently, same as the pre-shared walker. |
| | | 241 | | private static JsonElement UnwrapJsonString(JsonElement element, ref JsonDocument? scratch) |
| | | 242 | | { |
| | 82 | 243 | | if (element.ValueKind != JsonValueKind.String) |
| | 62 | 244 | | return element; |
| | | 245 | | |
| | 20 | 246 | | var text = element.GetString(); |
| | 14 | 247 | | if (text is null) |
| | 0 | 248 | | return element; |
| | | 249 | | |
| | 14 | 250 | | var trimmed = text.AsSpan().TrimStart(); |
| | 14 | 251 | | if (trimmed.Length == 0 || (trimmed[0] != '{' && trimmed[0] != '[')) |
| | 10 | 252 | | return element; |
| | | 253 | | |
| | | 254 | | JsonDocument parsed; |
| | | 255 | | try |
| | | 256 | | { |
| | 4 | 257 | | parsed = JsonDocument.Parse(text); |
| | 2 | 258 | | } |
| | 2 | 259 | | catch (JsonException) |
| | | 260 | | { |
| | 2 | 261 | | return element; |
| | | 262 | | } |
| | | 263 | | |
| | 2 | 264 | | scratch?.Dispose(); |
| | 2 | 265 | | scratch = parsed; |
| | 2 | 266 | | return parsed.RootElement; |
| | 2 | 267 | | } |
| | | 268 | | } |