Fixes Mocha Header Formatting

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());
+        }
+    }
+}

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 <see cref="MessageEnvelope"/>
 /// and sets it on the receive context for downstream processing.
 /// </summary>
-internal sealed class RabbitMQParsingMiddleware
+internal sealed class RabbitMQParsingMiddleware(TimeProvider timeProvider)
 {
     /// <summary>
     /// 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<RabbitMQReceiveFeature>();
         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();
-
     /// <summary>
     /// Creates a <see cref="ReceiveMiddlewareConfiguration"/> that wraps the parsing middleware singleton.
     /// </summary>
     /// <returns>A middleware configuration keyed as "RabbitMQParsing".</returns>
     public static ReceiveMiddlewareConfiguration Create()
-        => new(static (_, next) => ctx => s_instance.InvokeAsync(ctx, next), "RabbitMQParsing");
+        => new(
+            static (context, next) =>
+            {
+                var timeProvider = context.Services.GetRequiredService<TimeProvider>();
+                var middleware = new RabbitMQParsingMiddleware(timeProvider);
+                return ctx => middleware.InvokeAsync(ctx, next);
+            },
+            "RabbitMQParsing");
 }

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");
     }
 }
-
-/// <summary>
-/// Extension methods for building RabbitMQ-specific message headers from a <see cref="MessageEnvelope"/>.
-/// </summary>
-public static class RabbitMQDispatchContextExtensions
-{
-    internal static IDictionary<string, object?> 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<string, object?>(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;
-    }
-}