From bb7b9b8fbc042f15af2dce67687243d1df6c4912 Mon Sep 17 00:00:00 2001 From: Alexander Bohle Date: Mon, 3 Aug 2026 22:54:57 +0100 Subject: [PATCH 1/8] Fix Mocha header serialization for byte arrays and harden the value mapping A byte[] header value crashed envelope serialization from 16.5.0 onward: the AOT change swapped the converter's fallback to options.GetTypeInfo without adding a byte[] case, so the value reached a resolver that could not supply metadata for it. byte[] is now written as base64, as it was in 16.3.0. Header values no longer reach reflection at all. The converter's options carry a resolver for the header types only, so an unsupported value fails with a JsonException naming the header instead of being serialized by whatever resolver happened to be installed. Both trimming suppressions are gone and a Native AOT publish compiles without ILC warnings. The supported set widens to cover value-type sequences, string-keyed dictionaries, JsonNode, ArraySegment, ReadOnlyMemory and Memory. The reader gains a ulong rung, and no longer infers a DateTime from a string that happens to parse as one. On the RabbitMQ side, AMQP values are mapped in both directions at every level of a nested value, so a dead-lettered message carrying x-death can be republished. Types a field table cannot express are mapped to a legal wire form rather than failing the publish, a byte payload travels as the table's byte array type so its CLR type no longer depends on its content, and a value nested too deeply or containing itself is reported instead of exhausting the stack. Co-Authored-By: Claude Fable 5 --- .../RabbitMQDispatchEndpoint.cs | 186 +++++++++- .../RabbitMQMessageEnvelopeParser.cs | 63 +++- src/Mocha/src/Mocha/Assembly.cs | 1 + .../Mocha/Serialization/HeaderValueText.cs | 27 ++ .../Mocha/Serialization/HeadersConverter.cs | 258 ++++++++++++-- .../test/Mocha.TestHelpers/FaultHeaders.cs | 33 ++ .../Headers/HeadersSerializationTests.cs | 15 +- .../HeadersJsonConverterTests.cs | 273 +++++++++++++- .../Transport/MessageEnvelopeWriterTests.cs | 77 ++++ .../Behaviors/ErrorQueueTests.cs | 11 +- .../Behaviors/ErrorQueueTests.cs | 5 +- .../HeaderValueConsistencyTests.cs | 113 ++++++ .../RabbitMQDispatchContextExtensionsTests.cs | 333 +++++++++++++++++- .../RabbitMQMessageEnvelopeParserTests.cs | 134 +++++++ website/content/docs/mocha/messages.md | 64 ++++ 15 files changed, 1522 insertions(+), 71 deletions(-) create mode 100644 src/Mocha/src/Mocha/Serialization/HeaderValueText.cs create mode 100644 src/Mocha/test/Mocha.TestHelpers/FaultHeaders.cs create mode 100644 src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/HeaderValueConsistencyTests.cs diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs index d2d2844bda5..4fcfb370fbf 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs @@ -1,3 +1,7 @@ +using System.Collections; +using System.Globalization; +using System.Text.Json; +using System.Text.Json.Nodes; using Mocha.Middlewares; using RabbitMQ.Client; using static System.StringSplitOptions; @@ -223,18 +227,7 @@ public static class RabbitMQDispatchContextExtensions { foreach (var header in envelope.Headers) { - if (header.Value is DateTimeOffset dateTimeOffset) - { - headers[header.Key] = new AmqpTimestamp(dateTimeOffset.ToUnixTimeSeconds()); - } - else if (header.Value is DateTime dateTime) - { - headers[header.Key] = new AmqpTimestamp(new DateTimeOffset(dateTime).ToUnixTimeSeconds()); - } - else if (header.Value is not null) - { - headers[header.Key] = header.Value; - } + headers[header.Key] = ToTableValue(header.Value, header.Key, 0); } } @@ -275,4 +268,173 @@ public static class RabbitMQDispatchContextExtensions return headers; } + + /// + /// The greatest depth to which a header value is mapped. A value nested deeper, or one that + /// contains itself, is rejected. + /// + private const int MaxTableDepth = 64; + + /// + /// Maps a header value onto the CLR types an AMQP field table accepts, at every level of a nested + /// value. The inverse of the mapping the receive side applies. + /// + private static object? ToTableValue(object? value, string key, int depth) + { + if (depth > MaxTableDepth) + { + throw new InvalidOperationException( + $"The header '{key}' is nested more than {MaxTableDepth} levels deep, or holds a " + + "value that contains itself, and cannot be written as an AMQP field table."); + } + + switch (value) + { + // must precede the collection cases + case string: + return value; + + case DateTimeOffset dateTimeOffset: + return new AmqpTimestamp(dateTimeOffset.ToUnixTimeSeconds()); + + case DateTime dateTime: + return ToTimestamp(dateTime); + + // a field table has no unsigned 64 bit type + case ulong unsigned when unsigned > long.MaxValue: + return unsigned.ToString(CultureInfo.InvariantCulture); + + case ulong unsigned: + return (long)unsigned; + + // a field table decimal holds a 32 bit mantissa + case decimal number when !FitsTableDecimal(number): + return number.ToString(CultureInfo.InvariantCulture); + + // the text forms every transport shares + case char character: + return character.ToString(); + + case Guid guid: + return guid.ToString(); + + case Uri uri: + return HeaderValueText.From(uri); + + case TimeSpan timeSpan: + return HeaderValueText.From(timeSpan); + + case DateOnly date: + return HeaderValueText.From(date); + + case TimeOnly time: + return HeaderValueText.From(time); + + case Enum enumeration: + return HeaderValueText.From(enumeration); + + // the table's own shapes rather than JSON text + case JsonElement element: + return ToTableValue(HeadersJsonConverter.ReadHeaderValue(element), key, depth); + + case JsonDocument document: + return ToTableValue(HeadersJsonConverter.ReadHeaderValue(document.RootElement), key, depth); + + case JsonNode node: + using (var parsed = JsonDocument.Parse(node.ToJsonString())) + { + return ToTableValue(HeadersJsonConverter.ReadHeaderValue(parsed.RootElement), key, depth); + } + + // the field table's byte array type, not a long string + case byte[] bytes: + return new BinaryTableValue(bytes); + + // an empty segment has no array to copy, and a whole one needs no copy + case ArraySegment { Array: null }: + return new BinaryTableValue([]); + + case ArraySegment { Offset: 0 } whole when whole.Count == whole.Array!.Length: + return new BinaryTableValue(whole.Array); + + case ArraySegment segment: + return new BinaryTableValue(segment.ToArray()); + + case ReadOnlyMemory memory: + return new BinaryTableValue(memory.ToArray()); + + case Memory writableMemory: + return new BinaryTableValue(writableMemory.ToArray()); + + case IReadOnlyHeaders nested: + var mappedHeaders = new Dictionary(); + foreach (var header in nested) + { + mappedHeaders[header.Key] = ToTableValue(header.Value, key, depth + 1); + } + + return mappedHeaders; + + case IDictionary table: + return ToTable(table, key, depth); + + case IEnumerable sequence: + var mappedSequence = sequence is ICollection collection + ? new List(collection.Count) + : []; + foreach (var item in sequence) + { + mappedSequence.Add(ToTableValue(item, key, depth + 1)); + } + + return mappedSequence; + + default: + return value; + } + } + + private static Dictionary ToTable(IDictionary table, string key, int depth) + { + var mapped = new Dictionary(table.Count); + + foreach (DictionaryEntry entry in table) + { + // a field table names its entries with text; the client stringifies any other key + if (entry.Key is not { } entryKey) + { + continue; + } + + var name = entryKey as string ?? entryKey.ToString(); + + if (name is null) + { + continue; + } + + mapped[name] = ToTableValue(entry.Value, key, depth + 1); + } + + return mapped; + } + + /// + /// Converts a to the instant a field table carries. A local time is + /// converted from this host; a time with no zone is read as UTC. + /// + private static AmqpTimestamp ToTimestamp(DateTime value) + { + var utc = value.Kind == DateTimeKind.Local ? value.ToUniversalTime() : value; + + return new AmqpTimestamp(new DateTimeOffset(utc, TimeSpan.Zero).ToUnixTimeSeconds()); + } + + private static bool FitsTableDecimal(decimal value) + { + Span bits = stackalloc int[4]; + decimal.GetBits(value, bits); + + return bits[1] == 0 && bits[2] == 0 && (uint)bits[0] <= int.MaxValue; + } } diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs index 9429d3b1b7b..93e9485c136 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs @@ -1,4 +1,5 @@ using System.Text; +using System.Text.Unicode; using Mocha.Middlewares; using RabbitMQ.Client; using RabbitMQ.Client.Events; @@ -11,6 +12,13 @@ namespace Mocha.Transport.RabbitMQ; /// internal sealed class RabbitMQMessageEnvelopeParser { + /// + /// The range a can express, which bounds the timestamps mapped to one. + /// + private static readonly long s_minTimestampSeconds = DateTimeOffset.MinValue.ToUnixTimeSeconds(); + + private static readonly long s_maxTimestampSeconds = DateTimeOffset.MaxValue.ToUnixTimeSeconds(); + /// /// Converts a RabbitMQ delivery into a by mapping AMQP basic properties /// and custom headers to envelope fields. @@ -86,19 +94,58 @@ private static Headers BuildHeaders(IDictionary? headers) var result = new Headers(headers.Count); foreach (var (key, value) in headers) { - var strValue = value switch - { - byte[] bytes => Encoding.UTF8.GetString(bytes), - AmqpTimestamp timestamp => DateTimeOffset.FromUnixTimeSeconds(timestamp.UnixTime), - _ => value - }; - - result.Set(key, strValue); + result.Set(key, NormalizeValue(value)); } return result; } + /// + /// Maps AMQP wire types onto the CLR types the envelope serializer understands, at every level of + /// a nested value. + /// + private static object? NormalizeValue(object? value) + { + switch (value) + { + // a binary field is kept as bytes without being tested for text + case BinaryTableValue binary: + return binary.Bytes; + + // a long string carries text and binary alike, so content is the only signal + case byte[] bytes: + return Utf8.IsValid(bytes) ? Encoding.UTF8.GetString(bytes) : bytes; + + // a value outside the range a date can express is kept as the number it is + case AmqpTimestamp timestamp: + return timestamp.UnixTime >= s_minTimestampSeconds + && timestamp.UnixTime <= s_maxTimestampSeconds + ? DateTimeOffset.FromUnixTimeSeconds(timestamp.UnixTime) + : timestamp.UnixTime; + + case IDictionary table: + var mappedTable = new Dictionary(table.Count); + foreach (var (key, item) in table) + { + mappedTable[key] = NormalizeValue(item); + } + + return mappedTable; + + case IList array: + var mappedArray = new object?[array.Count]; + for (var i = 0; i < array.Count; i++) + { + mappedArray[i] = NormalizeValue(array[i]); + } + + return mappedArray; + + default: + return value; + } + } + /// /// Shared singleton instance of the parser. /// diff --git a/src/Mocha/src/Mocha/Assembly.cs b/src/Mocha/src/Mocha/Assembly.cs index 065271cac03..8d05c18ac8b 100644 --- a/src/Mocha/src/Mocha/Assembly.cs +++ b/src/Mocha/src/Mocha/Assembly.cs @@ -2,6 +2,7 @@ [assembly: InternalsVisibleTo("Mocha.Tests")] [assembly: InternalsVisibleTo("Mocha.Sagas")] +[assembly: InternalsVisibleTo("Mocha.TestHelpers")] [assembly: InternalsVisibleTo("Mocha.Sagas.TestHelpers")] [assembly: InternalsVisibleTo("Mocha.Sagas.Tests")] [assembly: InternalsVisibleTo("Mocha.Inbox")] diff --git a/src/Mocha/src/Mocha/Serialization/HeaderValueText.cs b/src/Mocha/src/Mocha/Serialization/HeaderValueText.cs new file mode 100644 index 00000000000..d51a211b4a2 --- /dev/null +++ b/src/Mocha/src/Mocha/Serialization/HeaderValueText.cs @@ -0,0 +1,27 @@ +using System.Globalization; + +namespace Mocha; + +/// +/// The text form a header value takes on the wire, shared by every transport that carries one as text. +/// +internal static class HeaderValueText +{ + /// + /// The string the URI was built from, escapes intact. + /// + public static string From(Uri value) + => value.OriginalString; + + public static string From(TimeSpan value) + => value.ToString("c", CultureInfo.InvariantCulture); + + public static string From(DateOnly value) + => value.ToString("O", CultureInfo.InvariantCulture); + + public static string From(TimeOnly value) + => value.ToString("O", CultureInfo.InvariantCulture); + + public static string From(Enum value) + => value.ToString(); +} diff --git a/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs b/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs index 03e7ec4e63f..ec94e295fe6 100644 --- a/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs +++ b/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs @@ -1,12 +1,44 @@ -using System.Globalization; +using System.Collections; using System.Text.Json; +using System.Text.Json.Nodes; using System.Text.Json.Serialization; +using System.Text.Json.Serialization.Metadata; namespace Mocha; /// /// JSON converter for serializing and deserializing instances as JSON objects with dynamic value types. /// +/// +/// +/// Header values are weakly typed, but the set of types that can be written is closed: +/// , , , +/// , the numeric types, , +/// , , , +/// , , , an enum, a byte payload +/// (written as base64), a JSON value (, , +/// ), nested headers, a dictionary with string keys, or a sequence of any of +/// these. Through , any other type is rejected with a +/// naming the header; through options carrying a reflection-based +/// resolver, that resolver serializes the value instead. +/// +/// +/// Reading produces a narrower set, because JSON carries no type tag: +/// , , , , +/// , , , a dictionary, or an array, so a +/// value keeps its JSON representation across a round trip but not always its CLR type. A +/// , , , or an +/// enum reads back as the string written, a byte payload as base64 text, a sequence as an array. +/// +/// +/// A number is read by value, not by the type that wrote it, taking the narrowest of +/// , and that holds it: a +/// , or a whole , or +/// , can come back as an . A fractional value reads as a +/// , losing precision, and a whole number beyond +/// reads as a written again in scientific notation. +/// +/// public class HeadersJsonConverter : JsonConverter { /// @@ -17,7 +49,27 @@ public class HeadersJsonConverter : JsonConverter /// /// Gets pre-configured with this converter registered. /// - public static readonly JsonSerializerOptions Options = new() { Converters = { Instance } }; + /// + /// These options resolve metadata for and only, and + /// are read-only. + /// + public static readonly JsonSerializerOptions Options = CreateOptions(); + + private static JsonSerializerOptions CreateOptions() + { + var options = new JsonSerializerOptions + { + Converters = { Instance }, + + // left unset, the serializer installs the reflection-based resolver on first use + TypeInfoResolver = new HeadersTypeInfoResolver() + }; + + // the writer and reader use the converter directly, so nothing else seals these + options.MakeReadOnly(); + + return options; + } /// public override Headers? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) @@ -64,12 +116,70 @@ public override void Write(Utf8JsonWriter writer, IHeaders value, JsonSerializer foreach (var header in value) { writer.WritePropertyName(header.Key); - WriteValue(writer, header.Value, options); + WriteValue(writer, header.Value, options, header.Key); } writer.WriteEndObject(); } + /// + /// Reads a JSON value into the types a header holds, the reading applies to a + /// document. + /// + /// The JSON value to read. + /// The value mapped onto the types a header can hold. + internal static object? ReadHeaderValue(JsonElement element) + { + switch (element.ValueKind) + { + case JsonValueKind.Object: + var table = new Dictionary(); + foreach (var property in element.EnumerateObject()) + { + table[property.Name] = ReadHeaderValue(property.Value); + } + + return table; + + case JsonValueKind.Array: + var items = new List(); + foreach (var item in element.EnumerateArray()) + { + items.Add(ReadHeaderValue(item)); + } + + return items; + + case JsonValueKind.String: + return element.GetString(); + + case JsonValueKind.Number: + if (element.TryGetInt32(out var intValue)) + { + return intValue; + } + if (element.TryGetInt64(out var longValue)) + { + return longValue; + } + if (element.TryGetUInt64(out var unsignedValue)) + { + return unsignedValue; + } + + return element.GetDouble(); + + case JsonValueKind.True: + return true; + + case JsonValueKind.False: + return false; + + default: + return null; + } + } + private static object? ReadValue(ref Utf8JsonReader reader, JsonSerializerOptions options) { switch (reader.TokenType) @@ -83,11 +193,8 @@ public override void Write(Utf8JsonWriter writer, IHeaders value, JsonSerializer case JsonTokenType.False: return false; + // JSON carries no type tag, so text stays text case JsonTokenType.String: - if (reader.TryGetDateTime(out var dateTime)) - { - return dateTime; - } return reader.GetString(); case JsonTokenType.Number: @@ -99,6 +206,10 @@ public override void Write(Utf8JsonWriter writer, IHeaders value, JsonSerializer { return longValue; } + if (reader.TryGetUInt64(out var ulongValue)) + { + return ulongValue; + } if (reader.TryGetDouble(out var doubleValue)) { return doubleValue; @@ -158,7 +269,7 @@ public override void Write(Utf8JsonWriter writer, IHeaders value, JsonSerializer throw new JsonException("Unexpected end of JSON in array"); } - private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSerializerOptions options) + private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSerializerOptions options, string key) { switch (value) { @@ -174,6 +285,22 @@ private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSeriali writer.WriteStringValue(stringValue); break; + case byte[] bytesValue: + writer.WriteBase64StringValue(bytesValue); + break; + + case ArraySegment bytesSegment: + writer.WriteBase64StringValue(bytesSegment); + break; + + case ReadOnlyMemory bytesMemory: + writer.WriteBase64StringValue(bytesMemory.Span); + break; + + case Memory writableBytesMemory: + writer.WriteBase64StringValue(writableBytesMemory.Span); + break; + case int intValue: writer.WriteNumberValue(intValue); break; @@ -210,57 +337,55 @@ private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSeriali jsonDocument.RootElement.WriteTo(writer); break; + // must precede the collection cases + case JsonNode jsonNode: + jsonNode.WriteTo(writer); + break; + case IDictionary dictionary: writer.WriteStartObject(); foreach (var kvp in dictionary) { writer.WritePropertyName(kvp.Key); - WriteValue(writer, kvp.Value, options); + WriteValue(writer, kvp.Value, options, key); } writer.WriteEndObject(); break; - case IEnumerable enumerable: - writer.WriteStartArray(); - foreach (var item in enumerable) - { - WriteValue(writer, item, options); - } - writer.WriteEndArray(); - break; - case IReadOnlyHeaders headers: writer.WriteStartObject(); foreach (var header in headers) { writer.WritePropertyName(header.Key); - WriteValue(writer, header.Value, options); + WriteValue(writer, header.Value, options, key); } writer.WriteEndObject(); break; + // the writer's own overload formats into the buffer instead of allocating a string case Guid g: writer.WriteStringValue(g); break; + // the text forms every transport shares case TimeSpan t: - writer.WriteStringValue(t.ToString("c", CultureInfo.InvariantCulture)); + writer.WriteStringValue(HeaderValueText.From(t)); break; case Uri u: - writer.WriteStringValue(u.ToString()); + writer.WriteStringValue(HeaderValueText.From(u)); break; case DateOnly d: - writer.WriteStringValue(d.ToString("O", CultureInfo.InvariantCulture)); + writer.WriteStringValue(HeaderValueText.From(d)); break; case TimeOnly to: - writer.WriteStringValue(to.ToString("O", CultureInfo.InvariantCulture)); + writer.WriteStringValue(HeaderValueText.From(to)); break; case Enum e: - writer.WriteStringValue(e.ToString()); + writer.WriteStringValue(HeaderValueText.From(e)); break; case short s: @@ -291,9 +416,92 @@ private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSeriali writer.WriteStringValue([c]); break; + // int[] and Dictionary have no covariant conversion to object? items + case IDictionary dictionary: + WriteDictionary(writer, dictionary, options, key); + break; + + case IEnumerable enumerable: + writer.WriteStartArray(); + foreach (var item in enumerable) + { + WriteValue(writer, item, options, key); + } + writer.WriteEndArray(); + break; + default: - JsonSerializer.Serialize(writer, value, options.GetTypeInfo(value.GetType())); + WriteUnknownValue(writer, value, options, key); break; } } + + private static void WriteDictionary( + Utf8JsonWriter writer, + IDictionary dictionary, + JsonSerializerOptions options, + string key) + { + writer.WriteStartObject(); + + foreach (DictionaryEntry entry in dictionary) + { + // a custom dictionary can hand back any key, a null included + if (entry.Key is not string name) + { + var keyType = entry.Key is { } entryKey ? entryKey.GetType().FullName : "null"; + + throw new JsonException( + $"The header '{key}' holds a dictionary keyed by '{keyType}' that cannot be " + + "written as JSON. Use string keys."); + } + + writer.WritePropertyName(name); + WriteValue(writer, entry.Value, options, key); + } + + writer.WriteEndObject(); + } + + private static void WriteUnknownValue( + Utf8JsonWriter writer, + object value, + JsonSerializerOptions options, + string key) + { + try + { + JsonSerializer.Serialize(writer, value, options.GetTypeInfo(value.GetType())); + } + catch (NotSupportedException ex) + { + throw new JsonException( + $"The header '{key}' holds a value of type '{value.GetType().FullName}' that cannot be " + + "written as JSON. Set the header to a supported value type, or register JSON type " + + "metadata for it.", + ex); + } + } + + /// + /// Resolves JSON type metadata for headers, and for nothing else. + /// + private sealed class HeadersTypeInfoResolver : IJsonTypeInfoResolver + { + public JsonTypeInfo? GetTypeInfo(Type type, JsonSerializerOptions options) + { + if (type == typeof(IHeaders)) + { + return JsonMetadataServices.CreateValueInfo(options, Instance); + } + + // the concrete type Read returns + if (type == typeof(Headers)) + { + return JsonMetadataServices.CreateValueInfo(options, Instance); + } + + return null; + } + } } diff --git a/src/Mocha/test/Mocha.TestHelpers/FaultHeaders.cs b/src/Mocha/test/Mocha.TestHelpers/FaultHeaders.cs new file mode 100644 index 00000000000..4f7784d5d3a --- /dev/null +++ b/src/Mocha/test/Mocha.TestHelpers/FaultHeaders.cs @@ -0,0 +1,33 @@ +namespace Mocha.TestHelpers; + +/// +/// The headers a faulted message carries, named by the constants the fault middleware writes. +/// +public static class FaultHeaders +{ + public static string ExceptionType => MessageHeaders.Fault.ExceptionType.Key; + + public static string Message => MessageHeaders.Fault.Message.Key; + + public static string StackTrace => MessageHeaders.Fault.StackTrace.Key; + + public static string Timestamp => MessageHeaders.Fault.Timestamp.Key; + + private static string[] Keys => [ExceptionType, Message, StackTrace, Timestamp]; + + /// + /// The CLR type each fault header value must have. + /// + public static Dictionary ExpectedShape() + => Keys.ToDictionary(key => key, _ => nameof(String)); + + /// + /// Maps the fault headers of a received message onto each value's CLR type, naming any absent. + /// + public static Dictionary Shape(IReadOnlyDictionary headers) + => Keys.ToDictionary( + key => key, + key => headers.TryGetValue(key, out var value) + ? value?.GetType().Name ?? "" + : ""); +} diff --git a/src/Mocha/test/Mocha.Tests/Headers/HeadersSerializationTests.cs b/src/Mocha/test/Mocha.Tests/Headers/HeadersSerializationTests.cs index 20eb7a950cd..9a686eaaf8a 100644 --- a/src/Mocha/test/Mocha.Tests/Headers/HeadersSerializationTests.cs +++ b/src/Mocha/test/Mocha.Tests/Headers/HeadersSerializationTests.cs @@ -41,26 +41,19 @@ public void RoundTrip_Should_PreserveNonNullValue_When_LossyTypeSerialized(strin } [Fact] - public void RoundTrip_Should_PreserveDateTime_When_Serialized() + public void RoundTrip_Should_PreserveDateTimeAsIsoText_When_Serialized() { // arrange var headers = new Headers(); - var dateTime = new DateTime(2024, 1, 15, 10, 30, 45, DateTimeKind.Utc); - headers.Set("dateTime", dateTime); + headers.Set("dateTime", new DateTime(2024, 1, 15, 10, 30, 45, DateTimeKind.Utc)); // act var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); var result = JsonSerializer.Deserialize(json, HeadersJsonConverter.Options); // assert - Assert.NotNull(result); - Assert.True(result.TryGetValue("dateTime", out var value)); - Assert.IsType(value); - var resultDateTime = (DateTime)value; - // DateTime may be deserialized with some precision loss - Assert.Equal(dateTime.Year, resultDateTime.Year); - Assert.Equal(dateTime.Month, resultDateTime.Month); - Assert.Equal(dateTime.Day, resultDateTime.Day); + result!.TryGetValue("dateTime", out var value); + Assert.Equal("2024-01-15T10:30:45Z", Assert.IsType(value)); } [Fact] diff --git a/src/Mocha/test/Mocha.Tests/Serialization/HeadersJsonConverterTests.cs b/src/Mocha/test/Mocha.Tests/Serialization/HeadersJsonConverterTests.cs index 233950a516f..df42f315ec9 100644 --- a/src/Mocha/test/Mocha.Tests/Serialization/HeadersJsonConverterTests.cs +++ b/src/Mocha/test/Mocha.Tests/Serialization/HeadersJsonConverterTests.cs @@ -1,5 +1,7 @@ using System.Globalization; using System.Text.Json; +using System.Text.Json.Nodes; +using System.Text.Json.Serialization.Metadata; namespace Mocha.Tests; @@ -196,17 +198,284 @@ public void Write_Should_EmitStringValue_When_HeaderIsChar() } [Fact] - public void Write_Should_SerializeCustomType_When_HeaderIsCustomType() + public void Write_Should_KeepEscapes_When_HeaderIsUri() { // arrange var headers = new Headers(); + headers.Set("callback", new Uri("https://example.com/a%20b")); + + // act + var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); + + // assert + Assert.Equal("""{"callback":"https://example.com/a%20b"}""", json); + } + + [Fact] + public void Write_Should_EmitBase64String_When_HeaderIsByteMemory() + { + // arrange + var headers = new Headers(); + headers.Set("readonly", new ReadOnlyMemory("COM1"u8.ToArray())); + headers.Set("writable", new Memory("COM1"u8.ToArray())); + + // act + var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); + + // assert + Assert.Equal("""{"readonly":"Q09NMQ==","writable":"Q09NMQ=="}""", json); + } + + [Fact] + public void Write_Should_EmitBase64String_When_HeaderIsByteArray() + { + // arrange + var headers = new Headers(); + headers.Set("signature", "signature-123"u8.ToArray()); + + // act + var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); + + // assert + Assert.Equal("""{"signature":"c2lnbmF0dXJlLTEyMw=="}""", json); + } + + [Fact] + public void Write_Should_EmitBase64Strings_When_ByteArraysAreNestedInHeader() + { + // arrange + // the shape a broker-generated x-death header has: an array of tables + var headers = new Headers(); + headers.Set( + "x-death", + new List + { + new Dictionary + { + ["reason"] = "rejected"u8.ToArray(), + ["count"] = 1L + } + }); + + // act + var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); + + // assert + Assert.Equal("""{"x-death":[{"reason":"cmVqZWN0ZWQ=","count":1}]}""", json); + } + + [Fact] + public void Write_Should_ThrowJsonExceptionNamingTheHeader_When_ValueTypeIsUnsupported() + { + // arrange + var headers = new Headers(); + headers.Set("handle", nint.Zero); + + // act + var exception = Record.Exception( + () => JsonSerializer.Serialize(headers, HeadersJsonConverter.Options)); + + // assert + Assert.Equal( + "The header 'handle' holds a value of type 'System.IntPtr' that cannot be written as " + + "JSON. Set the header to a supported value type, or register JSON type metadata for it.", + Assert.IsType(exception).Message); + } + + [Fact] + public void Write_Should_ThrowJsonExceptionNamingTheHeader_When_HeaderIsCustomType() + { + // arrange + // a custom type has no JSON metadata, and none is inferred through reflection + var headers = new Headers(); headers.Set("custom", new CustomHeaderDto { Name = "test" }); + // act + var exception = Record.Exception( + () => JsonSerializer.Serialize(headers, HeadersJsonConverter.Options)); + + // assert + Assert.Equal( + "The header 'custom' holds a value of type 'Mocha.Tests.HeadersJsonConverterTests+" + + "CustomHeaderDto' that cannot be written as JSON. Set the header to a supported value " + + "type, or register JSON type metadata for it.", + Assert.IsType(exception).Message); + } + + [Fact] + public void Write_Should_EmitArray_When_HeaderIsValueTypeSequence() + { + // arrange + // a sequence of value types has no covariant conversion to IEnumerable + var headers = new Headers(); + headers.Set("attempts", new[] { 1, 2, 3 }); + headers.Set("offsets", new List { 10L }); + + // act + var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); + + // assert + Assert.Equal("""{"attempts":[1,2,3],"offsets":[10]}""", json); + } + + [Fact] + public void Write_Should_EmitObject_When_HeaderIsValueTypeDictionary() + { + // arrange + var headers = new Headers(); + headers.Set("tags", new Dictionary { ["region"] = "eu" }); + headers.Set("counts", new SortedDictionary { ["retries"] = 2 }); + + // act + var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); + + // assert + Assert.Equal("""{"tags":{"region":"eu"},"counts":{"retries":2}}""", json); + } + + [Fact] + public void Write_Should_ThrowJsonExceptionNamingTheHeader_When_DictionaryKeyIsNotString() + { + // arrange + var headers = new Headers(); + headers.Set("weights", new Dictionary { [1] = "high" }); + + // act + var exception = Record.Exception( + () => JsonSerializer.Serialize(headers, HeadersJsonConverter.Options)); + + // assert + Assert.Equal( + "The header 'weights' holds a dictionary keyed by 'System.Int32' that cannot be written " + + "as JSON. Use string keys.", + Assert.IsType(exception).Message); + } + + [Fact] + public void Write_Should_EmitJsonVerbatim_When_HeaderIsJsonNode() + { + // arrange + var headers = new Headers(); + headers.Set("trace", new JsonObject { ["id"] = "abc", ["depth"] = 3 }); + headers.Set("hops", new JsonArray("a", "b")); + // act var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); // assert - Assert.Equal("""{"custom":{"Name":"test"}}""", json); + Assert.Equal("""{"trace":{"id":"abc","depth":3},"hops":["a","b"]}""", json); + } + + [Fact] + public void Write_Should_EmitBase64String_When_HeaderIsByteSegment() + { + // arrange + // a byte payload is base64 regardless of the container it arrives in + var headers = new Headers(); + headers.Set("signature", new ArraySegment("signature-123"u8.ToArray())); + + // act + var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); + + // assert + Assert.Equal("""{"signature":"c2lnbmF0dXJlLTEyMw=="}""", json); + } + + [Fact] + public void Read_Should_KeepText_When_ValueLooksLikeATimestamp() + { + // arrange + const string json = """{"fault-timestamp":"2026-08-03T10:00:00.0000000+02:00"}"""; + + // act + var headers = JsonSerializer.Deserialize(json, HeadersJsonConverter.Options); + + // assert + headers!.TryGetValue("fault-timestamp", out var value); + Assert.Equal("2026-08-03T10:00:00.0000000+02:00", Assert.IsType(value)); + } + + [Fact] + public void Read_Should_ReturnUnsignedLong_When_NumberExceedsSignedRange() + { + // arrange + const string json = """{"offset":18446744073709551615}"""; + + // act + var headers = JsonSerializer.Deserialize(json, HeadersJsonConverter.Options); + + // assert + headers!.TryGetValue("offset", out var value); + Assert.Equal(ulong.MaxValue, Assert.IsType(value)); + } + + [Fact] + public void ReadWrite_Should_PreserveJson_When_ValuesAreRoundTripped() + { + // arrange + // every canonical read type, including the two that used to be rewritten + const string json = """ + {"text":"2026-08-03T10:00:00.0000000Z","offset":18446744073709551615,"count":1,"ticks":9000000000,"ratio":2.5,"flag":true,"empty":null,"table":{"k":"v"},"list":[1,"a"]} + """; + + // act + var first = JsonSerializer.Serialize( + JsonSerializer.Deserialize(json, HeadersJsonConverter.Options), + HeadersJsonConverter.Options); + var second = JsonSerializer.Serialize( + JsonSerializer.Deserialize(first, HeadersJsonConverter.Options), + HeadersJsonConverter.Options); + + // assert + Assert.Equal(json, first); + Assert.Equal(first, second); + } + + [Fact] + public void Options_Should_RoundTrip_When_TheConcreteTypeIsUsed() + { + // arrange + var headers = new Headers(); + headers.Set("trace-id", "abc-123"); + + // act + var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); + var result = JsonSerializer.Deserialize(json, HeadersJsonConverter.Options); + + // assert + result!.TryGetValue("trace-id", out var value); + Assert.Equal("""{"trace-id":"abc-123"}""", json); + Assert.Equal("abc-123", value); + } + + [Fact] + public void Options_Should_BeSealed_When_FirstObserved() + { + // arrange + var options = HeadersJsonConverter.Options; + + // act + var exception = Record.Exception(() => options.TypeInfoResolver = new DefaultJsonTypeInfoResolver()); + + // assert + Assert.True(options.IsReadOnly); + Assert.IsType(exception); + } + + [Fact] + public void Options_Should_ResolveOnlyHeaders_When_MetadataIsRequested() + { + // arrange + // the resolver covers headers only, so nothing pulls in the reflection-based resolver + var resolver = HeadersJsonConverter.Options.TypeInfoResolver!; + + // act + var headers = resolver.GetTypeInfo(typeof(IHeaders), HeadersJsonConverter.Options); + var unrelated = resolver.GetTypeInfo(typeof(CustomHeaderDto), HeadersJsonConverter.Options); + + // assert + Assert.Equal(typeof(IHeaders), headers!.Type); + Assert.Null(unrelated); } private enum TestPriority diff --git a/src/Mocha/test/Mocha.Tests/Transport/MessageEnvelopeWriterTests.cs b/src/Mocha/test/Mocha.Tests/Transport/MessageEnvelopeWriterTests.cs index c854ffe59d9..27212086fda 100644 --- a/src/Mocha/test/Mocha.Tests/Transport/MessageEnvelopeWriterTests.cs +++ b/src/Mocha/test/Mocha.Tests/Transport/MessageEnvelopeWriterTests.cs @@ -1,6 +1,7 @@ using System.Collections.Immutable; using System.Text; using System.Text.Json; +using CookieCrumble; using Mocha.Middlewares; namespace Mocha.Tests; @@ -168,6 +169,60 @@ public void WriteMessage_Should_SerializeEmptyHeaders_When_HeadersCollectionIsEm Assert.Empty(headers.EnumerateObject()); } + [Fact] + public void WriteMessage_Should_WriteBase64_When_HeaderIsByteArray() + { + // arrange + var envelope = new MessageEnvelope { MessageId = "msg-008", Headers = new Headers() }; + envelope.Headers.Set("tenant_id", "tenant-123"u8.ToArray()); + + using var stream = new MemoryStream(); + using var writer = new Utf8JsonWriter(stream); + + // act + var envelopeWriter = new MessageEnvelopeWriter(writer); + envelopeWriter.WriteMessage(envelope); + writer.Flush(); + + // assert + Encoding.UTF8.GetString(stream.ToArray()) + .MatchInlineSnapshot("""{"messageId":"msg-008","headers":{"tenant_id":"dGVuYW50LTEyMw=="}}"""); + } + + [Fact] + public void WriteMessage_Should_WriteNestedValues_When_HeaderIsBrokerDeathTable() + { + // arrange + // the shape a broker-generated x-death header has once it reaches the envelope + var envelope = new MessageEnvelope { MessageId = "msg-009", Headers = new Headers() }; + envelope.Headers.Set( + "x-death", + new List + { + new Dictionary + { + ["queue"] = "product-queue", + ["reason"] = "rejected"u8.ToArray(), + ["count"] = 1L + } + }); + + using var stream = new MemoryStream(); + using var writer = new Utf8JsonWriter(stream); + + // act + var envelopeWriter = new MessageEnvelopeWriter(writer); + envelopeWriter.WriteMessage(envelope); + writer.Flush(); + + // assert + Encoding.UTF8.GetString(stream.ToArray()) + .MatchInlineSnapshot( + """ + {"messageId":"msg-009","headers":{"x-death":[{"queue":"product-queue","reason":"cmVqZWN0ZWQ=","count":1}]}} + """); + } + [Fact] public void WriteMessage_Should_SerializeBodyAsRawJson_When_BodyIsJson() { @@ -501,6 +556,28 @@ public void Roundtrip_Should_PreserveData_When_WritingAndReadingMinimalEnvelope( Assert.Null(result.ResponseAddress); } + [Fact] + public void Roundtrip_Should_PreserveHeaderText_When_HeaderHoldsATimestamp() + { + // arrange + const string timestamp = "2026-08-03T10:00:00.0000000+02:00"; + var original = new MessageEnvelope { MessageId = "msg-018", Headers = new Headers() }; + original.Headers.Set("fault-timestamp", timestamp); + + // act - write + using var stream = new MemoryStream(); + using var writer = new Utf8JsonWriter(stream); + new MessageEnvelopeWriter(writer).WriteMessage(original); + writer.Flush(); + + // act - read + var result = MessageEnvelopeReader.Parse(stream.ToArray()); + + // assert + result.Headers!.TryGetValue("fault-timestamp", out var value); + Assert.Equal(timestamp, Assert.IsType(value)); + } + [Fact] public void Roundtrip_Should_PreserveEmptyArray_When_EnclosedMessageTypesIsEmpty() { diff --git a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Behaviors/ErrorQueueTests.cs b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Behaviors/ErrorQueueTests.cs index 5474e080049..ec4c456a094 100644 --- a/src/Mocha/test/Mocha.Transport.InMemory.Tests/Behaviors/ErrorQueueTests.cs +++ b/src/Mocha/test/Mocha.Transport.InMemory.Tests/Behaviors/ErrorQueueTests.cs @@ -39,12 +39,9 @@ public async Task PublishAsync_Should_RouteToErrorQueue_When_HandlerThrows() // assert - faulted message lands in error queue with fault headers Assert.True(await capture.WaitAsync(s_timeout), "Error consumer did not receive the faulted message"); var headers = Assert.Single(capture.CapturedHeaders); - Assert.True(headers.TryGetValue("fault-exception-type", out var exType)); - Assert.Contains("InvalidOperationException", (string?)exType); - Assert.True(headers.TryGetValue("fault-message", out var faultMsg)); - Assert.Equal("Handler failed deliberately", (string?)faultMsg); - Assert.True(headers.ContainsKey("fault-stack-trace")); - Assert.True(headers.ContainsKey("fault-timestamp")); + Assert.Equal(FaultHeaders.ExpectedShape(), FaultHeaders.Shape(headers)); + Assert.Contains("InvalidOperationException", (string?)headers[FaultHeaders.ExceptionType]); + Assert.Equal("Handler failed deliberately", headers[FaultHeaders.Message]); } [Fact] @@ -80,7 +77,7 @@ public async Task SendAsync_Should_RouteToErrorQueue_When_HandlerThrows() // assert - faulted message lands in error queue Assert.True(await capture.WaitAsync(s_timeout), "Error consumer did not receive the faulted message"); var headers = Assert.Single(capture.CapturedHeaders); - Assert.True(headers.TryGetValue("fault-exception-type", out var exType)); + Assert.True(headers.TryGetValue(FaultHeaders.ExceptionType, out var exType)); Assert.Contains("InvalidOperationException", (string?)exType); } diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ErrorQueueTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ErrorQueueTests.cs index c99d0cd10f2..7a2901d9ecc 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ErrorQueueTests.cs +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Behaviors/ErrorQueueTests.cs @@ -48,10 +48,7 @@ public async Task PublishAsync_Should_RouteToErrorQueue_When_HandlerThrows() Assert.True(await capture.WaitAsync(s_timeout), "Error queue consumer did not receive the faulted message"); var headers = Assert.Single(capture.CapturedHeaders); - Assert.True(headers.ContainsKey("fault-exception-type"), "Missing fault-exception-type header"); - Assert.True(headers.ContainsKey("fault-message"), "Missing fault-message header"); - Assert.True(headers.ContainsKey("fault-stack-trace"), "Missing fault-stack-trace header"); - Assert.True(headers.ContainsKey("fault-timestamp"), "Missing fault-timestamp header"); + Assert.Equal(FaultHeaders.ExpectedShape(), FaultHeaders.Shape(headers)); } [Fact] diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/HeaderValueConsistencyTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/HeaderValueConsistencyTests.cs new file mode 100644 index 00000000000..bb4fad92a27 --- /dev/null +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/HeaderValueConsistencyTests.cs @@ -0,0 +1,113 @@ +using System.Text.Json; +using System.Text.Json.Nodes; +using Mocha.Middlewares; +using RabbitMQ.Client; +using RabbitMQ.Client.Events; + +namespace Mocha.Transport.RabbitMQ.Tests; + +/// +/// Asserts that the envelope serializer and the RabbitMQ dispatcher agree on what a header value +/// becomes. +/// +public class HeaderValueConsistencyTests +{ + public static TheoryData TextValues() + => new() + { + { "guid", Guid.Parse("12345678-1234-1234-1234-123456789012") }, + { "char", 'c' }, + { "uri", new Uri("https://example.com/a%20b") }, + { "timespan", TimeSpan.FromMinutes(90) }, + { "dateonly", new DateOnly(2026, 8, 3) }, + { "timeonly", new TimeOnly(10, 30, 15) }, + { "enum", DayOfWeek.Wednesday } + }; + + [Theory] + [MemberData(nameof(TextValues))] + public void BothEncoders_Should_WriteTheSameText_When_ValueTravelsAsText(string key, object value) + { + // arrange + var headers = new Headers([new HeaderValue { Key = key, Value = value }]); + var envelope = new MessageEnvelope { Headers = headers }; + + // act + var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); + var table = envelope.BuildHeaders(); + + // assert + Assert.Equal( + JsonSerializer.Serialize(new Dictionary { [key] = table[key] }), + json); + } + + [Theory] + [MemberData(nameof(ShapedValues))] + public void BothEncoders_Should_ProduceTheSameReading_When_ValueIsAContainer(string key, object value) + { + // arrange + var headers = new Headers([new HeaderValue { Key = key, Value = value }]); + var envelope = new MessageEnvelope { Headers = headers }; + + // act + var throughJson = JsonSerializer.Deserialize( + JsonSerializer.Serialize(headers, HeadersJsonConverter.Options), + HeadersJsonConverter.Options); + var throughTable = RabbitMQMessageEnvelopeParser.Instance + .Parse(CreateDelivery(envelope.BuildHeaders())); + + // assert + throughJson!.TryGetValue(key, out var fromJson); + throughTable.Headers!.TryGetValue(key, out var fromTable); + Assert.Equal(fromJson, fromTable); + } + + public static TheoryData ShapedValues() + => new() + { + { "dictionary", new Dictionary { ["k"] = "v" } }, + { "sequence", new object?[] { 1, "a" } }, + { "json-object", new JsonObject { ["k"] = 1 } }, + { "json-array", new JsonArray(1, 2) }, + { "json-text", JsonValue.Create("abc")! }, + // cloned so that the element outlives the document it was parsed from + { "json-element", JsonDocument.Parse("""{"k":1}""").RootElement.Clone() }, + { "json-document", JsonDocument.Parse("""[1,2]""") }, + { "nested-headers", new Headers([new HeaderValue { Key = "inner", Value = "v" }]) }, + { "null-inside", new Dictionary { ["k"] = null } } + }; + + [Fact] + public void BothEncoders_Should_KeepTheHeader_When_ItsValueIsNull() + { + // arrange + var headers = new Headers([new HeaderValue { Key = "x-null", Value = null }]); + var envelope = new MessageEnvelope { Headers = headers }; + + // act + var throughJson = JsonSerializer.Deserialize( + JsonSerializer.Serialize(headers, HeadersJsonConverter.Options), + HeadersJsonConverter.Options); + var throughTable = RabbitMQMessageEnvelopeParser.Instance + .Parse(CreateDelivery(envelope.BuildHeaders())); + + // assert + Assert.True(throughJson!.ContainsKey("x-null")); + Assert.True(throughTable.Headers!.ContainsKey("x-null")); + } + + private static BasicDeliverEventArgs CreateDelivery(IDictionary headers) + { + var properties = new BasicProperties { Headers = headers }; + + return new BasicDeliverEventArgs( + consumerTag: "tag", + deliveryTag: 1, + redelivered: false, + exchange: "exchange", + routingKey: "key", + properties: properties, + body: ReadOnlyMemory.Empty); + } +} diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQDispatchContextExtensionsTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQDispatchContextExtensionsTests.cs index 9a23b622598..ec01147c48b 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQDispatchContextExtensionsTests.cs +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQDispatchContextExtensionsTests.cs @@ -1,4 +1,7 @@ +using System.Collections; using System.Collections.Immutable; +using System.Text.Json; +using System.Text.Json.Nodes; using Mocha.Middlewares; using RabbitMQ.Client; @@ -44,7 +47,82 @@ public void BuildHeaders_Should_ConvertDateTime_When_HeaderContainsDateTime() } [Fact] - public void BuildHeaders_Should_SkipNullValues_When_HeaderValueIsNull() + public void BuildHeaders_Should_ConvertNestedTimestamps_When_HeaderIsDeathTable() + { + // arrange + // the x-death shape the receive side produces, whose nested date has no wire form + var envelope = new MessageEnvelope + { + Headers = new Headers([ + new HeaderValue + { + Key = "x-death", + Value = new object?[] + { + new Dictionary + { + ["queue"] = "product-queue", + ["time"] = DateTimeOffset.FromUnixTimeSeconds(1700000000) + } + } + } + ]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal( + new Dictionary + { + ["queue"] = "product-queue", + ["time"] = new AmqpTimestamp(1700000000) + }, + Assert.IsType>( + Assert.Single(Assert.IsType>(headers["x-death"])))); + } + + [Fact] + public void BuildHeaders_Should_ConvertToText_When_HeaderIsUnsignedLong() + { + // arrange + // a field table has no unsigned 64 bit type, and a double would round off digits + var envelope = new MessageEnvelope + { + Headers = new Headers([new HeaderValue { Key = "offset", Value = ulong.MaxValue }]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal("18446744073709551615", headers["offset"]); + } + + [Fact] + public void BuildHeaders_Should_DeclareBinary_When_HeaderIsABytePayload() + { + // arrange + // a long string carries text and binary alike, so sending one leaves the receive side guessing + var envelope = new MessageEnvelope + { + Headers = new Headers([ + new HeaderValue { Key = "x-signature", Value = "COM1"u8.ToArray() }, + new HeaderValue { Key = "x-segment", Value = new ArraySegment("COM2"u8.ToArray()) } + ]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal("COM1"u8.ToArray(), Assert.IsType(headers["x-signature"]).Bytes); + Assert.Equal("COM2"u8.ToArray(), Assert.IsType(headers["x-segment"]).Bytes); + } + + [Fact] + public void BuildHeaders_Should_KeepNullValues_When_HeaderValueIsNull() { // arrange var envelope = new MessageEnvelope @@ -56,7 +134,8 @@ public void BuildHeaders_Should_SkipNullValues_When_HeaderValueIsNull() var headers = envelope.BuildHeaders(); // assert - Assert.False(headers.ContainsKey("x-null")); + Assert.True(headers.ContainsKey("x-null")); + Assert.Null(headers["x-null"]); } [Fact] @@ -184,4 +263,254 @@ public void BuildHeaders_Should_ReturnEmptyHeaders_When_EnvelopeIsEmpty() // assert Assert.Empty(headers); } + + [Fact] + public void BuildHeaders_Should_MapToTableValues_When_TypesHaveNoWireForm() + { + // arrange + // every type a field table cannot represent + using var document = JsonDocument.Parse("""{"k":1}"""); + var envelope = new MessageEnvelope + { + Headers = new Headers( + [ + new HeaderValue { Key = "char", Value = 'c' }, + new HeaderValue { Key = "guid", Value = Guid.Parse("12345678-1234-1234-1234-123456789012") }, + new HeaderValue { Key = "uri", Value = new Uri("rabbitmq:///q/x") }, + new HeaderValue { Key = "timespan", Value = TimeSpan.FromMinutes(90) }, + new HeaderValue { Key = "dateonly", Value = new DateOnly(2026, 8, 3) }, + new HeaderValue { Key = "timeonly", Value = new TimeOnly(10, 30, 15) }, + new HeaderValue { Key = "enum", Value = TestPriority.High }, + new HeaderValue { Key = "element", Value = document.RootElement }, + new HeaderValue { Key = "node", Value = new JsonObject { ["k"] = 1 } }, + new HeaderValue + { + Key = "nested", + Value = new Headers([new HeaderValue { Key = "inner", Value = TimeSpan.Zero }]) + }, + new HeaderValue { Key = "set", Value = new HashSet { 7 } } + ]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal( + new Dictionary + { + ["char"] = "c", + ["guid"] = "12345678-1234-1234-1234-123456789012", + ["uri"] = "rabbitmq:///q/x", + ["timespan"] = "01:30:00", + ["dateonly"] = "2026-08-03", + ["timeonly"] = "10:30:15.0000000", + ["enum"] = "High", + ["element"] = new Dictionary { ["k"] = 1 }, + ["node"] = new Dictionary { ["k"] = 1 }, + ["nested"] = new Dictionary { ["inner"] = "00:00:00" }, + ["set"] = new List { 7L } + }, + headers); + } + + [Fact] + public void BuildHeaders_Should_ReadUnzonedTimeAsUtc_When_HeaderIsDateTime() + { + // arrange + var envelope = new MessageEnvelope + { + Headers = new Headers( + [ + new HeaderValue { Key = "unspecified", Value = new DateTime(2024, 1, 15, 10, 30, 45) }, + new HeaderValue { Key = "utc", Value = new DateTime(2024, 1, 15, 10, 30, 45, DateTimeKind.Utc) } + ]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal( + new AmqpTimestamp(1705314645), + Assert.IsType(headers["unspecified"])); + Assert.Equal(headers["utc"], headers["unspecified"]); + } + + [Fact] + public void BuildHeaders_Should_ProduceTheEpochFloor_When_HeaderIsTheDefaultDateTime() + { + // arrange + var envelope = new MessageEnvelope + { + Headers = new Headers([new HeaderValue { Key = "at", Value = default(DateTime) }]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal( + DateTimeOffset.MinValue.ToUnixTimeSeconds(), + Assert.IsType(headers["at"]).UnixTime); + } + + [Fact] + public void BuildHeaders_Should_CarryDecimalAsText_When_ItExceedsTheTableMantissa() + { + // arrange + // the table's decimal holds a 32 bit mantissa, so a longer one travels as text + var envelope = new MessageEnvelope + { + Headers = new Headers( + [ + new HeaderValue { Key = "fits", Value = 9.95m }, + new HeaderValue { Key = "exceeds", Value = 21474836.48m }, + new HeaderValue { Key = "negative", Value = -21474836.48m } + ]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal( + new Dictionary + { + ["fits"] = 9.95m, + ["exceeds"] = "21474836.48", + ["negative"] = "-21474836.48" + }, + headers); + } + + [Fact] + public void BuildHeaders_Should_MapJsonToTableShapes_When_HeaderIsAJsonValue() + { + // arrange + // writing the JSON text instead would hand the consumer a string holding JSON, quotes and all + var envelope = new MessageEnvelope + { + Headers = new Headers( + [ + new HeaderValue { Key = "object", Value = new JsonObject { ["k"] = 1 } }, + new HeaderValue { Key = "array", Value = new JsonArray(1, 2) }, + new HeaderValue { Key = "text", Value = JsonValue.Create("abc") }, + new HeaderValue { Key = "number", Value = JsonValue.Create(5) } + ]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal( + new Dictionary + { + ["object"] = new Dictionary { ["k"] = 1 }, + ["array"] = new List { 1, 2 }, + ["text"] = "abc", + ["number"] = 5 + }, + headers); + } + + [Fact] + public void BuildHeaders_Should_KeepEscapes_When_HeaderIsAUri() + { + // arrange + // ToString would decode %20 to a space, leaving text that no longer reads back as this URI + var envelope = new MessageEnvelope + { + Headers = new Headers( + [ + new HeaderValue { Key = "callback", Value = new Uri("https://example.com/a%20b") } + ]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal("https://example.com/a%20b", headers["callback"]); + } + + [Fact] + public void BuildHeaders_Should_MapSiblingValues_When_ATableKeyIsNotText() + { + // arrange + var envelope = new MessageEnvelope + { + Headers = new Headers( + [ + new HeaderValue + { + Key = "meta", + Value = new Hashtable + { + [1] = "a", + ["time"] = DateTimeOffset.FromUnixTimeSeconds(1700000000) + } + } + ]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal( + new Dictionary { ["1"] = "a", ["time"] = new AmqpTimestamp(1700000000) }, + Assert.IsType>(headers["meta"])); + } + + [Fact] + public void BuildHeaders_Should_DeclareBinary_When_HeaderIsAByteMemory() + { + // arrange + var envelope = new MessageEnvelope + { + Headers = new Headers( + [ + new HeaderValue { Key = "readonly", Value = new ReadOnlyMemory("hi"u8.ToArray()) }, + new HeaderValue { Key = "writable", Value = new Memory("hi"u8.ToArray()) }, + new HeaderValue { Key = "empty", Value = default(ArraySegment) } + ]) + }; + + // act + var headers = envelope.BuildHeaders(); + + // assert + Assert.Equal("hi"u8.ToArray(), Assert.IsType(headers["readonly"]).Bytes); + Assert.Equal("hi"u8.ToArray(), Assert.IsType(headers["writable"]).Bytes); + Assert.Empty(Assert.IsType(headers["empty"]).Bytes); + } + + [Fact] + public void BuildHeaders_Should_ThrowNamingTheHeader_When_AValueContainsItself() + { + // arrange + var cyclic = new List(); + cyclic.Add(cyclic); + var envelope = new MessageEnvelope + { + Headers = new Headers([new HeaderValue { Key = "x-cycle", Value = cyclic }]) + }; + + // act + var exception = Record.Exception(() => envelope.BuildHeaders()); + + // assert + Assert.Equal( + "The header 'x-cycle' is nested more than 64 levels deep, or holds a value that " + + "contains itself, and cannot be written as an AMQP field table.", + Assert.IsType(exception).Message); + } + + private enum TestPriority + { + Low, + High + } } diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs index f6e4f0f618b..c471a2e6924 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs @@ -1,3 +1,4 @@ +using Mocha.Middlewares; using RabbitMQ.Client; using RabbitMQ.Client.Events; @@ -159,6 +160,65 @@ public void Parse_Should_DecodeByteArrayHeaders_When_Utf8Bytes() Assert.Equal("rabbitmq:///q/source", envelope.SourceAddress); } + [Fact] + public void Parse_Should_KeepBytes_When_HeaderIsBinaryTableValue() + { + // arrange + // the shape a client that maps a byte array onto the binary field type sends + var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary + { + ["x-signature"] = new BinaryTableValue("signature-123"u8.ToArray()) + }); + + // act + var envelope = _parser.Parse(args); + + // assert + Assert.True(envelope.Headers!.TryGetValue("x-signature", out var signature)); + Assert.Equal("signature-123"u8.ToArray(), signature); + } + + [Fact] + public void Parse_Should_KeepBytes_When_BinaryTableValueIsNestedInHeader() + { + // arrange + var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary + { + ["x-origin"] = new Dictionary + { + ["signature"] = new BinaryTableValue("signature-123"u8.ToArray()) + } + }); + + // act + var envelope = _parser.Parse(args); + + // assert + envelope.Headers!.TryGetValue("x-origin", out var origin); + Assert.Equal( + new Dictionary { ["signature"] = "signature-123"u8.ToArray() }, + Assert.IsType>(origin)); + } + + [Fact] + public void Parse_Should_KeepBytes_When_AByteHeaderIsPublishedAndReadBack() + { + // arrange + var payload = "COM1"u8.ToArray(); + var published = new MessageEnvelope + { + Headers = new Headers([new HeaderValue { Key = "x-signature", Value = payload }]) + }; + var args = CreateDeliverEventArgs(props => props.Headers = published.BuildHeaders()); + + // act + var envelope = _parser.Parse(args); + + // assert + envelope.Headers!.TryGetValue("x-signature", out var signature); + Assert.Equal(payload, Assert.IsType(signature)); + } + [Fact] public void Parse_Should_ConvertAmqpTimestamp_When_HeaderIsTimestamp() { @@ -173,6 +233,80 @@ public void Parse_Should_ConvertAmqpTimestamp_When_HeaderIsTimestamp() Assert.Equal(DateTimeOffset.FromUnixTimeSeconds(1700000000), ts); } + [Fact] + public void Parse_Should_KeepTheNumber_When_TimestampIsOutOfRange() + { + // arrange + var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary + { + ["x-millis"] = new AmqpTimestamp(1700000000000), + ["x-extreme"] = new AmqpTimestamp(long.MaxValue) + }); + + // act + var envelope = _parser.Parse(args); + + // assert + envelope.Headers!.TryGetValue("x-millis", out var millis); + envelope.Headers.TryGetValue("x-extreme", out var extreme); + Assert.Equal(1700000000000L, millis); + Assert.Equal(long.MaxValue, extreme); + } + + [Fact] + public void Parse_Should_DecodeNestedByteArrays_When_HeaderIsDeathTable() + { + // arrange + // the shape the broker uses for x-death: an array of field tables + var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary + { + ["x-death"] = new List + { + new Dictionary + { + ["queue"] = "product-queue"u8.ToArray(), + ["reason"] = "rejected"u8.ToArray(), + ["time"] = new AmqpTimestamp(1700000000), + ["count"] = 1L + } + } + }); + + // act + var envelope = _parser.Parse(args); + + // assert + envelope.Headers!.TryGetValue("x-death", out var death); + var entry = Assert.IsType>(Assert.Single(Assert.IsType(death))); + Assert.Equal( + new Dictionary + { + ["queue"] = "product-queue", + ["reason"] = "rejected", + ["time"] = DateTimeOffset.FromUnixTimeSeconds(1700000000), + ["count"] = 1L + }, + entry); + } + + [Fact] + public void Parse_Should_KeepBytes_When_HeaderValueIsNotValidUtf8() + { + // arrange + var value = new byte[] { 0x00, 0xFF, 0xFE, 0x41 }; + var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary + { + ["x-binary"] = value + }); + + // act + var envelope = _parser.Parse(args); + + // assert + Assert.True(envelope.Headers!.TryGetValue("x-binary", out var parsed)); + Assert.Equal(value, parsed); + } + [Fact] public void Parse_Should_SetDeliveryCountOne_When_Redelivered() { diff --git a/website/content/docs/mocha/messages.md b/website/content/docs/mocha/messages.md index eb1ef004402..a51e86456a5 100644 --- a/website/content/docs/mocha/messages.md +++ b/website/content/docs/mocha/messages.md @@ -73,6 +73,70 @@ await bus.SendAsync( The bus merges your headers into the envelope's header collection before dispatching. +## What a header value can hold + +A header value is weakly typed, but the set of types Mocha carries is closed: `null`, `bool`, +`string`, `char`, the numeric types, `DateTime`, `DateTimeOffset`, `DateOnly`, `TimeOnly`, +`TimeSpan`, `Guid`, `Uri`, an enum, a byte payload (`byte[]`, `ArraySegment`, +`ReadOnlyMemory`, `Memory`), a JSON value (`JsonElement`, `JsonDocument`, `JsonNode`), +nested headers, a dictionary with string keys, or a sequence of any of these. + +The envelope serializer rejects any other type with an error naming the header. Setting a value always +succeeds; a transport that encodes headers itself applies its own rules, described at the end of this +section. + +A value keeps its representation across a hop but not always its CLR type, because neither JSON nor an +AMQP field table carries a .NET type tag. Through the envelope serializer, which the outbox and the +Entity Framework Core scheduled store use: + +| Written as | Read back as | +| ------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------- | +| `null`, `bool`, `string` | unchanged | +| any integer type | `int`, `long` or `ulong`, the smallest of those holding the value | +| `double`, `float`, `decimal` | the same integer types when the value is whole, else `double` | +| `DateTime`, `DateTimeOffset`, `Guid`, `Uri`, `TimeSpan`, `DateOnly`, `TimeOnly`, `char`, an enum | the text that was written | +| a byte payload | base64 text | +| a JSON object, nested headers, a dictionary | `Dictionary` | +| a JSON array, any sequence | `object?[]` | + +A number is read by value, not by the type that wrote it, so `5L` and `2.0` both come back as `int`, +and a `uint` above `int.MaxValue` as `long`. Two cases lose data: a whole number beyond +`ulong.MaxValue` is read as a `double` and rewritten in scientific notation, and a `decimal` carrying +more precision than a `double` holds loses the difference. + +A date-shaped string stays a string. Parse explicitly when you need a typed value: + +```csharp +using System.Globalization; + +if (context.Headers.TryGetValue("x-processed-at", out var value) + && DateTimeOffset.TryParse(value as string, CultureInfo.InvariantCulture, out var processedAt)) +{ + // ... +} +``` + +A transport that defines its own header encoding differs in the details. + +**RabbitMQ** maps a header onto an AMQP field table. A date arrives back as a `DateTimeOffset` +truncated to whole seconds, and a `DateTime` carrying no time zone is read as UTC. A byte payload +arrives back as bytes, travelling as the table's own byte array type, and a JSON value arrives in the +shapes listed above. The table has no unsigned 64-bit type and only a 32-bit decimal mantissa, so a +number above `long.MaxValue` and a longer `decimal` travel as text. A long string from another +publisher is decoded as text when it holds valid UTF-8. + +**Postgres** reads every number back as a `double`, and drops a header whose value is a dictionary, a +sequence, or `null`. Messages scheduled on this transport are stored with the Postgres encoding rather +than through the envelope serializer, so the table above does not describe them. + +Those typed arrivals hold for a message taken straight off the broker. One that passes through the +outbox, or that the scheduled store redelivers, is rehydrated by the envelope serializer on the way, +so the same date arrives as text and the same byte payload as base64. + +> Keep header values simple. Text and booleans behave the same everywhere, and whole numbers behave +> the same anywhere except Postgres; everything else is worth a round-trip test on the transport you +> actually deploy. + ## Access envelope metadata in a handler `IEventHandler` receives the deserialized message and a cancellation token - that is all. To read message IDs, correlation IDs, timestamps, or custom headers, implement `IConsumer` instead. The `IConsumeContext` parameter gives you both the deserialized message and all envelope fields: From e6a93b95a258a721621a9feb473c0be2abf67be3 Mon Sep 17 00:00:00 2001 From: alisan3 <41369666+alisan3@users.noreply.github.com> Date: Tue, 4 Aug 2026 07:32:07 +0100 Subject: [PATCH 2/8] Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs index 4fcfb370fbf..3352b709cb2 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs @@ -341,7 +341,7 @@ public static class RabbitMQDispatchContextExtensions return ToTableValue(HeadersJsonConverter.ReadHeaderValue(document.RootElement), key, depth); case JsonNode node: - using (var parsed = JsonDocument.Parse(node.ToJsonString())) + using (var parsed = JsonSerializer.SerializeToDocument(node)) { return ToTableValue(HeadersJsonConverter.ReadHeaderValue(parsed.RootElement), key, depth); } From 497bca22f5282a1e7e2f637b146310ac16423031 Mon Sep 17 00:00:00 2001 From: Alexander Bohle Date: Tue, 4 Aug 2026 09:24:10 +0100 Subject: [PATCH 3/8] Bound the depth of the header value read from a delivery NormalizeValue walked nested tables and arrays from the broker without a limit. The value it walks is built by the client's own recursive decode, but the walk runs from RabbitMQParsingMiddleware, deeper in the stack than that decode, so it has less headroom and a crafted delivery could exhaust it there. The receive path now carries the header key and depth, matching the publish path, and rejects a value nested past 64 levels with an error naming the header. Unlike the cap this replaces, it does not stop mapping and hand back an unmapped subtree, which produced an envelope that could not be serialized. Co-Authored-By: Claude Fable 5 --- .../RabbitMQMessageEnvelopeParser.cs | 24 +++++++++++++----- .../RabbitMQMessageEnvelopeParserTests.cs | 25 +++++++++++++++++++ 2 files changed, 43 insertions(+), 6 deletions(-) diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs index 93e9485c136..4636d79a4b2 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs @@ -12,6 +12,11 @@ namespace Mocha.Transport.RabbitMQ; /// internal sealed class RabbitMQMessageEnvelopeParser { + /// + /// The greatest depth to which a header value is read. A value nested deeper is rejected. + /// + private const int MaxHeaderDepth = 64; + /// /// The range a can express, which bounds the timestamps mapped to one. /// @@ -94,7 +99,7 @@ private static Headers BuildHeaders(IDictionary? headers) var result = new Headers(headers.Count); foreach (var (key, value) in headers) { - result.Set(key, NormalizeValue(value)); + result.Set(key, NormalizeValue(value, key, 0)); } return result; @@ -102,10 +107,17 @@ private static Headers BuildHeaders(IDictionary? headers) /// /// Maps AMQP wire types onto the CLR types the envelope serializer understands, at every level of - /// a nested value. + /// a nested value. Throws when a value is nested deeper than . /// - private static object? NormalizeValue(object? value) + private static object? NormalizeValue(object? value, string key, int depth) { + if (depth > MaxHeaderDepth) + { + throw new InvalidOperationException( + $"The header '{key}' is nested more than {MaxHeaderDepth} levels deep and cannot be " + + "read as a message header."); + } + switch (value) { // a binary field is kept as bytes without being tested for text @@ -125,9 +137,9 @@ private static Headers BuildHeaders(IDictionary? headers) case IDictionary table: var mappedTable = new Dictionary(table.Count); - foreach (var (key, item) in table) + foreach (var (entryKey, item) in table) { - mappedTable[key] = NormalizeValue(item); + mappedTable[entryKey] = NormalizeValue(item, key, depth + 1); } return mappedTable; @@ -136,7 +148,7 @@ private static Headers BuildHeaders(IDictionary? headers) var mappedArray = new object?[array.Count]; for (var i = 0; i < array.Count; i++) { - mappedArray[i] = NormalizeValue(array[i]); + mappedArray[i] = NormalizeValue(array[i], key, depth + 1); } return mappedArray; diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs index c471a2e6924..aea50812495 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs @@ -219,6 +219,31 @@ public void Parse_Should_KeepBytes_When_AByteHeaderIsPublishedAndReadBack() Assert.Equal(payload, Assert.IsType(signature)); } + [Fact] + public void Parse_Should_ThrowNamingTheHeader_When_HeaderIsNestedTooDeeply() + { + // a crafted delivery, nested past the depth a legitimate broker header reaches + object value = "leaf"; + for (var i = 0; i < 200; i++) + { + value = new List { value }; + } + + var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary + { + ["x-deep"] = value + }); + + // act + var exception = Record.Exception(() => _parser.Parse(args)); + + // assert + Assert.Equal( + "The header 'x-deep' is nested more than 64 levels deep and cannot be read as a message " + + "header.", + Assert.IsType(exception).Message); + } + [Fact] public void Parse_Should_ConvertAmqpTimestamp_When_HeaderIsTimestamp() { From b63f5d6dd2f4cd505a6cf4678b20f7ced4e8920e Mon Sep 17 00:00:00 2001 From: Alexander Bohle Date: Tue, 4 Aug 2026 09:31:25 +0100 Subject: [PATCH 4/8] fix unnessesary supression --- .../src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs index 3352b709cb2..5cc0fdd30e5 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs @@ -354,7 +354,7 @@ public static class RabbitMQDispatchContextExtensions case ArraySegment { Array: null }: return new BinaryTableValue([]); - case ArraySegment { Offset: 0 } whole when whole.Count == whole.Array!.Length: + case ArraySegment { Offset: 0 } whole when whole.Count == whole.Array.Length: return new BinaryTableValue(whole.Array); case ArraySegment segment: From 6ef2e65bf4dfc9dab1e24f1ef9b030ea327a516e Mon Sep 17 00:00:00 2001 From: Alexander Bohle Date: Tue, 4 Aug 2026 09:49:47 +0100 Subject: [PATCH 5/8] Use the spelling the dictionary carries in the header docs Co-Authored-By: Claude Fable 5 --- website/content/docs/mocha/messages.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/website/content/docs/mocha/messages.md b/website/content/docs/mocha/messages.md index a51e86456a5..35dd7c92f2e 100644 --- a/website/content/docs/mocha/messages.md +++ b/website/content/docs/mocha/messages.md @@ -120,7 +120,7 @@ A transport that defines its own header encoding differs in the details. **RabbitMQ** maps a header onto an AMQP field table. A date arrives back as a `DateTimeOffset` truncated to whole seconds, and a `DateTime` carrying no time zone is read as UTC. A byte payload -arrives back as bytes, travelling as the table's own byte array type, and a JSON value arrives in the +arrives back as bytes, carried by the table's own byte array type, and a JSON value arrives in the shapes listed above. The table has no unsigned 64-bit type and only a 32-bit decimal mantissa, so a number above `long.MaxValue` and a longer `decimal` travel as text. A long string from another publisher is decoded as text when it holds valid UTF-8. From 561a97e31e7af0525b4084a1308be25e7b9b8a42 Mon Sep 17 00:00:00 2001 From: Pascal Senn Date: Wed, 12 Aug 2026 11:18:46 +0200 Subject: [PATCH 6/8] Unify RabbitMQ header parsing and formatting --- .../Extensions/DateTimeExtensions.cs | 22 ++ .../Receive/RabbitMQParsingMiddleware.cs | 16 +- .../RabbitMQDispatchEndpoint.cs | 261 +--------------- .../RabbitMQFieldTableValueConverter.cs | 278 ++++++++++++++++++ .../RabbitMQMessageEnvelopeFormatter.cs | 107 +++++++ .../RabbitMQMessageEnvelopeParser.cs | 172 +++++------ .../RabbitMQMessageHeaders.cs | 48 +-- .../Mocha/Serialization/HeaderValueText.cs | 27 -- .../Mocha/Serialization/HeadersConverter.cs | 121 +------- src/Mocha/src/Mocha/ThrowHelper.cs | 20 ++ .../HeadersJsonConverterTests.cs | 21 +- .../Transport/MessageEnvelopeWriterTests.cs | 12 +- .../HeaderValueConsistencyTests.cs | 93 +++--- ... RabbitMQMessageEnvelopeFormatterTests.cs} | 149 ++++++---- .../RabbitMQMessageEnvelopeParserTests.cs | 72 ++--- website/content/docs/mocha/messages.md | 89 ++---- 16 files changed, 750 insertions(+), 758 deletions(-) create mode 100644 src/Mocha/src/Mocha.Transport.RabbitMQ/Extensions/DateTimeExtensions.cs create mode 100644 src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQFieldTableValueConverter.cs create mode 100644 src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeFormatter.cs delete mode 100644 src/Mocha/src/Mocha/Serialization/HeaderValueText.cs rename src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/{RabbitMQDispatchContextExtensionsTests.cs => RabbitMQMessageEnvelopeFormatterTests.cs} (68%) diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/Extensions/DateTimeExtensions.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/Extensions/DateTimeExtensions.cs new file mode 100644 index 00000000000..18288035863 --- /dev/null +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/Extensions/DateTimeExtensions.cs @@ -0,0 +1,22 @@ +using RabbitMQ.Client; + +namespace Mocha.Transport.RabbitMQ; + +internal static class DateTimeExtensions +{ + extension(DateTimeOffset value) + { + internal AmqpTimestamp ToAmqpTimestamp() + => new(value.ToUnixTimeSeconds()); + } + + extension(DateTime value) + { + internal AmqpTimestamp ToAmqpTimestamp() + { + var utc = value.Kind == DateTimeKind.Local ? value.ToUniversalTime() : value; + + return new AmqpTimestamp(new DateTimeOffset(utc, TimeSpan.Zero).ToUnixTimeSeconds()); + } + } +} diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/Middlewares/Receive/RabbitMQParsingMiddleware.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/Middlewares/Receive/RabbitMQParsingMiddleware.cs index 26d78a8fd62..7676f4c77cb 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/Middlewares/Receive/RabbitMQParsingMiddleware.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/Middlewares/Receive/RabbitMQParsingMiddleware.cs @@ -1,3 +1,4 @@ +using Microsoft.Extensions.DependencyInjection; using Mocha.Features; using Mocha.Middlewares; using Mocha.Transport.RabbitMQ.Features; @@ -8,7 +9,7 @@ namespace Mocha.Transport.RabbitMQ.Middlewares; /// Receive middleware that parses the raw RabbitMQ delivery into a /// and sets it on the receive context for downstream processing. /// -internal sealed class RabbitMQParsingMiddleware +internal sealed class RabbitMQParsingMiddleware(TimeProvider timeProvider) { /// /// Parses the RabbitMQ delivery event args into a message envelope and invokes the next middleware. @@ -20,19 +21,24 @@ public async ValueTask InvokeAsync(IReceiveContext context, ReceiveDelegate next var feature = context.Features.GetOrSet(); var eventArgs = feature.EventArgs; - var envelope = RabbitMQMessageEnvelopeParser.Instance.Parse(eventArgs); + var envelope = RabbitMQMessageEnvelopeParser.Instance.Parse(eventArgs, timeProvider); context.SetEnvelope(envelope); await next(context); } - private static readonly RabbitMQParsingMiddleware s_instance = new(); - /// /// Creates a that wraps the parsing middleware singleton. /// /// A middleware configuration keyed as "RabbitMQParsing". public static ReceiveMiddlewareConfiguration Create() - => new(static (_, next) => ctx => s_instance.InvokeAsync(ctx, next), "RabbitMQParsing"); + => new( + static (context, next) => + { + var timeProvider = context.Services.GetRequiredService(); + var middleware = new RabbitMQParsingMiddleware(timeProvider); + return ctx => middleware.InvokeAsync(ctx, next); + }, + "RabbitMQParsing"); } diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs index 5cc0fdd30e5..bd5010e01e5 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs @@ -1,7 +1,4 @@ -using System.Collections; -using System.Globalization; -using System.Text.Json; -using System.Text.Json.Nodes; +using Microsoft.Extensions.DependencyInjection; using Mocha.Middlewares; using RabbitMQ.Client; using static System.StringSplitOptions; @@ -35,11 +32,12 @@ protected override async ValueTask DispatchAsync(IDispatchContext context) var dispatcher = transport.Dispatcher; var cancellationToken = context.CancellationToken; + var timeProvider = context.Services.GetRequiredService(); var channel = await dispatcher.RentChannelAsync(cancellationToken); try { await EnsureProvisionedAsync(channel, cancellationToken); - await DispatchAsync(channel, envelope, cancellationToken); + await DispatchAsync(channel, envelope, timeProvider, cancellationToken); } finally { @@ -50,6 +48,7 @@ protected override async ValueTask DispatchAsync(IDispatchContext context) private async ValueTask DispatchAsync( IChannel channel, MessageEnvelope envelope, + TimeProvider timeProvider, CancellationToken cancellationToken) { var exchangeName = CachedString.Empty; @@ -125,24 +124,7 @@ private async ValueTask DispatchAsync( } } - var headers = envelope.BuildHeaders(); - - var messageType = envelope.MessageType ?? headers.Get(RabbitMQMessageHeaders.MessageType); - - var properties = new BasicProperties - { - MessageId = envelope.MessageId, - CorrelationId = envelope.CorrelationId, - Type = messageType, - Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()), - ReplyTo = envelope.ResponseAddress, - Headers = headers, - ContentType = envelope.ContentType, - DeliveryMode = DeliveryModes.Persistent - // TODO wire up durable - // TODO expiration - // TODO priority - }; + var properties = RabbitMQMessageEnvelopeFormatter.Format(envelope, timeProvider); await channel.BasicPublishAsync(exchangeName, routingKey, true, properties, envelope.Body, cancellationToken); } @@ -205,236 +187,3 @@ Exchange as TopologyResource ?? throw new InvalidOperationException("Destination is not set"); } } - -/// -/// Extension methods for building RabbitMQ-specific message headers from a . -/// -public static class RabbitMQDispatchContextExtensions -{ - internal static IDictionary BuildHeaders(this MessageEnvelope envelope) - { - var headerCount = - (envelope.ConversationId is not null ? 1 : 0) - + (envelope.CausationId is not null ? 1 : 0) - + (envelope.SourceAddress is not null ? 1 : 0) - + (envelope.DestinationAddress is not null ? 1 : 0) - + (envelope.FaultAddress is not null ? 1 : 0) - + (envelope.Headers?.Count ?? 0); - - var headers = new Dictionary(headerCount); - - if (envelope.Headers is not null) - { - foreach (var header in envelope.Headers) - { - headers[header.Key] = ToTableValue(header.Value, header.Key, 0); - } - } - - if (envelope.ConversationId is not null) - { - headers.Set(RabbitMQMessageHeaders.ConversationId, envelope.ConversationId); - } - - if (envelope.CausationId is not null) - { - headers.Set(RabbitMQMessageHeaders.CausationId, envelope.CausationId); - } - - if (envelope.SourceAddress is not null) - { - headers.Set(RabbitMQMessageHeaders.SourceAddress, envelope.SourceAddress); - } - - if (envelope.DestinationAddress is not null) - { - headers.Set(RabbitMQMessageHeaders.DestinationAddress, envelope.DestinationAddress); - } - - if (envelope.FaultAddress is not null) - { - headers.Set(RabbitMQMessageHeaders.FaultAddress, envelope.FaultAddress); - } - - if (envelope.EnclosedMessageTypes is { Length: > 0 }) - { - headers.Set(RabbitMQMessageHeaders.EnclosedMessageTypes, envelope.EnclosedMessageTypes.Value); - } - - if (envelope.MessageType is not null) - { - headers.Set(RabbitMQMessageHeaders.MessageType, envelope.MessageType); - } - - return headers; - } - - /// - /// The greatest depth to which a header value is mapped. A value nested deeper, or one that - /// contains itself, is rejected. - /// - private const int MaxTableDepth = 64; - - /// - /// Maps a header value onto the CLR types an AMQP field table accepts, at every level of a nested - /// value. The inverse of the mapping the receive side applies. - /// - private static object? ToTableValue(object? value, string key, int depth) - { - if (depth > MaxTableDepth) - { - throw new InvalidOperationException( - $"The header '{key}' is nested more than {MaxTableDepth} levels deep, or holds a " - + "value that contains itself, and cannot be written as an AMQP field table."); - } - - switch (value) - { - // must precede the collection cases - case string: - return value; - - case DateTimeOffset dateTimeOffset: - return new AmqpTimestamp(dateTimeOffset.ToUnixTimeSeconds()); - - case DateTime dateTime: - return ToTimestamp(dateTime); - - // a field table has no unsigned 64 bit type - case ulong unsigned when unsigned > long.MaxValue: - return unsigned.ToString(CultureInfo.InvariantCulture); - - case ulong unsigned: - return (long)unsigned; - - // a field table decimal holds a 32 bit mantissa - case decimal number when !FitsTableDecimal(number): - return number.ToString(CultureInfo.InvariantCulture); - - // the text forms every transport shares - case char character: - return character.ToString(); - - case Guid guid: - return guid.ToString(); - - case Uri uri: - return HeaderValueText.From(uri); - - case TimeSpan timeSpan: - return HeaderValueText.From(timeSpan); - - case DateOnly date: - return HeaderValueText.From(date); - - case TimeOnly time: - return HeaderValueText.From(time); - - case Enum enumeration: - return HeaderValueText.From(enumeration); - - // the table's own shapes rather than JSON text - case JsonElement element: - return ToTableValue(HeadersJsonConverter.ReadHeaderValue(element), key, depth); - - case JsonDocument document: - return ToTableValue(HeadersJsonConverter.ReadHeaderValue(document.RootElement), key, depth); - - case JsonNode node: - using (var parsed = JsonSerializer.SerializeToDocument(node)) - { - return ToTableValue(HeadersJsonConverter.ReadHeaderValue(parsed.RootElement), key, depth); - } - - // the field table's byte array type, not a long string - case byte[] bytes: - return new BinaryTableValue(bytes); - - // an empty segment has no array to copy, and a whole one needs no copy - case ArraySegment { Array: null }: - return new BinaryTableValue([]); - - case ArraySegment { Offset: 0 } whole when whole.Count == whole.Array.Length: - return new BinaryTableValue(whole.Array); - - case ArraySegment segment: - return new BinaryTableValue(segment.ToArray()); - - case ReadOnlyMemory memory: - return new BinaryTableValue(memory.ToArray()); - - case Memory writableMemory: - return new BinaryTableValue(writableMemory.ToArray()); - - case IReadOnlyHeaders nested: - var mappedHeaders = new Dictionary(); - foreach (var header in nested) - { - mappedHeaders[header.Key] = ToTableValue(header.Value, key, depth + 1); - } - - return mappedHeaders; - - case IDictionary table: - return ToTable(table, key, depth); - - case IEnumerable sequence: - var mappedSequence = sequence is ICollection collection - ? new List(collection.Count) - : []; - foreach (var item in sequence) - { - mappedSequence.Add(ToTableValue(item, key, depth + 1)); - } - - return mappedSequence; - - default: - return value; - } - } - - private static Dictionary ToTable(IDictionary table, string key, int depth) - { - var mapped = new Dictionary(table.Count); - - foreach (DictionaryEntry entry in table) - { - // a field table names its entries with text; the client stringifies any other key - if (entry.Key is not { } entryKey) - { - continue; - } - - var name = entryKey as string ?? entryKey.ToString(); - - if (name is null) - { - continue; - } - - mapped[name] = ToTableValue(entry.Value, key, depth + 1); - } - - return mapped; - } - - /// - /// Converts a to the instant a field table carries. A local time is - /// converted from this host; a time with no zone is read as UTC. - /// - private static AmqpTimestamp ToTimestamp(DateTime value) - { - var utc = value.Kind == DateTimeKind.Local ? value.ToUniversalTime() : value; - - return new AmqpTimestamp(new DateTimeOffset(utc, TimeSpan.Zero).ToUnixTimeSeconds()); - } - - private static bool FitsTableDecimal(decimal value) - { - Span bits = stackalloc int[4]; - decimal.GetBits(value, bits); - - return bits[1] == 0 && bits[2] == 0 && (uint)bits[0] <= int.MaxValue; - } -} diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQFieldTableValueConverter.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQFieldTableValueConverter.cs new file mode 100644 index 00000000000..3c1a062b88b --- /dev/null +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQFieldTableValueConverter.cs @@ -0,0 +1,278 @@ +using System.Collections; +using System.Globalization; +using System.Text; +using System.Text.Json; +using System.Text.Json.Nodes; +using System.Text.Unicode; +using RabbitMQ.Client; + +namespace Mocha.Transport.RabbitMQ; + +/// +/// Maps message header values to and from RabbitMQ AMQP field-table values. +/// +internal static class RabbitMQFieldTableValueConverter +{ + /// + /// The greatest depth to which a header value is mapped in either direction. + /// + private const int MaxHeaderDepth = 64; + + /// + /// The range a can express, which bounds the timestamps mapped to one. + /// + private static readonly long s_minTimestampSeconds = DateTimeOffset.MinValue.ToUnixTimeSeconds(); + + private static readonly long s_maxTimestampSeconds = DateTimeOffset.MaxValue.ToUnixTimeSeconds(); + + public static object? ToFieldTableValue(this object? value, string key) => value.ToFieldTableValue(key, 0); + + public static object? FromFieldTableValue(this object? value, string key) => value.FromFieldTableValue(key, 0); + + private static object? ToFieldTableValue(this object? value, string key, int depth) + { + EnsureDepth(key, depth); + + switch (value) + { + case string: + return value; + + case DateTimeOffset dateTimeOffset: + return dateTimeOffset.ToAmqpTimestamp(); + + case DateTime dateTime: + return dateTime.ToAmqpTimestamp(); + + // AMQP field tables have no unsigned 64-bit integer type. + case ulong unsigned when unsigned > long.MaxValue: + return unsigned.ToString(CultureInfo.InvariantCulture); + + case ulong unsigned: + return (long)unsigned; + + // AMQP field table decimals are limited to a 32-bit mantissa. + case decimal number when !FitsTableDecimal(number): + return number.ToString(CultureInfo.InvariantCulture); + + case char character: + return character.ToString(); + + case Guid guid: + return guid.ToString(); + + case Uri uri: + return uri.OriginalString; + + case TimeSpan timeSpan: + return timeSpan.ToString("c", CultureInfo.InvariantCulture); + + case DateOnly date: + return date.ToString("O", CultureInfo.InvariantCulture); + + case TimeOnly time: + return time.ToString("O", CultureInfo.InvariantCulture); + + case Enum enumeration: + return enumeration.ToString(); + + case JsonElement element: + return element.ToFieldTableValue(key, depth); + + case JsonDocument document: + return document.RootElement.ToFieldTableValue(key, depth); + + case JsonNode node: + using (var document = JsonSerializer.SerializeToDocument(node)) + { + return document.RootElement.ToFieldTableValue(key, depth); + } + + // BinaryTableValue selects the AMQP byte-array type instead of a long string. + case byte[] bytes: + return new BinaryTableValue(bytes); + + case ArraySegment { Array: null }: + return new BinaryTableValue([]); + + case ArraySegment { Offset: 0 } whole when whole.Count == whole.Array.Length: + return new BinaryTableValue(whole.Array); + + case ArraySegment segment: + return new BinaryTableValue(segment.ToArray()); + + case ReadOnlyMemory memory: + return new BinaryTableValue(memory.ToArray()); + + case Memory writableMemory: + return new BinaryTableValue(writableMemory.ToArray()); + + case IReadOnlyHeaders nested: + var mappedHeaders = new Dictionary(); + foreach (var header in nested) + { + mappedHeaders[header.Key] = header.Value.ToFieldTableValue(key, depth + 1); + } + + return mappedHeaders; + + case IDictionary table: + return table.ToFieldTable(key, depth); + + case IEnumerable sequence: + var mappedSequence = sequence is ICollection collection ? new List(collection.Count) : []; + foreach (var item in sequence) + { + mappedSequence.Add(item.ToFieldTableValue(key, depth + 1)); + } + + return mappedSequence; + + default: + return value; + } + } + + private static object? ToFieldTableValue(this JsonElement element, string key, int depth) + { + EnsureDepth(key, depth); + + switch (element.ValueKind) + { + case JsonValueKind.Object: + var table = new Dictionary(); + foreach (var property in element.EnumerateObject()) + { + table[property.Name] = property.Value.ToFieldTableValue(key, depth + 1); + } + + return table; + + case JsonValueKind.Array: + var items = new List(); + foreach (var item in element.EnumerateArray()) + { + items.Add(item.ToFieldTableValue(key, depth + 1)); + } + + return items; + + case JsonValueKind.String: + return element.GetString(); + + case JsonValueKind.Number: + if (element.TryGetInt32(out var intValue)) + { + return intValue; + } + if (element.TryGetInt64(out var longValue)) + { + return longValue; + } + if (element.TryGetUInt64(out var unsignedValue)) + { + return unsignedValue <= long.MaxValue + ? (long)unsignedValue + : unsignedValue.ToString(CultureInfo.InvariantCulture); + } + + return element.GetDouble(); + + case JsonValueKind.True: + return true; + + case JsonValueKind.False: + return false; + + default: + return null; + } + } + + private static Dictionary ToFieldTable(this IDictionary table, string key, int depth) + { + var mapped = new Dictionary(table.Count); + + foreach (DictionaryEntry entry in table) + { + // a field table names its entries with text; the client stringifies any other key + if (entry.Key is not { } entryKey) + { + continue; + } + + var name = entryKey as string ?? entryKey.ToString(); + + if (name is null) + { + continue; + } + + mapped[name] = entry.Value.ToFieldTableValue(key, depth + 1); + } + + return mapped; + } + + private static object? FromFieldTableValue(this object? value, string key, int depth) + { + EnsureDepth(key, depth); + + switch (value) + { + // a binary field is kept as bytes without being tested for text + case BinaryTableValue binary: + return binary.Bytes; + + // a long string carries text and binary alike, so content is the only signal + case byte[] bytes: + return Utf8.IsValid(bytes) ? Encoding.UTF8.GetString(bytes) : bytes; + + // a value outside the range a date can express is kept as the number it is + case AmqpTimestamp timestamp: + return timestamp.UnixTime >= s_minTimestampSeconds && timestamp.UnixTime <= s_maxTimestampSeconds + ? DateTimeOffset.FromUnixTimeSeconds(timestamp.UnixTime) + : timestamp.UnixTime; + + case IDictionary table: + var mappedTable = new Dictionary(table.Count); + foreach (var (entryKey, item) in table) + { + mappedTable[entryKey] = item.FromFieldTableValue(key, depth + 1); + } + + return mappedTable; + + case IList array: + var mappedArray = new object?[array.Count]; + for (var i = 0; i < array.Count; i++) + { + mappedArray[i] = array[i].FromFieldTableValue(key, depth + 1); + } + + return mappedArray; + + default: + return value; + } + } + + private static void EnsureDepth(string key, int depth) + { + if (depth > MaxHeaderDepth) + { + throw new InvalidOperationException( + $"The header '{key}' exceeds the maximum AMQP field-table nesting depth of {MaxHeaderDepth}."); + } + } + + private static bool FitsTableDecimal(decimal value) + { + // decimal stores a 96-bit mantissa in the first three words, while an AMQP field-table + // decimal has only a signed 32-bit mantissa. + Span bits = stackalloc int[4]; + decimal.GetBits(value, bits); + + return bits[1] == 0 && bits[2] == 0 && (uint)bits[0] <= int.MaxValue; + } +} diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeFormatter.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeFormatter.cs new file mode 100644 index 00000000000..53a68feeff1 --- /dev/null +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeFormatter.cs @@ -0,0 +1,107 @@ +using System.Globalization; +using Mocha.Middlewares; +using RabbitMQ.Client; + +namespace Mocha.Transport.RabbitMQ; + +internal static class RabbitMQMessageEnvelopeFormatter +{ + public static BasicProperties Format(MessageEnvelope envelope, TimeProvider timeProvider) + { + var sentAt = timeProvider.GetUtcNow(); + var headers = FormatHeaders(envelope); + var messageType = envelope.MessageType ?? headers.Get(RabbitMQMessageHeaders.MessageType); + + return new BasicProperties + { + MessageId = envelope.MessageId, + CorrelationId = envelope.CorrelationId, + Type = messageType, + Timestamp = sentAt.ToAmqpTimestamp(), + Expiration = envelope.DeliverBy.ToExpiration(sentAt), + ReplyTo = envelope.ResponseAddress, + Headers = headers, + ContentType = envelope.ContentType, + DeliveryMode = DeliveryModes.Persistent + }; + } + + private static IDictionary FormatHeaders(MessageEnvelope envelope) + { + var headerCount = + (envelope.ConversationId is not null ? 1 : 0) + + (envelope.CausationId is not null ? 1 : 0) + + (envelope.SourceAddress is not null ? 1 : 0) + + (envelope.DestinationAddress is not null ? 1 : 0) + + (envelope.FaultAddress is not null ? 1 : 0) + + (envelope.EnclosedMessageTypes is { Length: > 0 } ? 1 : 0) + + (envelope.MessageType is not null ? 1 : 0) + + (envelope.Headers?.Count ?? 0); + + var headers = new Dictionary(headerCount); + + if (envelope.Headers is not null) + { + foreach (var header in envelope.Headers) + { + headers[header.Key] = header.Value.ToFieldTableValue(header.Key); + } + } + + if (envelope.ConversationId is not null) + { + headers.Set(RabbitMQMessageHeaders.ConversationId, envelope.ConversationId); + } + + if (envelope.CausationId is not null) + { + headers.Set(RabbitMQMessageHeaders.CausationId, envelope.CausationId); + } + + if (envelope.SourceAddress is not null) + { + headers.Set(RabbitMQMessageHeaders.SourceAddress, envelope.SourceAddress); + } + + if (envelope.DestinationAddress is not null) + { + headers.Set(RabbitMQMessageHeaders.DestinationAddress, envelope.DestinationAddress); + } + + if (envelope.FaultAddress is not null) + { + headers.Set(RabbitMQMessageHeaders.FaultAddress, envelope.FaultAddress); + } + + if (envelope.EnclosedMessageTypes is { Length: > 0 }) + { + headers.Set(RabbitMQMessageHeaders.EnclosedMessageTypes, envelope.EnclosedMessageTypes.Value); + } + + if (envelope.MessageType is not null) + { + headers.Set(RabbitMQMessageHeaders.MessageType, envelope.MessageType); + } + + return headers; + } +} + +file static class Extensions +{ + public static string? ToExpiration(this DateTimeOffset? deliverBy, DateTimeOffset sentAt) + { + if (deliverBy is null) + { + return null; + } + + var timeToLive = deliverBy.Value - sentAt; + if (timeToLive <= TimeSpan.Zero) + { + return "0"; + } + + return ((long)Math.Ceiling(timeToLive.TotalMilliseconds)).ToString(CultureInfo.InvariantCulture); + } +} diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs index 4636d79a4b2..e280a22c14e 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs @@ -1,5 +1,5 @@ +using System.Collections.Immutable; using System.Text; -using System.Text.Unicode; using Mocha.Middlewares; using RabbitMQ.Client; using RabbitMQ.Client.Events; @@ -12,48 +12,40 @@ namespace Mocha.Transport.RabbitMQ; /// internal sealed class RabbitMQMessageEnvelopeParser { - /// - /// The greatest depth to which a header value is read. A value nested deeper is rejected. - /// - private const int MaxHeaderDepth = 64; - - /// - /// The range a can express, which bounds the timestamps mapped to one. - /// - private static readonly long s_minTimestampSeconds = DateTimeOffset.MinValue.ToUnixTimeSeconds(); - - private static readonly long s_maxTimestampSeconds = DateTimeOffset.MaxValue.ToUnixTimeSeconds(); - /// /// Converts a RabbitMQ delivery into a by mapping AMQP basic properties /// and custom headers to envelope fields. /// /// The delivery event args containing the message body, properties, and metadata. + /// The clock used when expiration has no send timestamp. /// A fully populated message envelope ready for the receive middleware pipeline. - public MessageEnvelope Parse(BasicDeliverEventArgs eventArgs) + public MessageEnvelope Parse(BasicDeliverEventArgs eventArgs, TimeProvider timeProvider) { var props = eventArgs.BasicProperties; - var sentAt = props.Timestamp.UnixTime > 0 - ? DateTimeOffset.FromUnixTimeSeconds(props.Timestamp.UnixTime) - : (DateTimeOffset?)null; + var sentAt = + props.Timestamp.UnixTime > 0 + ? DateTimeOffset.FromUnixTimeSeconds(props.Timestamp.UnixTime) + : (DateTimeOffset?)null; + + var headers = props.Headers.ParseHeaders(); var envelope = new MessageEnvelope { MessageId = props.MessageId, CorrelationId = props.CorrelationId, - ConversationId = props.Headers?.GetString(RabbitMQMessageHeaders.ConversationId), - CausationId = props.Headers?.GetString(RabbitMQMessageHeaders.CausationId), - SourceAddress = props.Headers?.GetString(RabbitMQMessageHeaders.SourceAddress), - DestinationAddress = props.Headers?.GetString(RabbitMQMessageHeaders.DestinationAddress), + ConversationId = headers.GetString(RabbitMQMessageHeaders.ConversationId), + CausationId = headers.GetString(RabbitMQMessageHeaders.CausationId), + SourceAddress = headers.GetString(RabbitMQMessageHeaders.SourceAddress), + DestinationAddress = headers.GetString(RabbitMQMessageHeaders.DestinationAddress), ResponseAddress = props.ReplyTo, - FaultAddress = props.Headers?.GetString(RabbitMQMessageHeaders.FaultAddress), + FaultAddress = headers.GetString(RabbitMQMessageHeaders.FaultAddress), ContentType = props.ContentType, - MessageType = props.Type ?? props.Headers?.GetString(RabbitMQMessageHeaders.MessageType), + MessageType = props.Type ?? headers.GetString(RabbitMQMessageHeaders.MessageType), SentAt = sentAt, - DeliverBy = ParseExpiration(props.Expiration, sentAt), - DeliveryCount = GetDeliveryCount(props.Headers, eventArgs.Redelivered), - Headers = BuildHeaders(props.Headers), - EnclosedMessageTypes = props.Headers?.GetStringArray(RabbitMQMessageHeaders.EnclosedMessageTypes) ?? [], + DeliverBy = props.Expiration.ParseExpiration(sentAt, timeProvider), + DeliveryCount = headers.GetDeliveryCount(eventArgs.Redelivered), + Headers = headers, + EnclosedMessageTypes = headers.GetStringArray(RabbitMQMessageHeaders.EnclosedMessageTypes), Body = eventArgs.Body }; @@ -61,105 +53,81 @@ public MessageEnvelope Parse(BasicDeliverEventArgs eventArgs) } /// - /// Returns the delivery count from the quorum queue x-delivery-count header when - /// available; otherwise falls back to the classic queue Redelivered flag. + /// Shared singleton instance of the parser. /// - private static int GetDeliveryCount(IDictionary? headers, bool redelivered) + public static readonly RabbitMQMessageEnvelopeParser Instance = new(); +} + +file static class Extensions +{ + public static Headers ParseHeaders(this IDictionary? fieldTable) { - if (headers is not null - && headers.TryGetValue("x-delivery-count", out var value) - && value is long count) + if (fieldTable is null || fieldTable.Count == 0) { - return count > int.MaxValue ? int.MaxValue : (int)count; + return Headers.Empty(); } - return redelivered ? 1 : 0; - } - - private static DateTimeOffset? ParseExpiration(string? expiration, DateTimeOffset? sentAt) - { - if (string.IsNullOrEmpty(expiration) || !long.TryParse(expiration, out var ms)) + var headers = new Headers(fieldTable.Count); + foreach (var (key, value) in fieldTable) { - return null; + headers.Set(key, value.FromFieldTableValue(key)); } - // AMQP expiration is a per-message TTL in milliseconds set at publish time. - // Compute deliver-by relative to the send timestamp when available. - var origin = sentAt ?? DateTimeOffset.UtcNow; - return origin.AddMilliseconds(ms); + return headers; } - private static Headers BuildHeaders(IDictionary? headers) + public static string? GetString(this Headers headers, ContextDataKey key) { - if (headers is null || headers.Count == 0) + var value = headers.GetValue(key.Key); + return value is byte[] bytes ? Encoding.UTF8.GetString(bytes) : value?.ToString(); + } + + public static ImmutableArray GetStringArray( + this Headers headers, + ContextDataKey> key) + { + if (headers.GetValue(key.Key) is not object?[] values) { - return Headers.Empty(); + return []; } - var result = new Headers(headers.Count); - foreach (var (key, value) in headers) + var builder = ImmutableArray.CreateBuilder(values.Length); + foreach (var value in values) { - result.Set(key, NormalizeValue(value, key, 0)); + if (value is string text) + { + builder.Add(text); + } + else if (value is byte[] bytes) + { + builder.Add(Encoding.UTF8.GetString(bytes)); + } } - return result; + return builder.ToImmutable(); } - /// - /// Maps AMQP wire types onto the CLR types the envelope serializer understands, at every level of - /// a nested value. Throws when a value is nested deeper than . - /// - private static object? NormalizeValue(object? value, string key, int depth) + public static int GetDeliveryCount(this Headers headers, bool redelivered) { - if (depth > MaxHeaderDepth) + if (headers.TryGet(RabbitMQMessageHeaders.DeliveryCount, out var count)) { - throw new InvalidOperationException( - $"The header '{key}' is nested more than {MaxHeaderDepth} levels deep and cannot be " - + "read as a message header."); + return count > int.MaxValue ? int.MaxValue : (int)count; } - switch (value) + return redelivered ? 1 : 0; + } + + public static DateTimeOffset? ParseExpiration( + this string? expiration, + DateTimeOffset? sentAt, + TimeProvider timeProvider) + { + if (string.IsNullOrEmpty(expiration) || !long.TryParse(expiration, out var milliseconds)) { - // a binary field is kept as bytes without being tested for text - case BinaryTableValue binary: - return binary.Bytes; - - // a long string carries text and binary alike, so content is the only signal - case byte[] bytes: - return Utf8.IsValid(bytes) ? Encoding.UTF8.GetString(bytes) : bytes; - - // a value outside the range a date can express is kept as the number it is - case AmqpTimestamp timestamp: - return timestamp.UnixTime >= s_minTimestampSeconds - && timestamp.UnixTime <= s_maxTimestampSeconds - ? DateTimeOffset.FromUnixTimeSeconds(timestamp.UnixTime) - : timestamp.UnixTime; - - case IDictionary table: - var mappedTable = new Dictionary(table.Count); - foreach (var (entryKey, item) in table) - { - mappedTable[entryKey] = NormalizeValue(item, key, depth + 1); - } - - return mappedTable; - - case IList array: - var mappedArray = new object?[array.Count]; - for (var i = 0; i < array.Count; i++) - { - mappedArray[i] = NormalizeValue(array[i], key, depth + 1); - } - - return mappedArray; - - default: - return value; + return null; } - } - /// - /// Shared singleton instance of the parser. - /// - public static readonly RabbitMQMessageEnvelopeParser Instance = new(); + var origin = sentAt ?? timeProvider.GetUtcNow(); + return origin.AddMilliseconds(milliseconds); + } } diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageHeaders.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageHeaders.cs index c5924bdd01a..15429e02b87 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageHeaders.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageHeaders.cs @@ -1,5 +1,4 @@ using System.Collections.Immutable; -using System.Text; namespace Mocha.Transport.RabbitMQ; @@ -53,52 +52,9 @@ internal static class RabbitMQMessageHeaders /// public static readonly ContextDataKey> EnclosedMessageTypes = new( "x-enclosed-message-types"); -} -/// -/// Extension methods for reading typed values from RabbitMQ message headers. -/// -internal static class RabbitMQMessageHeaderExtensions -{ /// - /// Extracts a string value from the headers dictionary, decoding from UTF-8 bytes if the raw value is a byte array. + /// Header key for the delivery count maintained by RabbitMQ quorum queues. /// - /// The RabbitMQ message headers dictionary. - /// The context data key identifying the header to read. - /// The decoded string value, or null if the header is absent or cannot be converted. - public static string? GetString(this IDictionary headers, ContextDataKey key) - { - if (headers.TryGetValue(key.Key, out var value) && value is byte[] bytes) - { - return Encoding.UTF8.GetString(bytes); - } - - return value?.ToString(); - } - - /// - /// Extracts an immutable array of strings from a header value stored as a List<object> of UTF-8 byte arrays. - /// - /// The RabbitMQ message headers dictionary. - /// The context data key identifying the header to read. - /// An immutable array of decoded strings, or an empty array if the header is absent or not in the expected format. - public static ImmutableArray GetStringArray( - this IDictionary headers, - ContextDataKey> key) - { - if (headers.TryGetValue(key.Key, out var value) && value is List listOfObjects) - { - var builder = ImmutableArray.CreateBuilder(listOfObjects.Count); - foreach (var obj in listOfObjects) - { - if (obj is byte[] bytes) - { - builder.Add(Encoding.UTF8.GetString(bytes)); - } - } - return builder.ToImmutableArray(); - } - - return []; - } + public static readonly ContextDataKey DeliveryCount = new("x-delivery-count"); } diff --git a/src/Mocha/src/Mocha/Serialization/HeaderValueText.cs b/src/Mocha/src/Mocha/Serialization/HeaderValueText.cs deleted file mode 100644 index d51a211b4a2..00000000000 --- a/src/Mocha/src/Mocha/Serialization/HeaderValueText.cs +++ /dev/null @@ -1,27 +0,0 @@ -using System.Globalization; - -namespace Mocha; - -/// -/// The text form a header value takes on the wire, shared by every transport that carries one as text. -/// -internal static class HeaderValueText -{ - /// - /// The string the URI was built from, escapes intact. - /// - public static string From(Uri value) - => value.OriginalString; - - public static string From(TimeSpan value) - => value.ToString("c", CultureInfo.InvariantCulture); - - public static string From(DateOnly value) - => value.ToString("O", CultureInfo.InvariantCulture); - - public static string From(TimeOnly value) - => value.ToString("O", CultureInfo.InvariantCulture); - - public static string From(Enum value) - => value.ToString(); -} diff --git a/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs b/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs index ec94e295fe6..06228cd5ed9 100644 --- a/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs +++ b/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs @@ -1,4 +1,5 @@ using System.Collections; +using System.Globalization; using System.Text.Json; using System.Text.Json.Nodes; using System.Text.Json.Serialization; @@ -9,36 +10,6 @@ namespace Mocha; /// /// JSON converter for serializing and deserializing instances as JSON objects with dynamic value types. /// -/// -/// -/// Header values are weakly typed, but the set of types that can be written is closed: -/// , , , -/// , the numeric types, , -/// , , , -/// , , , an enum, a byte payload -/// (written as base64), a JSON value (, , -/// ), nested headers, a dictionary with string keys, or a sequence of any of -/// these. Through , any other type is rejected with a -/// naming the header; through options carrying a reflection-based -/// resolver, that resolver serializes the value instead. -/// -/// -/// Reading produces a narrower set, because JSON carries no type tag: -/// , , , , -/// , , , a dictionary, or an array, so a -/// value keeps its JSON representation across a round trip but not always its CLR type. A -/// , , , or an -/// enum reads back as the string written, a byte payload as base64 text, a sequence as an array. -/// -/// -/// A number is read by value, not by the type that wrote it, taking the narrowest of -/// , and that holds it: a -/// , or a whole , or -/// , can come back as an . A fractional value reads as a -/// , losing precision, and a whole number beyond -/// reads as a written again in scientific notation. -/// -/// public class HeadersJsonConverter : JsonConverter { /// @@ -49,10 +20,6 @@ public class HeadersJsonConverter : JsonConverter /// /// Gets pre-configured with this converter registered. /// - /// - /// These options resolve metadata for and only, and - /// are read-only. - /// public static readonly JsonSerializerOptions Options = CreateOptions(); private static JsonSerializerOptions CreateOptions() @@ -122,64 +89,6 @@ public override void Write(Utf8JsonWriter writer, IHeaders value, JsonSerializer writer.WriteEndObject(); } - /// - /// Reads a JSON value into the types a header holds, the reading applies to a - /// document. - /// - /// The JSON value to read. - /// The value mapped onto the types a header can hold. - internal static object? ReadHeaderValue(JsonElement element) - { - switch (element.ValueKind) - { - case JsonValueKind.Object: - var table = new Dictionary(); - foreach (var property in element.EnumerateObject()) - { - table[property.Name] = ReadHeaderValue(property.Value); - } - - return table; - - case JsonValueKind.Array: - var items = new List(); - foreach (var item in element.EnumerateArray()) - { - items.Add(ReadHeaderValue(item)); - } - - return items; - - case JsonValueKind.String: - return element.GetString(); - - case JsonValueKind.Number: - if (element.TryGetInt32(out var intValue)) - { - return intValue; - } - if (element.TryGetInt64(out var longValue)) - { - return longValue; - } - if (element.TryGetUInt64(out var unsignedValue)) - { - return unsignedValue; - } - - return element.GetDouble(); - - case JsonValueKind.True: - return true; - - case JsonValueKind.False: - return false; - - default: - return null; - } - } - private static object? ReadValue(ref Utf8JsonReader reader, JsonSerializerOptions options) { switch (reader.TokenType) @@ -193,7 +102,6 @@ public override void Write(Utf8JsonWriter writer, IHeaders value, JsonSerializer case JsonTokenType.False: return false; - // JSON carries no type tag, so text stays text case JsonTokenType.String: return reader.GetString(); @@ -337,7 +245,6 @@ private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSeriali jsonDocument.RootElement.WriteTo(writer); break; - // must precede the collection cases case JsonNode jsonNode: jsonNode.WriteTo(writer); break; @@ -362,30 +269,28 @@ private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSeriali writer.WriteEndObject(); break; - // the writer's own overload formats into the buffer instead of allocating a string case Guid g: writer.WriteStringValue(g); break; - // the text forms every transport shares case TimeSpan t: - writer.WriteStringValue(HeaderValueText.From(t)); + writer.WriteStringValue(t.ToString("c", CultureInfo.InvariantCulture)); break; case Uri u: - writer.WriteStringValue(HeaderValueText.From(u)); + writer.WriteStringValue(u.OriginalString); break; case DateOnly d: - writer.WriteStringValue(HeaderValueText.From(d)); + writer.WriteStringValue(d.ToString("O", CultureInfo.InvariantCulture)); break; case TimeOnly to: - writer.WriteStringValue(HeaderValueText.From(to)); + writer.WriteStringValue(to.ToString("O", CultureInfo.InvariantCulture)); break; case Enum e: - writer.WriteStringValue(HeaderValueText.From(e)); + writer.WriteStringValue(e.ToString()); break; case short s: @@ -416,7 +321,6 @@ private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSeriali writer.WriteStringValue([c]); break; - // int[] and Dictionary have no covariant conversion to object? items case IDictionary dictionary: WriteDictionary(writer, dictionary, options, key); break; @@ -446,14 +350,9 @@ private static void WriteDictionary( foreach (DictionaryEntry entry in dictionary) { - // a custom dictionary can hand back any key, a null included if (entry.Key is not string name) { - var keyType = entry.Key is { } entryKey ? entryKey.GetType().FullName : "null"; - - throw new JsonException( - $"The header '{key}' holds a dictionary keyed by '{keyType}' that cannot be " - + "written as JSON. Use string keys."); + throw ThrowHelper.HeaderDictionaryKeyMustBeString(key, entry.Key); } writer.WritePropertyName(name); @@ -475,11 +374,7 @@ private static void WriteUnknownValue( } catch (NotSupportedException ex) { - throw new JsonException( - $"The header '{key}' holds a value of type '{value.GetType().FullName}' that cannot be " - + "written as JSON. Set the header to a supported value type, or register JSON type " - + "metadata for it.", - ex); + throw ThrowHelper.HeaderValueNotSupported(key, value, ex); } } diff --git a/src/Mocha/src/Mocha/ThrowHelper.cs b/src/Mocha/src/Mocha/ThrowHelper.cs index bdc572171a8..28fc1d94a73 100644 --- a/src/Mocha/src/Mocha/ThrowHelper.cs +++ b/src/Mocha/src/Mocha/ThrowHelper.cs @@ -1,9 +1,29 @@ +using System.Text.Json; using Mocha.Scheduling; namespace Mocha; internal static class ThrowHelper { + public static Exception HeaderDictionaryKeyMustBeString(string headerKey, object? dictionaryKey) + { + var keyType = dictionaryKey?.GetType().FullName ?? "null"; + + return new JsonException( + $"The header '{headerKey}' holds a dictionary keyed by '{keyType}' that cannot be " + + "written as JSON. Use string keys."); + } + + public static Exception HeaderValueNotSupported( + string headerKey, + object value, + NotSupportedException innerException) + => new JsonException( + $"The header '{headerKey}' holds a value of type '{value.GetType().FullName}' that cannot be " + + "written as JSON. Set the header to a supported value type, or register JSON type " + + "metadata for it.", + innerException); + public static Exception BeforeAndAfterConflict() => new ArgumentException( "Only one of 'before' or 'after' can be specified at the same time."); diff --git a/src/Mocha/test/Mocha.Tests/Serialization/HeadersJsonConverterTests.cs b/src/Mocha/test/Mocha.Tests/Serialization/HeadersJsonConverterTests.cs index df42f315ec9..c683ece1e50 100644 --- a/src/Mocha/test/Mocha.Tests/Serialization/HeadersJsonConverterTests.cs +++ b/src/Mocha/test/Mocha.Tests/Serialization/HeadersJsonConverterTests.cs @@ -250,11 +250,7 @@ public void Write_Should_EmitBase64Strings_When_ByteArraysAreNestedInHeader() "x-death", new List { - new Dictionary - { - ["reason"] = "rejected"u8.ToArray(), - ["count"] = 1L - } + new Dictionary { ["reason"] = "rejected"u8.ToArray(), ["count"] = 1L } }); // act @@ -272,8 +268,9 @@ public void Write_Should_ThrowJsonExceptionNamingTheHeader_When_ValueTypeIsUnsup headers.Set("handle", nint.Zero); // act - var exception = Record.Exception( - () => JsonSerializer.Serialize(headers, HeadersJsonConverter.Options)); + var exception = Record.Exception(() => + JsonSerializer.Serialize(headers, HeadersJsonConverter.Options) + ); // assert Assert.Equal( @@ -291,8 +288,9 @@ public void Write_Should_ThrowJsonExceptionNamingTheHeader_When_HeaderIsCustomTy headers.Set("custom", new CustomHeaderDto { Name = "test" }); // act - var exception = Record.Exception( - () => JsonSerializer.Serialize(headers, HeadersJsonConverter.Options)); + var exception = Record.Exception(() => + JsonSerializer.Serialize(headers, HeadersJsonConverter.Options) + ); // assert Assert.Equal( @@ -341,8 +339,9 @@ public void Write_Should_ThrowJsonExceptionNamingTheHeader_When_DictionaryKeyIsN headers.Set("weights", new Dictionary { [1] = "high" }); // act - var exception = Record.Exception( - () => JsonSerializer.Serialize(headers, HeadersJsonConverter.Options)); + var exception = Record.Exception(() => + JsonSerializer.Serialize(headers, HeadersJsonConverter.Options) + ); // assert Assert.Equal( diff --git a/src/Mocha/test/Mocha.Tests/Transport/MessageEnvelopeWriterTests.cs b/src/Mocha/test/Mocha.Tests/Transport/MessageEnvelopeWriterTests.cs index 27212086fda..1efffa11907 100644 --- a/src/Mocha/test/Mocha.Tests/Transport/MessageEnvelopeWriterTests.cs +++ b/src/Mocha/test/Mocha.Tests/Transport/MessageEnvelopeWriterTests.cs @@ -185,7 +185,8 @@ public void WriteMessage_Should_WriteBase64_When_HeaderIsByteArray() writer.Flush(); // assert - Encoding.UTF8.GetString(stream.ToArray()) + Encoding + .UTF8.GetString(stream.ToArray()) .MatchInlineSnapshot("""{"messageId":"msg-008","headers":{"tenant_id":"dGVuYW50LTEyMw=="}}"""); } @@ -216,7 +217,8 @@ public void WriteMessage_Should_WriteNestedValues_When_HeaderIsBrokerDeathTable( writer.Flush(); // assert - Encoding.UTF8.GetString(stream.ToArray()) + Encoding + .UTF8.GetString(stream.ToArray()) .MatchInlineSnapshot( """ {"messageId":"msg-009","headers":{"x-death":[{"queue":"product-queue","reason":"cmVqZWN0ZWQ=","count":1}]}} @@ -754,11 +756,7 @@ public void Roundtrip_Should_PreserveScheduledTime_When_ScheduledTimeIsSet() public void Roundtrip_Should_PreserveNullScheduledTime_When_ScheduledTimeIsNull() { // arrange - var original = new MessageEnvelope - { - MessageId = "msg-024", - MessageType = "urn:message:TestEvent" - }; + var original = new MessageEnvelope { MessageId = "msg-024", MessageType = "urn:message:TestEvent" }; // act - write using var stream = new MemoryStream(); diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/HeaderValueConsistencyTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/HeaderValueConsistencyTests.cs index bb4fad92a27..d19bc5d8919 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/HeaderValueConsistencyTests.cs +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/HeaderValueConsistencyTests.cs @@ -12,18 +12,6 @@ namespace Mocha.Transport.RabbitMQ.Tests; /// public class HeaderValueConsistencyTests { - public static TheoryData TextValues() - => new() - { - { "guid", Guid.Parse("12345678-1234-1234-1234-123456789012") }, - { "char", 'c' }, - { "uri", new Uri("https://example.com/a%20b") }, - { "timespan", TimeSpan.FromMinutes(90) }, - { "dateonly", new DateOnly(2026, 8, 3) }, - { "timeonly", new TimeOnly(10, 30, 15) }, - { "enum", DayOfWeek.Wednesday } - }; - [Theory] [MemberData(nameof(TextValues))] public void BothEncoders_Should_WriteTheSameText_When_ValueTravelsAsText(string key, object value) @@ -34,12 +22,10 @@ public void BothEncoders_Should_WriteTheSameText_When_ValueTravelsAsText(string // act var json = JsonSerializer.Serialize(headers, HeadersJsonConverter.Options); - var table = envelope.BuildHeaders(); + var table = RabbitMQMessageEnvelopeFormatter.Format(envelope, TimeProvider.System).Headers!; // assert - Assert.Equal( - JsonSerializer.Serialize(new Dictionary { [key] = table[key] }), - json); + Assert.Equal(JsonSerializer.Serialize(new Dictionary { [key] = table[key] }), json); } [Theory] @@ -51,11 +37,13 @@ public void BothEncoders_Should_ProduceTheSameReading_When_ValueIsAContainer(str var envelope = new MessageEnvelope { Headers = headers }; // act - var throughJson = JsonSerializer.Deserialize( - JsonSerializer.Serialize(headers, HeadersJsonConverter.Options), - HeadersJsonConverter.Options); - var throughTable = RabbitMQMessageEnvelopeParser.Instance - .Parse(CreateDelivery(envelope.BuildHeaders())); + var throughJson = + JsonSerializer.Deserialize( + JsonSerializer.Serialize(headers, HeadersJsonConverter.Options), + HeadersJsonConverter.Options); + var throughTable = RabbitMQMessageEnvelopeParser.Instance.Parse( + CreateDelivery(RabbitMQMessageEnvelopeFormatter.Format(envelope, TimeProvider.System).Headers!), + TimeProvider.System); // assert throughJson!.TryGetValue(key, out var fromJson); @@ -63,21 +51,6 @@ public void BothEncoders_Should_ProduceTheSameReading_When_ValueIsAContainer(str Assert.Equal(fromJson, fromTable); } - public static TheoryData ShapedValues() - => new() - { - { "dictionary", new Dictionary { ["k"] = "v" } }, - { "sequence", new object?[] { 1, "a" } }, - { "json-object", new JsonObject { ["k"] = 1 } }, - { "json-array", new JsonArray(1, 2) }, - { "json-text", JsonValue.Create("abc")! }, - // cloned so that the element outlives the document it was parsed from - { "json-element", JsonDocument.Parse("""{"k":1}""").RootElement.Clone() }, - { "json-document", JsonDocument.Parse("""[1,2]""") }, - { "nested-headers", new Headers([new HeaderValue { Key = "inner", Value = "v" }]) }, - { "null-inside", new Dictionary { ["k"] = null } } - }; - [Fact] public void BothEncoders_Should_KeepTheHeader_When_ItsValueIsNull() { @@ -86,17 +59,55 @@ public void BothEncoders_Should_KeepTheHeader_When_ItsValueIsNull() var envelope = new MessageEnvelope { Headers = headers }; // act - var throughJson = JsonSerializer.Deserialize( - JsonSerializer.Serialize(headers, HeadersJsonConverter.Options), - HeadersJsonConverter.Options); - var throughTable = RabbitMQMessageEnvelopeParser.Instance - .Parse(CreateDelivery(envelope.BuildHeaders())); + var throughJson = + JsonSerializer.Deserialize( + JsonSerializer.Serialize(headers, HeadersJsonConverter.Options), + HeadersJsonConverter.Options); + var throughTable = RabbitMQMessageEnvelopeParser.Instance.Parse( + CreateDelivery(RabbitMQMessageEnvelopeFormatter.Format(envelope, TimeProvider.System).Headers!), + TimeProvider.System); // assert Assert.True(throughJson!.ContainsKey("x-null")); Assert.True(throughTable.Headers!.ContainsKey("x-null")); } + public static TheoryData TextValues() + => new() + { + { "guid", Guid.Parse("12345678-1234-1234-1234-123456789012") }, + { "char", 'c' }, + { "uri", new Uri("https://example.com/a%20b") }, + { "timespan", TimeSpan.FromMinutes(90) }, + { "dateonly", new DateOnly(2026, 8, 3) }, + { "timeonly", new TimeOnly(10, 30, 15) }, + { "enum", DayOfWeek.Wednesday } + }; + + public static TheoryData ShapedValues() + => new() + { + { + "dictionary", + new Dictionary { ["k"] = "v" } + }, + { "sequence", new object?[] { 1, "a" } }, + { + "json-object", + new JsonObject { ["k"] = 1 } + }, + { "json-array", new JsonArray(1, 2) }, + { "json-text", JsonValue.Create("abc")! }, + // cloned so that the element outlives the document it was parsed from + { "json-element", JsonDocument.Parse("""{"k":1}""").RootElement.Clone() }, + { "json-document", JsonDocument.Parse("""[1,2]""") }, + { "nested-headers", new Headers([new HeaderValue { Key = "inner", Value = "v" }]) }, + { + "null-inside", + new Dictionary { ["k"] = null } + } + }; + private static BasicDeliverEventArgs CreateDelivery(IDictionary headers) { var properties = new BasicProperties { Headers = headers }; diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQDispatchContextExtensionsTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs similarity index 68% rename from src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQDispatchContextExtensionsTests.cs rename to src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs index ec01147c48b..a87d8f25d49 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQDispatchContextExtensionsTests.cs +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs @@ -7,10 +7,53 @@ namespace Mocha.Transport.RabbitMQ.Tests; -public class RabbitMQDispatchContextExtensionsTests +public class RabbitMQMessageEnvelopeFormatterTests { + private static readonly DateTimeOffset s_now = new(2026, 8, 12, 10, 30, 0, TimeSpan.Zero); + private static readonly TimeProvider s_timeProvider = new FixedTimeProvider(s_now); + + [Fact] + public void Format_Should_SetBasicProperties_When_EnvelopeContainsMetadata() + { + // arrange + var envelope = new MessageEnvelope + { + MessageId = "message-1", + CorrelationId = "correlation-1", + ResponseAddress = "reply-queue", + ContentType = "application/json", + MessageType = "OrderCreated" + }; + + // act + var properties = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider); + + // assert + Assert.Equal("message-1", properties.MessageId); + Assert.Equal("correlation-1", properties.CorrelationId); + Assert.Equal("reply-queue", properties.ReplyTo); + Assert.Equal("application/json", properties.ContentType); + Assert.Equal("OrderCreated", properties.Type); + Assert.Equal(DeliveryModes.Persistent, properties.DeliveryMode); + Assert.Equal(s_now.ToUnixTimeSeconds(), properties.Timestamp.UnixTime); + Assert.Null(properties.Expiration); + } + [Fact] - public void BuildHeaders_Should_ConvertDateTimeOffset_When_HeaderContainsDateTimeOffset() + public void Format_Should_SetExpiration_When_EnvelopeHasDeliverBy() + { + // arrange + var envelope = new MessageEnvelope { DeliverBy = s_now.AddSeconds(60) }; + + // act + var properties = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider); + + // assert + Assert.Equal("60000", properties.Expiration); + } + + [Fact] + public void Format_Should_ConvertDateTimeOffset_When_HeaderContainsDateTimeOffset() { // arrange var dto = new DateTimeOffset(2024, 1, 15, 12, 0, 0, TimeSpan.Zero); @@ -20,7 +63,7 @@ public void BuildHeaders_Should_ConvertDateTimeOffset_When_HeaderContainsDateTim }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert var timestamp = Assert.IsType(headers["x-timestamp"]); @@ -28,7 +71,7 @@ public void BuildHeaders_Should_ConvertDateTimeOffset_When_HeaderContainsDateTim } [Fact] - public void BuildHeaders_Should_ConvertDateTime_When_HeaderContainsDateTime() + public void Format_Should_ConvertDateTime_When_HeaderContainsDateTime() { // arrange var dt = new DateTime(2024, 1, 15, 12, 0, 0, DateTimeKind.Utc); @@ -38,7 +81,7 @@ public void BuildHeaders_Should_ConvertDateTime_When_HeaderContainsDateTime() }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert var timestamp = Assert.IsType(headers["x-timestamp"]); @@ -47,7 +90,7 @@ public void BuildHeaders_Should_ConvertDateTime_When_HeaderContainsDateTime() } [Fact] - public void BuildHeaders_Should_ConvertNestedTimestamps_When_HeaderIsDeathTable() + public void Format_Should_ConvertNestedTimestamps_When_HeaderIsDeathTable() { // arrange // the x-death shape the receive side produces, whose nested date has no wire form @@ -70,7 +113,7 @@ public void BuildHeaders_Should_ConvertNestedTimestamps_When_HeaderIsDeathTable( }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal( @@ -84,7 +127,7 @@ public void BuildHeaders_Should_ConvertNestedTimestamps_When_HeaderIsDeathTable( } [Fact] - public void BuildHeaders_Should_ConvertToText_When_HeaderIsUnsignedLong() + public void Format_Should_ConvertToText_When_HeaderIsUnsignedLong() { // arrange // a field table has no unsigned 64 bit type, and a double would round off digits @@ -94,14 +137,14 @@ public void BuildHeaders_Should_ConvertToText_When_HeaderIsUnsignedLong() }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("18446744073709551615", headers["offset"]); } [Fact] - public void BuildHeaders_Should_DeclareBinary_When_HeaderIsABytePayload() + public void Format_Should_DeclareBinary_When_HeaderIsABytePayload() { // arrange // a long string carries text and binary alike, so sending one leaves the receive side guessing @@ -114,7 +157,7 @@ public void BuildHeaders_Should_DeclareBinary_When_HeaderIsABytePayload() }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("COM1"u8.ToArray(), Assert.IsType(headers["x-signature"]).Bytes); @@ -122,7 +165,7 @@ public void BuildHeaders_Should_DeclareBinary_When_HeaderIsABytePayload() } [Fact] - public void BuildHeaders_Should_KeepNullValues_When_HeaderValueIsNull() + public void Format_Should_KeepNullValues_When_HeaderValueIsNull() { // arrange var envelope = new MessageEnvelope @@ -131,7 +174,7 @@ public void BuildHeaders_Should_KeepNullValues_When_HeaderValueIsNull() }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.True(headers.ContainsKey("x-null")); @@ -139,7 +182,7 @@ public void BuildHeaders_Should_KeepNullValues_When_HeaderValueIsNull() } [Fact] - public void BuildHeaders_Should_PassThroughOtherTypes_When_HeaderContainsString() + public void Format_Should_PassThroughOtherTypes_When_HeaderContainsString() { // arrange var envelope = new MessageEnvelope @@ -151,7 +194,7 @@ public void BuildHeaders_Should_PassThroughOtherTypes_When_HeaderContainsString( }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("hello", headers["x-string"]); @@ -159,79 +202,79 @@ public void BuildHeaders_Should_PassThroughOtherTypes_When_HeaderContainsString( } [Fact] - public void BuildHeaders_Should_SetConversationId_When_EnvelopeHasConversationId() + public void Format_Should_SetConversationId_When_EnvelopeHasConversationId() { // arrange var envelope = new MessageEnvelope { ConversationId = "conv-123" }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("conv-123", headers[RabbitMQMessageHeaders.ConversationId.Key]); } [Fact] - public void BuildHeaders_Should_SetCausationId_When_EnvelopeHasCausationId() + public void Format_Should_SetCausationId_When_EnvelopeHasCausationId() { // arrange var envelope = new MessageEnvelope { CausationId = "cause-456" }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("cause-456", headers[RabbitMQMessageHeaders.CausationId.Key]); } [Fact] - public void BuildHeaders_Should_SetSourceAddress_When_EnvelopeHasSourceAddress() + public void Format_Should_SetSourceAddress_When_EnvelopeHasSourceAddress() { // arrange var envelope = new MessageEnvelope { SourceAddress = "rabbitmq:///q/source-q" }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("rabbitmq:///q/source-q", headers[RabbitMQMessageHeaders.SourceAddress.Key]); } [Fact] - public void BuildHeaders_Should_SetDestinationAddress_When_EnvelopeHasDestinationAddress() + public void Format_Should_SetDestinationAddress_When_EnvelopeHasDestinationAddress() { // arrange var envelope = new MessageEnvelope { DestinationAddress = "rabbitmq:///q/dest-q" }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("rabbitmq:///q/dest-q", headers[RabbitMQMessageHeaders.DestinationAddress.Key]); } [Fact] - public void BuildHeaders_Should_SetFaultAddress_When_EnvelopeHasFaultAddress() + public void Format_Should_SetFaultAddress_When_EnvelopeHasFaultAddress() { // arrange var envelope = new MessageEnvelope { FaultAddress = "rabbitmq:///q/fault-q" }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("rabbitmq:///q/fault-q", headers[RabbitMQMessageHeaders.FaultAddress.Key]); } [Fact] - public void BuildHeaders_Should_SetEnclosedMessageTypes_When_EnvelopeHasTypes() + public void Format_Should_SetEnclosedMessageTypes_When_EnvelopeHasTypes() { // arrange var types = ImmutableArray.Create("urn:message:OrderCreated", "urn:message:IEvent"); var envelope = new MessageEnvelope { EnclosedMessageTypes = types }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.True(headers.ContainsKey(RabbitMQMessageHeaders.EnclosedMessageTypes.Key)); @@ -239,33 +282,33 @@ public void BuildHeaders_Should_SetEnclosedMessageTypes_When_EnvelopeHasTypes() } [Fact] - public void BuildHeaders_Should_SetMessageType_When_EnvelopeHasMessageType() + public void Format_Should_SetMessageType_When_EnvelopeHasMessageType() { // arrange var envelope = new MessageEnvelope { MessageType = "urn:message:OrderCreated" }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("urn:message:OrderCreated", headers[RabbitMQMessageHeaders.MessageType.Key]); } [Fact] - public void BuildHeaders_Should_ReturnEmptyHeaders_When_EnvelopeIsEmpty() + public void Format_Should_ReturnEmptyHeaders_When_EnvelopeIsEmpty() { // arrange var envelope = new MessageEnvelope(); // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Empty(headers); } [Fact] - public void BuildHeaders_Should_MapToTableValues_When_TypesHaveNoWireForm() + public void Format_Should_MapToTableValues_When_TypesHaveNoWireForm() { // arrange // every type a field table cannot represent @@ -293,7 +336,7 @@ public void BuildHeaders_Should_MapToTableValues_When_TypesHaveNoWireForm() }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal( @@ -315,7 +358,7 @@ public void BuildHeaders_Should_MapToTableValues_When_TypesHaveNoWireForm() } [Fact] - public void BuildHeaders_Should_ReadUnzonedTimeAsUtc_When_HeaderIsDateTime() + public void Format_Should_ReadUnzonedTimeAsUtc_When_HeaderIsDateTime() { // arrange var envelope = new MessageEnvelope @@ -328,7 +371,7 @@ public void BuildHeaders_Should_ReadUnzonedTimeAsUtc_When_HeaderIsDateTime() }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal( @@ -338,7 +381,7 @@ public void BuildHeaders_Should_ReadUnzonedTimeAsUtc_When_HeaderIsDateTime() } [Fact] - public void BuildHeaders_Should_ProduceTheEpochFloor_When_HeaderIsTheDefaultDateTime() + public void Format_Should_ProduceTheEpochFloor_When_HeaderIsTheDefaultDateTime() { // arrange var envelope = new MessageEnvelope @@ -347,7 +390,7 @@ public void BuildHeaders_Should_ProduceTheEpochFloor_When_HeaderIsTheDefaultDate }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal( @@ -356,7 +399,7 @@ public void BuildHeaders_Should_ProduceTheEpochFloor_When_HeaderIsTheDefaultDate } [Fact] - public void BuildHeaders_Should_CarryDecimalAsText_When_ItExceedsTheTableMantissa() + public void Format_Should_CarryDecimalAsText_When_ItExceedsTheTableMantissa() { // arrange // the table's decimal holds a 32 bit mantissa, so a longer one travels as text @@ -371,7 +414,7 @@ public void BuildHeaders_Should_CarryDecimalAsText_When_ItExceedsTheTableMantiss }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal( @@ -385,7 +428,7 @@ public void BuildHeaders_Should_CarryDecimalAsText_When_ItExceedsTheTableMantiss } [Fact] - public void BuildHeaders_Should_MapJsonToTableShapes_When_HeaderIsAJsonValue() + public void Format_Should_MapJsonToTableShapes_When_HeaderIsAJsonValue() { // arrange // writing the JSON text instead would hand the consumer a string holding JSON, quotes and all @@ -401,7 +444,7 @@ public void BuildHeaders_Should_MapJsonToTableShapes_When_HeaderIsAJsonValue() }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal( @@ -416,7 +459,7 @@ public void BuildHeaders_Should_MapJsonToTableShapes_When_HeaderIsAJsonValue() } [Fact] - public void BuildHeaders_Should_KeepEscapes_When_HeaderIsAUri() + public void Format_Should_KeepEscapes_When_HeaderIsAUri() { // arrange // ToString would decode %20 to a space, leaving text that no longer reads back as this URI @@ -429,14 +472,14 @@ public void BuildHeaders_Should_KeepEscapes_When_HeaderIsAUri() }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("https://example.com/a%20b", headers["callback"]); } [Fact] - public void BuildHeaders_Should_MapSiblingValues_When_ATableKeyIsNotText() + public void Format_Should_MapSiblingValues_When_ATableKeyIsNotText() { // arrange var envelope = new MessageEnvelope @@ -456,7 +499,7 @@ public void BuildHeaders_Should_MapSiblingValues_When_ATableKeyIsNotText() }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal( @@ -465,7 +508,7 @@ public void BuildHeaders_Should_MapSiblingValues_When_ATableKeyIsNotText() } [Fact] - public void BuildHeaders_Should_DeclareBinary_When_HeaderIsAByteMemory() + public void Format_Should_DeclareBinary_When_HeaderIsAByteMemory() { // arrange var envelope = new MessageEnvelope @@ -479,7 +522,7 @@ public void BuildHeaders_Should_DeclareBinary_When_HeaderIsAByteMemory() }; // act - var headers = envelope.BuildHeaders(); + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert Assert.Equal("hi"u8.ToArray(), Assert.IsType(headers["readonly"]).Bytes); @@ -488,7 +531,7 @@ public void BuildHeaders_Should_DeclareBinary_When_HeaderIsAByteMemory() } [Fact] - public void BuildHeaders_Should_ThrowNamingTheHeader_When_AValueContainsItself() + public void Format_Should_ThrowNamingTheHeader_When_AValueContainsItself() { // arrange var cyclic = new List(); @@ -499,12 +542,11 @@ public void BuildHeaders_Should_ThrowNamingTheHeader_When_AValueContainsItself() }; // act - var exception = Record.Exception(() => envelope.BuildHeaders()); + var exception = Record.Exception(() => RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!); // assert Assert.Equal( - "The header 'x-cycle' is nested more than 64 levels deep, or holds a value that " - + "contains itself, and cannot be written as an AMQP field table.", + "The header 'x-cycle' exceeds the maximum AMQP field-table nesting depth of 64.", Assert.IsType(exception).Message); } @@ -513,4 +555,9 @@ private enum TestPriority Low, High } + + private sealed class FixedTimeProvider(DateTimeOffset utcNow) : TimeProvider + { + public override DateTimeOffset GetUtcNow() => utcNow; + } } diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs index aea50812495..cb332824ed0 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs @@ -15,7 +15,7 @@ public void Parse_Should_ExtractMessageId_When_PropertySet() var args = CreateDeliverEventArgs(props => props.MessageId = "msg-123"); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal("msg-123", envelope.MessageId); @@ -28,7 +28,7 @@ public void Parse_Should_ExtractCorrelationId_When_PropertySet() var args = CreateDeliverEventArgs(props => props.CorrelationId = "corr-456"); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal("corr-456", envelope.CorrelationId); @@ -41,7 +41,7 @@ public void Parse_Should_ExtractResponseAddress_When_ReplyToSet() var args = CreateDeliverEventArgs(props => props.ReplyTo = "reply-queue"); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal("reply-queue", envelope.ResponseAddress); @@ -54,7 +54,7 @@ public void Parse_Should_ExtractContentType_When_PropertySet() var args = CreateDeliverEventArgs(props => props.ContentType = "application/json"); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal("application/json", envelope.ContentType); @@ -67,7 +67,7 @@ public void Parse_Should_ExtractMessageType_When_TypePropertySet() var args = CreateDeliverEventArgs(props => props.Type = "OrderCreated"); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal("OrderCreated", envelope.MessageType); @@ -80,7 +80,7 @@ public void Parse_Should_FallbackToHeader_When_TypePropertyNull() var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary { ["x-message-type"] = "FallbackType"u8.ToArray() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal("FallbackType", envelope.MessageType); @@ -93,7 +93,7 @@ public void Parse_Should_ExtractTimestamp_When_UnixTimePositive() var args = CreateDeliverEventArgs(props => props.Timestamp = new AmqpTimestamp(1700000000)); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.NotNull(envelope.SentAt); @@ -107,7 +107,7 @@ public void Parse_Should_ReturnNullSentAt_When_TimestampZero() var args = CreateDeliverEventArgs(props => props.Timestamp = new AmqpTimestamp(0)); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Null(envelope.SentAt); @@ -121,7 +121,7 @@ public void Parse_Should_ExtractBody_When_BodyPresent() var args = CreateDeliverEventArgs(body: body); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal(body, envelope.Body.ToArray()); @@ -134,7 +134,7 @@ public void Parse_Should_ExtractCustomHeaders_When_HeadersPresent() var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary { ["x-custom"] = "custom-value"u8.ToArray() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.True(envelope.Headers!.TryGetValue("x-custom", out var value)); @@ -154,7 +154,7 @@ public void Parse_Should_DecodeByteArrayHeaders_When_Utf8Bytes() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal("rabbitmq:///q/source", envelope.SourceAddress); @@ -171,7 +171,7 @@ public void Parse_Should_KeepBytes_When_HeaderIsBinaryTableValue() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.True(envelope.Headers!.TryGetValue("x-signature", out var signature)); @@ -191,7 +191,7 @@ public void Parse_Should_KeepBytes_When_BinaryTableValueIsNestedInHeader() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert envelope.Headers!.TryGetValue("x-origin", out var origin); @@ -209,10 +209,11 @@ public void Parse_Should_KeepBytes_When_AByteHeaderIsPublishedAndReadBack() { Headers = new Headers([new HeaderValue { Key = "x-signature", Value = payload }]) }; - var args = CreateDeliverEventArgs(props => props.Headers = published.BuildHeaders()); + var args = CreateDeliverEventArgs(props => + props.Headers = RabbitMQMessageEnvelopeFormatter.Format(published, TimeProvider.System).Headers); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert envelope.Headers!.TryGetValue("x-signature", out var signature); @@ -235,12 +236,11 @@ public void Parse_Should_ThrowNamingTheHeader_When_HeaderIsNestedTooDeeply() }); // act - var exception = Record.Exception(() => _parser.Parse(args)); + var exception = Record.Exception(() => _parser.Parse(args, TimeProvider.System)); // assert Assert.Equal( - "The header 'x-deep' is nested more than 64 levels deep and cannot be read as a message " - + "header.", + "The header 'x-deep' exceeds the maximum AMQP field-table nesting depth of 64.", Assert.IsType(exception).Message); } @@ -251,7 +251,7 @@ public void Parse_Should_ConvertAmqpTimestamp_When_HeaderIsTimestamp() var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary { ["x-timestamp"] = new AmqpTimestamp(1700000000) }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.True(envelope.Headers!.TryGetValue("x-timestamp", out var ts)); @@ -269,7 +269,7 @@ public void Parse_Should_KeepTheNumber_When_TimestampIsOutOfRange() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert envelope.Headers!.TryGetValue("x-millis", out var millis); @@ -298,7 +298,7 @@ public void Parse_Should_DecodeNestedByteArrays_When_HeaderIsDeathTable() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert envelope.Headers!.TryGetValue("x-death", out var death); @@ -325,7 +325,7 @@ public void Parse_Should_KeepBytes_When_HeaderValueIsNotValidUtf8() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.True(envelope.Headers!.TryGetValue("x-binary", out var parsed)); @@ -339,7 +339,7 @@ public void Parse_Should_SetDeliveryCountOne_When_Redelivered() var args = CreateDeliverEventArgs(redelivered: true); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal(1, envelope.DeliveryCount); @@ -352,7 +352,7 @@ public void Parse_Should_SetDeliveryCountZero_When_NotRedelivered() var args = CreateDeliverEventArgs(redelivered: false); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal(0, envelope.DeliveryCount); @@ -370,7 +370,7 @@ public void Parse_Should_UseXDeliveryCount_When_QuorumQueueHeaderPresent() redelivered: true); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert - exact count from header, not the boolean fallback Assert.Equal(3, envelope.DeliveryCount); @@ -388,7 +388,7 @@ public void Parse_Should_UseXDeliveryCount_When_ValueIsZero() redelivered: false); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal(0, envelope.DeliveryCount); @@ -406,7 +406,7 @@ public void Parse_Should_FallbackToRedelivered_When_XDeliveryCountAbsent() redelivered: true); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert - falls back to redelivered flag Assert.Equal(1, envelope.DeliveryCount); @@ -425,7 +425,7 @@ public void Parse_Should_ExtractEnclosedMessageTypes_When_HeaderPresent() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.NotNull(envelope.EnclosedMessageTypes); @@ -447,7 +447,7 @@ public void Parse_Should_ComputeDeliverByFromSentAt_When_TimestampAndExpirationP }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.NotNull(envelope.DeliverBy); @@ -464,9 +464,9 @@ public void Parse_Should_FallbackToUtcNow_When_ExpirationPresentButNoTimestamp() }); // act - var beforeParse = DateTimeOffset.UtcNow; - var envelope = _parser.Parse(args); - var afterParse = DateTimeOffset.UtcNow; + var beforeParse = TimeProvider.System.GetUtcNow(); + var envelope = _parser.Parse(args, TimeProvider.System); + var afterParse = TimeProvider.System.GetUtcNow(); // assert Assert.NotNull(envelope.DeliverBy); @@ -482,7 +482,7 @@ public void Parse_Should_ReturnNullDeliverBy_When_ExpirationEmpty() var args = CreateDeliverEventArgs(props => props.Expiration = null); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Null(envelope.DeliverBy); @@ -495,7 +495,7 @@ public void Parse_Should_ExtractConversationId_When_HeaderPresent() var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary { ["x-conversation-id"] = "conv-789"u8.ToArray() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal("conv-789", envelope.ConversationId); @@ -508,7 +508,7 @@ public void Parse_Should_ExtractCausationId_When_HeaderPresent() var args = CreateDeliverEventArgs(props => props.Headers = new Dictionary { ["x-causation-id"] = "cause-101"u8.ToArray() }); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.Equal("cause-101", envelope.CausationId); @@ -521,7 +521,7 @@ public void Parse_Should_ReturnEmptyHeaders_When_NoHeaders() var args = CreateDeliverEventArgs(); // act - var envelope = _parser.Parse(args); + var envelope = _parser.Parse(args, TimeProvider.System); // assert Assert.NotNull(envelope.Headers); diff --git a/website/content/docs/mocha/messages.md b/website/content/docs/mocha/messages.md index 35dd7c92f2e..b74bce5c01c 100644 --- a/website/content/docs/mocha/messages.md +++ b/website/content/docs/mocha/messages.md @@ -73,69 +73,32 @@ await bus.SendAsync( The bus merges your headers into the envelope's header collection before dispatching. -## What a header value can hold - -A header value is weakly typed, but the set of types Mocha carries is closed: `null`, `bool`, -`string`, `char`, the numeric types, `DateTime`, `DateTimeOffset`, `DateOnly`, `TimeOnly`, -`TimeSpan`, `Guid`, `Uri`, an enum, a byte payload (`byte[]`, `ArraySegment`, -`ReadOnlyMemory`, `Memory`), a JSON value (`JsonElement`, `JsonDocument`, `JsonNode`), -nested headers, a dictionary with string keys, or a sequence of any of these. - -The envelope serializer rejects any other type with an error naming the header. Setting a value always -succeeds; a transport that encodes headers itself applies its own rules, described at the end of this -section. - -A value keeps its representation across a hop but not always its CLR type, because neither JSON nor an -AMQP field table carries a .NET type tag. Through the envelope serializer, which the outbox and the -Entity Framework Core scheduled store use: - -| Written as | Read back as | -| ------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------- | -| `null`, `bool`, `string` | unchanged | -| any integer type | `int`, `long` or `ulong`, the smallest of those holding the value | -| `double`, `float`, `decimal` | the same integer types when the value is whole, else `double` | -| `DateTime`, `DateTimeOffset`, `Guid`, `Uri`, `TimeSpan`, `DateOnly`, `TimeOnly`, `char`, an enum | the text that was written | -| a byte payload | base64 text | -| a JSON object, nested headers, a dictionary | `Dictionary` | -| a JSON array, any sequence | `object?[]` | - -A number is read by value, not by the type that wrote it, so `5L` and `2.0` both come back as `int`, -and a `uint` above `int.MaxValue` as `long`. Two cases lose data: a whole number beyond -`ulong.MaxValue` is read as a `double` and rewritten in scientific notation, and a `decimal` carrying -more precision than a `double` holds loses the difference. - -A date-shaped string stays a string. Parse explicitly when you need a typed value: - -```csharp -using System.Globalization; - -if (context.Headers.TryGetValue("x-processed-at", out var value) - && DateTimeOffset.TryParse(value as string, CultureInfo.InvariantCulture, out var processedAt)) -{ - // ... -} -``` - -A transport that defines its own header encoding differs in the details. - -**RabbitMQ** maps a header onto an AMQP field table. A date arrives back as a `DateTimeOffset` -truncated to whole seconds, and a `DateTime` carrying no time zone is read as UTC. A byte payload -arrives back as bytes, carried by the table's own byte array type, and a JSON value arrives in the -shapes listed above. The table has no unsigned 64-bit type and only a 32-bit decimal mantissa, so a -number above `long.MaxValue` and a longer `decimal` travel as text. A long string from another -publisher is decoded as text when it holds valid UTF-8. - -**Postgres** reads every number back as a `double`, and drops a header whose value is a dictionary, a -sequence, or `null`. Messages scheduled on this transport are stored with the Postgres encoding rather -than through the envelope serializer, so the table above does not describe them. - -Those typed arrivals hold for a message taken straight off the broker. One that passes through the -outbox, or that the scheduled store redelivers, is rehydrated by the envelope serializer on the way, -so the same date arrives as text and the same byte payload as base64. - -> Keep header values simple. Text and booleans behave the same everywhere, and whole numbers behave -> the same anywhere except Postgres; everything else is worth a round-trip test on the transport you -> actually deploy. +## Message headers + +A header value must map to one of these portable representations: `null`, a boolean, a string, a +number, an object with string keys, or an array. You can set headers with the following CLR values; +Mocha converts them as shown: + +| Value supplied | Header value | +| ---------------------------------------------------------------------- | ---------------------------------------------------- | +| `null`, `bool`, `string` | `null`, boolean, or string | +| `sbyte`, `byte`, `short`, `ushort`, `int`, `uint`, `long`, `ulong` | integer number | +| `float`, `double`, `decimal` | number | +| `char` | one-character string | +| `Guid` | GUID string | +| `Uri` | original URI string | +| an enum | member name | +| `DateTime`, `DateTimeOffset`, `DateOnly`, `TimeOnly` | ISO 8601 string | +| `TimeSpan` | constant-format (`c`) string | +| `byte[]`, `ArraySegment`, `ReadOnlyMemory`, `Memory` | base64 string | +| a JSON scalar | the corresponding `null`, boolean, string, or number | +| a JSON object, nested headers, or a dictionary with string keys | object whose values are mapped recursively | +| a JSON array or sequence | array whose values are mapped recursively | + +Dictionary keys must be strings, and every value in a dictionary or sequence must itself be a +supported header value. Use a string when the exact text representation matters, such as for a +high-precision decimal or an application-specific identifier. Other CLR types are not supported as +header values. ## Access envelope metadata in a handler From d5e34c812642744606345de3a559e142d3404add Mon Sep 17 00:00:00 2001 From: Pascal Senn Date: Wed, 12 Aug 2026 12:03:30 +0200 Subject: [PATCH 7/8] clenaup --- .../Extensions/DateTimeExtensions.cs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/Extensions/DateTimeExtensions.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/Extensions/DateTimeExtensions.cs index 18288035863..e6e27c64fdb 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/Extensions/DateTimeExtensions.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/Extensions/DateTimeExtensions.cs @@ -6,13 +6,12 @@ internal static class DateTimeExtensions { extension(DateTimeOffset value) { - internal AmqpTimestamp ToAmqpTimestamp() - => new(value.ToUnixTimeSeconds()); + public AmqpTimestamp ToAmqpTimestamp() => new(value.ToUnixTimeSeconds()); } extension(DateTime value) { - internal AmqpTimestamp ToAmqpTimestamp() + public AmqpTimestamp ToAmqpTimestamp() { var utc = value.Kind == DateTimeKind.Local ? value.ToUniversalTime() : value; From be18b9e56d152646c5da51aec7cd7ee94d9c6d73 Mon Sep 17 00:00:00 2001 From: Pascal Senn Date: Wed, 12 Aug 2026 12:15:14 +0200 Subject: [PATCH 8/8] Use fallback time provider for RabbitMQ dispatch --- .../src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs index bd5010e01e5..1e03bc24122 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs @@ -1,4 +1,3 @@ -using Microsoft.Extensions.DependencyInjection; using Mocha.Middlewares; using RabbitMQ.Client; using static System.StringSplitOptions; @@ -32,7 +31,7 @@ protected override async ValueTask DispatchAsync(IDispatchContext context) var dispatcher = transport.Dispatcher; var cancellationToken = context.CancellationToken; - var timeProvider = context.Services.GetRequiredService(); + var timeProvider = context.Services.GetTimeProvider(); var channel = await dispatcher.RentChannelAsync(cancellationToken); try {