Map ParentId to MT-Activity-Id in MassTransit interop mode

src/Transports/AWS/Wolverine.AmazonSns/Internal/MassTransitMapper.cs

@@ -25,14 +25,24 @@ public string BuildMessageBody(Envelope envelope)
 
     public IEnumerable<KeyValuePair<string, MessageAttributeValue>> ToAttributes(Envelope envelope)
     {
-        yield break;
+        if (!string.IsNullOrEmpty(envelope.ParentId))
+        {
+            yield return new KeyValuePair<string, MessageAttributeValue>(
+                MassTransitHeaders.ActivityId,
+                new MessageAttributeValue { DataType = "String", StringValue = envelope.ParentId });
+        }
     }
 
     public void ReadEnvelopeData(Envelope envelope, string messageBody, IDictionary<string, MessageAttributeValue> attributes)
     {
         // TODO -- this could be more efficient of course
         envelope.Data = Encoding.UTF8.GetBytes(messageBody);
         envelope.Serializer = _serializer;
+
+        if (attributes.TryGetValue(MassTransitHeaders.ActivityId, out var activityId))
+        {
+            envelope.ParentId = activityId.StringValue;
+        }
     }
 
 }

src/Transports/AWS/Wolverine.AmazonSqs/Internal/MassTransitMapper.cs

@@ -28,17 +28,27 @@ public string BuildMessageBody(Envelope envelope)
 
     public IEnumerable<KeyValuePair<string, MessageAttributeValue>> ToAttributes(Envelope envelope)
     {
-        yield break;
+        if (!string.IsNullOrEmpty(envelope.ParentId))
+        {
+            yield return new KeyValuePair<string, MessageAttributeValue>(
+                MassTransitHeaders.ActivityId,
+                new MessageAttributeValue { DataType = "String", StringValue = envelope.ParentId });
+        }
     }
 
     public void ReadEnvelopeData(Envelope envelope, string messageBody, IDictionary<string, MessageAttributeValue> attributes)
     {
         // TODO -- this could be more efficient of course
         envelope.Data = Encoding.UTF8.GetBytes(messageBody);
-        
+
         // This is the really important part
         // of the mapping
         envelope.Serializer = _serializer;
+
+        if (attributes.TryGetValue(MassTransitHeaders.ActivityId, out var activityId))
+        {
+            envelope.ParentId = activityId.StringValue;
+        }
     }
 }
 

src/Wolverine/AssemblyAttributes.cs

@@ -19,6 +19,7 @@
 [assembly: InternalsVisibleTo("Wolverine.RabbitMq.Tests")]
 [assembly: InternalsVisibleTo("Wolverine.AzureServiceBus")]
 [assembly: InternalsVisibleTo("Wolverine.AmazonSqs")]
+[assembly: InternalsVisibleTo("Wolverine.AmazonSns")]
 [assembly: InternalsVisibleTo("Wolverine.ConfluentKafka")]
 [assembly: InternalsVisibleTo("Wolverine.AzureServiceBus.Tests")]
 [assembly: InternalsVisibleTo("PersistenceTests")]

src/Wolverine/Runtime/Interop/MassTransit/MassTransitHeaders.cs

@@ -83,6 +83,11 @@ internal static class MassTransitHeaders
     /// </summary>
     public const string InitiatingConversationId = "MT-InitiatingConversationId";
 
+    /// <summary>
+    ///     The W3C Activity Id used for distributed trace propagation
+    /// </summary>
+    public const string ActivityId = "MT-Activity-Id";
+
     /// <summary>
     ///     MessageId - <see cref="MessageEnvelope" />
     /// </summary>

src/Wolverine/Transports/EnvelopeMapper.cs

@@ -119,9 +119,10 @@ public void InteropWithMassTransit(Action<IMassTransitInterop>? configure = null
         {
             var serializer = new MassTransitJsonSerializer(e);
             configure?.Invoke(serializer);
-            
+
             MapPropertyToHeader(x => x.MessageType!, MassTransitHeaders.MessageType);
-
+            MapPropertyToHeader(x => x.ParentId!, MassTransitHeaders.ActivityId);
+
             _endpoint.DefaultSerializer = serializer;
 
             var replyUri = new Lazy<string>(() => e.MassTransitReplyUri()?.ToString() ?? string.Empty);