Pass basic props to RabbitMQActivitySource.BasicPublish to allow messageid tag to be included in trace span

YayBurritos · YayBurritos · commit 20a71f46dc7c · 2025-06-25T13:27:01.000-04:00
diff --git a/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs b/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs
@@ -61,7 +61,7 @@ await MaybeEnforceFlowControlAsync(cancellationToken)
                 var cmd = new BasicPublish(exchange, routingKey, mandatory, default);
 
                 using Activity? sendActivity = RabbitMQActivitySource.PublisherHasListeners
-                    ? RabbitMQActivitySource.BasicPublish(routingKey, exchange, body.Length)
+                    ? RabbitMQActivitySource.BasicPublish(routingKey, exchange, body.Length, basicProperties)
                     : default;
 
                 ulong publishSequenceNumber = 0;
@@ -116,7 +116,7 @@ await MaybeEnforceFlowControlAsync(cancellationToken)
                 var cmd = new BasicPublishMemory(exchange.Bytes, routingKey.Bytes, mandatory, default);
 
                 using Activity? sendActivity = RabbitMQActivitySource.PublisherHasListeners
-                    ? RabbitMQActivitySource.BasicPublish(routingKey.Value, exchange.Value, body.Length)
+                    ? RabbitMQActivitySource.BasicPublish(routingKey.Value, exchange.Value, body.Length, basicProperties)
                     : default;
 
                 ulong publishSequenceNumber = 0;
diff --git a/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs b/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs
@@ -61,7 +61,7 @@ public static class RabbitMQActivitySource
             new KeyValuePair<string, object?>(ProtocolVersion, "0.9.1")
         };
 
-        internal static Activity? BasicPublish(string routingKey, string exchange, int bodySize,
+        internal static Activity? BasicPublish(string routingKey, string exchange, int bodySize, IReadOnlyBasicProperties basicProperties,
             ActivityContext linkedContext = default)
         {
             if (!s_publisherSource.HasListeners())
@@ -78,7 +78,7 @@ public static class RabbitMQActivitySource
                     ActivityKind.Producer, linkedContext);
             if (activity != null && activity.IsAllDataRequested)
             {
-                PopulateMessagingTags(MessagingOperationTypeSend, MessagingOperationNameBasicPublish, routingKey, exchange, 0, bodySize, activity);
+                PopulateMessagingTags(MessagingOperationTypeSend, MessagingOperationNameBasicPublish, routingKey, exchange, 0, basicProperties, bodySize, activity);
             }
 
             return activity;
diff --git a/projects/Test/SequentialIntegration/TestActivitySource.cs b/projects/Test/SequentialIntegration/TestActivitySource.cs
@@ -289,9 +289,11 @@ public async Task TestPublisherWithPublicationAddressAndConsumerActivityTagsAsyn
         }
 
         [Theory]
-        [InlineData(true)]
-        [InlineData(false)]
-        public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName)
+        [InlineData(true, true)]
+        [InlineData(false, true)]
+        [InlineData(true, false)]
+        [InlineData(false, false)]
+        public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName, bool verifyMessageIdTag)
         {
             RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
             var activities = new List<Activity>();
@@ -300,18 +302,20 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera
             string queue = $"queue-{Guid.NewGuid()}";
             const string msg = "for basic.get";
 
+            var basicProps = new BasicProperties() { MessageId = Guid.NewGuid().ToString() };
+
             try
             {
                 await _channel.QueueDeclareAsync(queue, false, false, false, null);
-                await _channel.BasicPublishAsync("", queue, true, Encoding.UTF8.GetBytes(msg));
+                await _channel.BasicPublishAsync("", queue, true, basicProps, Encoding.UTF8.GetBytes(msg));
                 QueueDeclareOk ok = await _channel.QueueDeclarePassiveAsync(queue);
                 Assert.Equal(1u, ok.MessageCount);
                 BasicGetResult res = await _channel.BasicGetAsync(queue, true);
                 Assert.Equal(msg, Encoding.UTF8.GetString(res.Body.ToArray()));
                 ok = await _channel.QueueDeclarePassiveAsync(queue);
                 Assert.Equal(0u, ok.MessageCount);
                 await Task.Delay(500);
-                AssertActivityData(useRoutingKeyAsOperationName, queue, activities, false);
+                AssertActivityData(useRoutingKeyAsOperationName, queue, activities, false, verifyMessageIdTag);
             }
             finally
             {
@@ -400,7 +404,7 @@ private static ActivityListener StartActivityListener(List<Activity> activities)
         }
 
         private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueName,
-            List<Activity> activityList, bool isDeliver = false)
+            List<Activity> activityList, bool isDeliver = false, bool verifyMessageIdTag = false)
         {
             string childName = isDeliver ? "deliver" : "fetch";
             Activity[] activities = activityList.ToArray();
@@ -444,6 +448,12 @@ private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueN
             AssertIntTagGreaterThanZero(sendActivity, RabbitMQActivitySource.MessagingEnvelopeSize);
             AssertIntTagGreaterThanZero(sendActivity, RabbitMQActivitySource.MessagingBodySize);
             AssertIntTagGreaterThanZero(receiveActivity, RabbitMQActivitySource.MessagingBodySize);
+
+            if (verifyMessageIdTag)
+            {
+                AssertStringTagNotNullOrEmpty(sendActivity, RabbitMQActivitySource.MessageId);
+                AssertStringTagNotNullOrEmpty(receiveActivity, RabbitMQActivitySource.MessageId);
+            }
         }
     }
 }
diff --git a/projects/Test/SequentialIntegration/TestOpenTelemetry.cs b/projects/Test/SequentialIntegration/TestOpenTelemetry.cs
@@ -301,9 +301,11 @@ public async Task TestPublisherWithCachedStringsAndConsumerActivityTagsAsync(boo
         }
 
         [Theory]
-        [InlineData(true)]
-        [InlineData(false)]
-        public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName)
+        [InlineData(true, true)]
+        [InlineData(false, true)]
+        [InlineData(true, false)]
+        [InlineData(false, false)]
+        public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName, bool verifyMessageIdTag)
         {
             var exportedItems = new List<Activity>();
             using var tracer = Sdk.CreateTracerProviderBuilder()
@@ -318,10 +320,12 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera
             string queue = $"queue-{Guid.NewGuid()}";
             const string msg = "for basic.get";
 
+            var basicProps = new BasicProperties() { MessageId = Guid.NewGuid().ToString() };
+
             try
             {
                 await _channel.QueueDeclareAsync(queue, false, false, false, null);
-                await _channel.BasicPublishAsync("", queue, true, Encoding.UTF8.GetBytes(msg));
+                await _channel.BasicPublishAsync("", queue, true, basicProps, Encoding.UTF8.GetBytes(msg));
                 Baggage.ClearBaggage();
                 Assert.Null(Baggage.GetBaggage("TestItem"));
                 QueueDeclareOk ok = await _channel.QueueDeclarePassiveAsync(queue);
@@ -331,7 +335,7 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera
                 ok = await _channel.QueueDeclarePassiveAsync(queue);
                 Assert.Equal(0u, ok.MessageCount);
                 await Task.Delay(500);
-                AssertActivityData(useRoutingKeyAsOperationName, queue, exportedItems, false);
+                AssertActivityData(useRoutingKeyAsOperationName, queue, exportedItems, false, verifyMessageIdTag);
             }
             finally
             {
@@ -340,7 +344,7 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera
         }
 
         private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueName,
-            List<Activity> activityList, bool isDeliver = false, string baggageGuid = null)
+            List<Activity> activityList, bool isDeliver = false, bool verifyMessageIdTag = false)
         {
             string childName = isDeliver ? "deliver" : "fetch";
             string childType = isDeliver ? "process" : "receive";
@@ -385,6 +389,12 @@ private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueN
             AssertStringTagEquals(receiveActivity, RabbitMQActivitySource.MessagingOperationName, childName);
             AssertStringTagEquals(sendActivity, RabbitMQActivitySource.MessagingOperationType, "send");
             AssertStringTagEquals(sendActivity, RabbitMQActivitySource.MessagingOperationName, "publish");
+
+            if (verifyMessageIdTag)
+            {
+                AssertStringTagNotNullOrEmpty(sendActivity, RabbitMQActivitySource.MessageId);
+                AssertStringTagNotNullOrEmpty(receiveActivity, RabbitMQActivitySource.MessageId);
+            }
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -61,7 +61,7 @@ public static class RabbitMQActivitySource`
`61`	`61`	`new KeyValuePair<string, object?>(ProtocolVersion, "0.9.1")`
`62`	`62`	`};`
`63`	`63`
`64`		`- internal static Activity? BasicPublish(string routingKey, string exchange, int bodySize,`
	`64`	`+ internal static Activity? BasicPublish(string routingKey, string exchange, int bodySize, IReadOnlyBasicProperties basicProperties,`
`65`	`65`	`ActivityContext linkedContext = default)`
`66`	`66`	`{`
`67`	`67`	`if (!s_publisherSource.HasListeners())`
`@@ -78,7 +78,7 @@ public static class RabbitMQActivitySource`
`78`	`78`	`ActivityKind.Producer, linkedContext);`
`79`	`79`	`if (activity != null && activity.IsAllDataRequested)`
`80`	`80`	`{`
`81`		`- PopulateMessagingTags(MessagingOperationTypeSend, MessagingOperationNameBasicPublish, routingKey, exchange, 0, bodySize, activity);`
	`81`	`+ PopulateMessagingTags(MessagingOperationTypeSend, MessagingOperationNameBasicPublish, routingKey, exchange, 0, basicProperties, bodySize, activity);`
`82`	`82`	`}`
`83`	`83`
`84`	`84`	`return activity;`
Original file line number	Diff line number	Diff line change
`@@ -289,9 +289,11 @@ public async Task TestPublisherWithPublicationAddressAndConsumerActivityTagsAsyn`
`289`	`289`	`}`
`290`	`290`
`291`	`291`	`[Theory]`
`292`		`- [InlineData(true)]`
`293`		`- [InlineData(false)]`
`294`		`- public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName)`
	`292`	`+ [InlineData(true, true)]`
	`293`	`+ [InlineData(false, true)]`
	`294`	`+ [InlineData(true, false)]`
	`295`	`+ [InlineData(false, false)]`
	`296`	`+ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName, bool verifyMessageIdTag)`
`295`	`297`	`{`
`296`	`298`	`RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;`
`297`	`299`	`var activities = new List<Activity>();`
`@@ -300,18 +302,20 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera`
`300`	`302`	`string queue = $"queue-{Guid.NewGuid()}";`
`301`	`303`	`const string msg = "for basic.get";`
`302`	`304`
	`305`	`+ var basicProps = new BasicProperties() { MessageId = Guid.NewGuid().ToString() };`
	`306`	`+`
`303`	`307`	`try`
`304`	`308`	`{`
`305`	`309`	`await _channel.QueueDeclareAsync(queue, false, false, false, null);`
`306`		`- await _channel.BasicPublishAsync("", queue, true, Encoding.UTF8.GetBytes(msg));`
	`310`	`+ await _channel.BasicPublishAsync("", queue, true, basicProps, Encoding.UTF8.GetBytes(msg));`
`307`	`311`	`QueueDeclareOk ok = await _channel.QueueDeclarePassiveAsync(queue);`
`308`	`312`	`Assert.Equal(1u, ok.MessageCount);`
`309`	`313`	`BasicGetResult res = await _channel.BasicGetAsync(queue, true);`
`310`	`314`	`Assert.Equal(msg, Encoding.UTF8.GetString(res.Body.ToArray()));`
`311`	`315`	`ok = await _channel.QueueDeclarePassiveAsync(queue);`
`312`	`316`	`Assert.Equal(0u, ok.MessageCount);`
`313`	`317`	`await Task.Delay(500);`
`314`		`- AssertActivityData(useRoutingKeyAsOperationName, queue, activities, false);`
	`318`	`+ AssertActivityData(useRoutingKeyAsOperationName, queue, activities, false, verifyMessageIdTag);`
`315`	`319`	`}`
`316`	`320`	`finally`
`317`	`321`	`{`
`@@ -400,7 +404,7 @@ private static ActivityListener StartActivityListener(List<Activity> activities)`
`400`	`404`	`}`
`401`	`405`
`402`	`406`	`private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueName,`
`403`		`- List<Activity> activityList, bool isDeliver = false)`
	`407`	`+ List<Activity> activityList, bool isDeliver = false, bool verifyMessageIdTag = false)`
`404`	`408`	`{`
`405`	`409`	`string childName = isDeliver ? "deliver" : "fetch";`
`406`	`410`	`Activity[] activities = activityList.ToArray();`
`@@ -444,6 +448,12 @@ private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueN`
`444`	`448`	`AssertIntTagGreaterThanZero(sendActivity, RabbitMQActivitySource.MessagingEnvelopeSize);`
`445`	`449`	`AssertIntTagGreaterThanZero(sendActivity, RabbitMQActivitySource.MessagingBodySize);`
`446`	`450`	`AssertIntTagGreaterThanZero(receiveActivity, RabbitMQActivitySource.MessagingBodySize);`
	`451`	`+`
	`452`	`+ if (verifyMessageIdTag)`
	`453`	`+ {`
	`454`	`+ AssertStringTagNotNullOrEmpty(sendActivity, RabbitMQActivitySource.MessageId);`
	`455`	`+ AssertStringTagNotNullOrEmpty(receiveActivity, RabbitMQActivitySource.MessageId);`
	`456`	`+ }`
`447`	`457`	`}`
`448`	`458`	`}`
`449`	`459`	`}`
Original file line number	Diff line number	Diff line change
`@@ -301,9 +301,11 @@ public async Task TestPublisherWithCachedStringsAndConsumerActivityTagsAsync(boo`
`301`	`301`	`}`
`302`	`302`
`303`	`303`	`[Theory]`
`304`		`- [InlineData(true)]`
`305`		`- [InlineData(false)]`
`306`		`- public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName)`
	`304`	`+ [InlineData(true, true)]`
	`305`	`+ [InlineData(false, true)]`
	`306`	`+ [InlineData(true, false)]`
	`307`	`+ [InlineData(false, false)]`
	`308`	`+ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName, bool verifyMessageIdTag)`
`307`	`309`	`{`
`308`	`310`	`var exportedItems = new List<Activity>();`
`309`	`311`	`using var tracer = Sdk.CreateTracerProviderBuilder()`
`@@ -318,10 +320,12 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera`
`318`	`320`	`string queue = $"queue-{Guid.NewGuid()}";`
`319`	`321`	`const string msg = "for basic.get";`
`320`	`322`
	`323`	`+ var basicProps = new BasicProperties() { MessageId = Guid.NewGuid().ToString() };`
	`324`	`+`
`321`	`325`	`try`
`322`	`326`	`{`
`323`	`327`	`await _channel.QueueDeclareAsync(queue, false, false, false, null);`
`324`		`- await _channel.BasicPublishAsync("", queue, true, Encoding.UTF8.GetBytes(msg));`
	`328`	`+ await _channel.BasicPublishAsync("", queue, true, basicProps, Encoding.UTF8.GetBytes(msg));`
`325`	`329`	`Baggage.ClearBaggage();`
`326`	`330`	`Assert.Null(Baggage.GetBaggage("TestItem"));`
`327`	`331`	`QueueDeclareOk ok = await _channel.QueueDeclarePassiveAsync(queue);`
`@@ -331,7 +335,7 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera`
`331`	`335`	`ok = await _channel.QueueDeclarePassiveAsync(queue);`
`332`	`336`	`Assert.Equal(0u, ok.MessageCount);`
`333`	`337`	`await Task.Delay(500);`
`334`		`- AssertActivityData(useRoutingKeyAsOperationName, queue, exportedItems, false);`
	`338`	`+ AssertActivityData(useRoutingKeyAsOperationName, queue, exportedItems, false, verifyMessageIdTag);`
`335`	`339`	`}`
`336`	`340`	`finally`
`337`	`341`	`{`
`@@ -340,7 +344,7 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera`
`340`	`344`	`}`
`341`	`345`
`342`	`346`	`private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueName,`
`343`		`- List<Activity> activityList, bool isDeliver = false, string baggageGuid = null)`
	`347`	`+ List<Activity> activityList, bool isDeliver = false, bool verifyMessageIdTag = false)`
`344`	`348`	`{`
`345`	`349`	`string childName = isDeliver ? "deliver" : "fetch";`
`346`	`350`	`string childType = isDeliver ? "process" : "receive";`
`@@ -385,6 +389,12 @@ private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueN`
`385`	`389`	`AssertStringTagEquals(receiveActivity, RabbitMQActivitySource.MessagingOperationName, childName);`
`386`	`390`	`AssertStringTagEquals(sendActivity, RabbitMQActivitySource.MessagingOperationType, "send");`
`387`	`391`	`AssertStringTagEquals(sendActivity, RabbitMQActivitySource.MessagingOperationName, "publish");`
	`392`	`+`
	`393`	`+ if (verifyMessageIdTag)`
	`394`	`+ {`
	`395`	`+ AssertStringTagNotNullOrEmpty(sendActivity, RabbitMQActivitySource.MessageId);`
	`396`	`+ AssertStringTagNotNullOrEmpty(receiveActivity, RabbitMQActivitySource.MessageId);`
	`397`	`+ }`
`388`	`398`	`}`
`389`	`399`	`}`
`390`	`400`	`}`