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..e6e27c64fdb --- /dev/null +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/Extensions/DateTimeExtensions.cs @@ -0,0 +1,21 @@ +using RabbitMQ.Client; + +namespace Mocha.Transport.RabbitMQ; + +internal static class DateTimeExtensions +{ + extension(DateTimeOffset value) + { + public AmqpTimestamp ToAmqpTimestamp() => new(value.ToUnixTimeSeconds()); + } + + extension(DateTime value) + { + public 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 d2d2844bda5..1e03bc24122 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQDispatchEndpoint.cs @@ -31,11 +31,12 @@ protected override async ValueTask DispatchAsync(IDispatchContext context) var dispatcher = transport.Dispatcher; var cancellationToken = context.CancellationToken; + var timeProvider = context.Services.GetTimeProvider(); var channel = await dispatcher.RentChannelAsync(cancellationToken); try { await EnsureProvisionedAsync(channel, cancellationToken); - await DispatchAsync(channel, envelope, cancellationToken); + await DispatchAsync(channel, envelope, timeProvider, cancellationToken); } finally { @@ -46,6 +47,7 @@ protected override async ValueTask DispatchAsync(IDispatchContext context) private async ValueTask DispatchAsync( IChannel channel, MessageEnvelope envelope, + TimeProvider timeProvider, CancellationToken cancellationToken) { var exchangeName = CachedString.Empty; @@ -121,24 +123,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); } @@ -201,78 +186,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) - { - 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; - } - } - } - - 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; - } -} 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 9429d3b1b7b..e280a22c14e 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs @@ -1,3 +1,4 @@ +using System.Collections.Immutable; using System.Text; using Mocha.Middlewares; using RabbitMQ.Client; @@ -16,31 +17,35 @@ internal sealed class RabbitMQMessageEnvelopeParser /// 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 }; @@ -48,59 +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) { - var strValue = value switch + if (value is string text) + { + builder.Add(text); + } + else if (value is byte[] bytes) { - byte[] bytes => Encoding.UTF8.GetString(bytes), - AmqpTimestamp timestamp => DateTimeOffset.FromUnixTimeSeconds(timestamp.UnixTime), - _ => value - }; + builder.Add(Encoding.UTF8.GetString(bytes)); + } + } + + return builder.ToImmutable(); + } - result.Set(key, strValue); + public static int GetDeliveryCount(this Headers headers, bool redelivered) + { + if (headers.TryGet(RabbitMQMessageHeaders.DeliveryCount, out var count)) + { + return count > int.MaxValue ? int.MaxValue : (int)count; } - return result; + return redelivered ? 1 : 0; } - /// - /// Shared singleton instance of the parser. - /// - public static readonly RabbitMQMessageEnvelopeParser Instance = new(); + public static DateTimeOffset? ParseExpiration( + this string? expiration, + DateTimeOffset? sentAt, + TimeProvider timeProvider) + { + if (string.IsNullOrEmpty(expiration) || !long.TryParse(expiration, out var milliseconds)) + { + return null; + } + + 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/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/HeadersConverter.cs b/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs index 03e7ec4e63f..06228cd5ed9 100644 --- a/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs +++ b/src/Mocha/src/Mocha/Serialization/HeadersConverter.cs @@ -1,6 +1,9 @@ +using System.Collections; using System.Globalization; using System.Text.Json; +using System.Text.Json.Nodes; using System.Text.Json.Serialization; +using System.Text.Json.Serialization.Metadata; namespace Mocha; @@ -17,7 +20,23 @@ public class HeadersJsonConverter : JsonConverter /// /// Gets pre-configured with this converter registered. /// - public static readonly JsonSerializerOptions Options = new() { Converters = { Instance } }; + 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,7 +83,7 @@ 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(); @@ -84,10 +103,6 @@ public override void Write(Utf8JsonWriter writer, IHeaders value, JsonSerializer return false; case JsonTokenType.String: - if (reader.TryGetDateTime(out var dateTime)) - { - return dateTime; - } return reader.GetString(); case JsonTokenType.Number: @@ -99,6 +114,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 +177,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 +193,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,31 +245,26 @@ private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSeriali jsonDocument.RootElement.WriteTo(writer); break; + 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; @@ -248,7 +278,7 @@ private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSeriali break; case Uri u: - writer.WriteStringValue(u.ToString()); + writer.WriteStringValue(u.OriginalString); break; case DateOnly d: @@ -291,9 +321,82 @@ private static void WriteValue(Utf8JsonWriter writer, object? value, JsonSeriali writer.WriteStringValue([c]); break; + 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) + { + if (entry.Key is not string name) + { + throw ThrowHelper.HeaderDictionaryKeyMustBeString(key, entry.Key); + } + + 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 ThrowHelper.HeaderValueNotSupported(key, value, 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/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.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..c683ece1e50 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,283 @@ 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..1efffa11907 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,62 @@ 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 +558,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() { @@ -677,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.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..d19bc5d8919 --- /dev/null +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/HeaderValueConsistencyTests.cs @@ -0,0 +1,124 @@ +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 +{ + [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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, TimeProvider.System).Headers!; + + // 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(RabbitMQMessageEnvelopeFormatter.Format(envelope, TimeProvider.System).Headers!), + TimeProvider.System); + + // assert + throughJson!.TryGetValue(key, out var fromJson); + throughTable.Headers!.TryGetValue(key, out var fromTable); + Assert.Equal(fromJson, fromTable); + } + + [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(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 }; + + 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 deleted file mode 100644 index 9a23b622598..00000000000 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQDispatchContextExtensionsTests.cs +++ /dev/null @@ -1,187 +0,0 @@ -using System.Collections.Immutable; -using Mocha.Middlewares; -using RabbitMQ.Client; - -namespace Mocha.Transport.RabbitMQ.Tests; - -public class RabbitMQDispatchContextExtensionsTests -{ - [Fact] - public void BuildHeaders_Should_ConvertDateTimeOffset_When_HeaderContainsDateTimeOffset() - { - // arrange - var dto = new DateTimeOffset(2024, 1, 15, 12, 0, 0, TimeSpan.Zero); - var envelope = new MessageEnvelope - { - Headers = new Headers([new HeaderValue { Key = "x-timestamp", Value = dto }]) - }; - - // act - var headers = envelope.BuildHeaders(); - - // assert - var timestamp = Assert.IsType(headers["x-timestamp"]); - Assert.Equal(dto.ToUnixTimeSeconds(), timestamp.UnixTime); - } - - [Fact] - public void BuildHeaders_Should_ConvertDateTime_When_HeaderContainsDateTime() - { - // arrange - var dt = new DateTime(2024, 1, 15, 12, 0, 0, DateTimeKind.Utc); - var envelope = new MessageEnvelope - { - Headers = new Headers([new HeaderValue { Key = "x-timestamp", Value = dt }]) - }; - - // act - var headers = envelope.BuildHeaders(); - - // assert - var timestamp = Assert.IsType(headers["x-timestamp"]); - var expected = new DateTimeOffset(dt).ToUnixTimeSeconds(); - Assert.Equal(expected, timestamp.UnixTime); - } - - [Fact] - public void BuildHeaders_Should_SkipNullValues_When_HeaderValueIsNull() - { - // arrange - var envelope = new MessageEnvelope - { - Headers = new Headers([new HeaderValue { Key = "x-null", Value = null }]) - }; - - // act - var headers = envelope.BuildHeaders(); - - // assert - Assert.False(headers.ContainsKey("x-null")); - } - - [Fact] - public void BuildHeaders_Should_PassThroughOtherTypes_When_HeaderContainsString() - { - // arrange - var envelope = new MessageEnvelope - { - Headers = new Headers([ - new HeaderValue { Key = "x-string", Value = "hello" }, - new HeaderValue { Key = "x-int", Value = 42 } - ]) - }; - - // act - var headers = envelope.BuildHeaders(); - - // assert - Assert.Equal("hello", headers["x-string"]); - Assert.Equal(42, headers["x-int"]); - } - - [Fact] - public void BuildHeaders_Should_SetConversationId_When_EnvelopeHasConversationId() - { - // arrange - var envelope = new MessageEnvelope { ConversationId = "conv-123" }; - - // act - var headers = envelope.BuildHeaders(); - - // assert - Assert.Equal("conv-123", headers[RabbitMQMessageHeaders.ConversationId.Key]); - } - - [Fact] - public void BuildHeaders_Should_SetCausationId_When_EnvelopeHasCausationId() - { - // arrange - var envelope = new MessageEnvelope { CausationId = "cause-456" }; - - // act - var headers = envelope.BuildHeaders(); - - // assert - Assert.Equal("cause-456", headers[RabbitMQMessageHeaders.CausationId.Key]); - } - - [Fact] - public void BuildHeaders_Should_SetSourceAddress_When_EnvelopeHasSourceAddress() - { - // arrange - var envelope = new MessageEnvelope { SourceAddress = "rabbitmq:///q/source-q" }; - - // act - var headers = envelope.BuildHeaders(); - - // assert - Assert.Equal("rabbitmq:///q/source-q", headers[RabbitMQMessageHeaders.SourceAddress.Key]); - } - - [Fact] - public void BuildHeaders_Should_SetDestinationAddress_When_EnvelopeHasDestinationAddress() - { - // arrange - var envelope = new MessageEnvelope { DestinationAddress = "rabbitmq:///q/dest-q" }; - - // act - var headers = envelope.BuildHeaders(); - - // assert - Assert.Equal("rabbitmq:///q/dest-q", headers[RabbitMQMessageHeaders.DestinationAddress.Key]); - } - - [Fact] - public void BuildHeaders_Should_SetFaultAddress_When_EnvelopeHasFaultAddress() - { - // arrange - var envelope = new MessageEnvelope { FaultAddress = "rabbitmq:///q/fault-q" }; - - // act - var headers = envelope.BuildHeaders(); - - // assert - Assert.Equal("rabbitmq:///q/fault-q", headers[RabbitMQMessageHeaders.FaultAddress.Key]); - } - - [Fact] - public void BuildHeaders_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(); - - // assert - Assert.True(headers.ContainsKey(RabbitMQMessageHeaders.EnclosedMessageTypes.Key)); - Assert.Equal(types, headers[RabbitMQMessageHeaders.EnclosedMessageTypes.Key]); - } - - [Fact] - public void BuildHeaders_Should_SetMessageType_When_EnvelopeHasMessageType() - { - // arrange - var envelope = new MessageEnvelope { MessageType = "urn:message:OrderCreated" }; - - // act - var headers = envelope.BuildHeaders(); - - // assert - Assert.Equal("urn:message:OrderCreated", headers[RabbitMQMessageHeaders.MessageType.Key]); - } - - [Fact] - public void BuildHeaders_Should_ReturnEmptyHeaders_When_EnvelopeIsEmpty() - { - // arrange - var envelope = new MessageEnvelope(); - - // act - var headers = envelope.BuildHeaders(); - - // assert - Assert.Empty(headers); - } -} diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs new file mode 100644 index 00000000000..a87d8f25d49 --- /dev/null +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs @@ -0,0 +1,563 @@ +using System.Collections; +using System.Collections.Immutable; +using System.Text.Json; +using System.Text.Json.Nodes; +using Mocha.Middlewares; +using RabbitMQ.Client; + +namespace Mocha.Transport.RabbitMQ.Tests; + +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 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); + var envelope = new MessageEnvelope + { + Headers = new Headers([new HeaderValue { Key = "x-timestamp", Value = dto }]) + }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + var timestamp = Assert.IsType(headers["x-timestamp"]); + Assert.Equal(dto.ToUnixTimeSeconds(), timestamp.UnixTime); + } + + [Fact] + public void Format_Should_ConvertDateTime_When_HeaderContainsDateTime() + { + // arrange + var dt = new DateTime(2024, 1, 15, 12, 0, 0, DateTimeKind.Utc); + var envelope = new MessageEnvelope + { + Headers = new Headers([new HeaderValue { Key = "x-timestamp", Value = dt }]) + }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + var timestamp = Assert.IsType(headers["x-timestamp"]); + var expected = new DateTimeOffset(dt).ToUnixTimeSeconds(); + Assert.Equal(expected, timestamp.UnixTime); + } + + [Fact] + public void Format_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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal( + new Dictionary + { + ["queue"] = "product-queue", + ["time"] = new AmqpTimestamp(1700000000) + }, + Assert.IsType>( + Assert.Single(Assert.IsType>(headers["x-death"])))); + } + + [Fact] + public void Format_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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal("18446744073709551615", headers["offset"]); + } + + [Fact] + public void Format_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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // 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 Format_Should_KeepNullValues_When_HeaderValueIsNull() + { + // arrange + var envelope = new MessageEnvelope + { + Headers = new Headers([new HeaderValue { Key = "x-null", Value = null }]) + }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.True(headers.ContainsKey("x-null")); + Assert.Null(headers["x-null"]); + } + + [Fact] + public void Format_Should_PassThroughOtherTypes_When_HeaderContainsString() + { + // arrange + var envelope = new MessageEnvelope + { + Headers = new Headers([ + new HeaderValue { Key = "x-string", Value = "hello" }, + new HeaderValue { Key = "x-int", Value = 42 } + ]) + }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal("hello", headers["x-string"]); + Assert.Equal(42, headers["x-int"]); + } + + [Fact] + public void Format_Should_SetConversationId_When_EnvelopeHasConversationId() + { + // arrange + var envelope = new MessageEnvelope { ConversationId = "conv-123" }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal("conv-123", headers[RabbitMQMessageHeaders.ConversationId.Key]); + } + + [Fact] + public void Format_Should_SetCausationId_When_EnvelopeHasCausationId() + { + // arrange + var envelope = new MessageEnvelope { CausationId = "cause-456" }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal("cause-456", headers[RabbitMQMessageHeaders.CausationId.Key]); + } + + [Fact] + public void Format_Should_SetSourceAddress_When_EnvelopeHasSourceAddress() + { + // arrange + var envelope = new MessageEnvelope { SourceAddress = "rabbitmq:///q/source-q" }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal("rabbitmq:///q/source-q", headers[RabbitMQMessageHeaders.SourceAddress.Key]); + } + + [Fact] + public void Format_Should_SetDestinationAddress_When_EnvelopeHasDestinationAddress() + { + // arrange + var envelope = new MessageEnvelope { DestinationAddress = "rabbitmq:///q/dest-q" }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal("rabbitmq:///q/dest-q", headers[RabbitMQMessageHeaders.DestinationAddress.Key]); + } + + [Fact] + public void Format_Should_SetFaultAddress_When_EnvelopeHasFaultAddress() + { + // arrange + var envelope = new MessageEnvelope { FaultAddress = "rabbitmq:///q/fault-q" }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal("rabbitmq:///q/fault-q", headers[RabbitMQMessageHeaders.FaultAddress.Key]); + } + + [Fact] + 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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.True(headers.ContainsKey(RabbitMQMessageHeaders.EnclosedMessageTypes.Key)); + Assert.Equal(types, headers[RabbitMQMessageHeaders.EnclosedMessageTypes.Key]); + } + + [Fact] + public void Format_Should_SetMessageType_When_EnvelopeHasMessageType() + { + // arrange + var envelope = new MessageEnvelope { MessageType = "urn:message:OrderCreated" }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal("urn:message:OrderCreated", headers[RabbitMQMessageHeaders.MessageType.Key]); + } + + [Fact] + public void Format_Should_ReturnEmptyHeaders_When_EnvelopeIsEmpty() + { + // arrange + var envelope = new MessageEnvelope(); + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Empty(headers); + } + + [Fact] + public void Format_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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // 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 Format_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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal( + new AmqpTimestamp(1705314645), + Assert.IsType(headers["unspecified"])); + Assert.Equal(headers["utc"], headers["unspecified"]); + } + + [Fact] + public void Format_Should_ProduceTheEpochFloor_When_HeaderIsTheDefaultDateTime() + { + // arrange + var envelope = new MessageEnvelope + { + Headers = new Headers([new HeaderValue { Key = "at", Value = default(DateTime) }]) + }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal( + DateTimeOffset.MinValue.ToUnixTimeSeconds(), + Assert.IsType(headers["at"]).UnixTime); + } + + [Fact] + public void Format_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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal( + new Dictionary + { + ["fits"] = 9.95m, + ["exceeds"] = "21474836.48", + ["negative"] = "-21474836.48" + }, + headers); + } + + [Fact] + public void Format_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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal( + new Dictionary + { + ["object"] = new Dictionary { ["k"] = 1 }, + ["array"] = new List { 1, 2 }, + ["text"] = "abc", + ["number"] = 5 + }, + headers); + } + + [Fact] + 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 + var envelope = new MessageEnvelope + { + Headers = new Headers( + [ + new HeaderValue { Key = "callback", Value = new Uri("https://example.com/a%20b") } + ]) + }; + + // act + var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal("https://example.com/a%20b", headers["callback"]); + } + + [Fact] + public void Format_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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // assert + Assert.Equal( + new Dictionary { ["1"] = "a", ["time"] = new AmqpTimestamp(1700000000) }, + Assert.IsType>(headers["meta"])); + } + + [Fact] + public void Format_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 = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; + + // 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 Format_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(() => RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!); + + // assert + Assert.Equal( + "The header 'x-cycle' exceeds the maximum AMQP field-table nesting depth of 64.", + Assert.IsType(exception).Message); + } + + 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 f6e4f0f618b..cb332824ed0 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; @@ -14,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); @@ -27,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); @@ -40,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); @@ -53,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); @@ -66,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); @@ -79,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); @@ -92,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); @@ -106,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); @@ -120,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()); @@ -133,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)); @@ -153,12 +154,96 @@ 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); } + [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, TimeProvider.System); + + // 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, TimeProvider.System); + + // 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 = RabbitMQMessageEnvelopeFormatter.Format(published, TimeProvider.System).Headers); + + // act + var envelope = _parser.Parse(args, TimeProvider.System); + + // assert + envelope.Headers!.TryGetValue("x-signature", out var signature); + 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, TimeProvider.System)); + + // assert + Assert.Equal( + "The header 'x-deep' exceeds the maximum AMQP field-table nesting depth of 64.", + Assert.IsType(exception).Message); + } + [Fact] public void Parse_Should_ConvertAmqpTimestamp_When_HeaderIsTimestamp() { @@ -166,13 +251,87 @@ 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)); 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, TimeProvider.System); + + // 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, TimeProvider.System); + + // 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, TimeProvider.System); + + // assert + Assert.True(envelope.Headers!.TryGetValue("x-binary", out var parsed)); + Assert.Equal(value, parsed); + } + [Fact] public void Parse_Should_SetDeliveryCountOne_When_Redelivered() { @@ -180,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); @@ -193,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); @@ -211,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); @@ -229,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); @@ -247,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); @@ -266,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); @@ -288,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); @@ -305,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); @@ -323,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); @@ -336,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); @@ -349,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); @@ -362,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 eb1ef004402..b74bce5c01c 100644 --- a/website/content/docs/mocha/messages.md +++ b/website/content/docs/mocha/messages.md @@ -73,6 +73,33 @@ await bus.SendAsync( The bus merges your headers into the envelope's header collection before dispatching. +## 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 `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: