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

updated tests for messageid tag

corrected broken tests after bad conflict resolution from merging main into feature branch

Author: YayBurritos

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;

projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs

@@ -66,7 +66,7 @@ public static bool UseRoutingKeyAsOperationName
             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())
@@ -83,7 +83,7 @@ public static bool UseRoutingKeyAsOperationName
                     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;

projects/Test/SequentialIntegration/TestActivitySource.cs

@@ -75,120 +75,6 @@ void AssertIntTagGreaterThanZero(Activity activity, string name)
             Assert.True(activity.GetTagItem(name) is int result && result > 0);
         }
 
-        [Theory]
-        [InlineData(true, true)]
-        [InlineData(true, false)]
-        [InlineData(false, true)]
-        [InlineData(false, false)]
-        public async Task TestPublisherAndConsumerActivityTags(bool useRoutingKeyAsOperationName, bool usePublisherAsParent)
-        {
-            RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
-            RabbitMQActivitySource.TracingOptions.UsePublisherAsParent = usePublisherAsParent;
-            var _activities = new List<Activity>();
-            using ActivityListener activityListener = StartActivityListener(_activities);
-            await Task.Delay(500);
-            string queueName = $"{Guid.NewGuid()}";
-            QueueDeclareOk q = await _channel.QueueDeclareAsync(queueName);
-            byte[] sendBody = Encoding.UTF8.GetBytes("hi");
-            byte[] consumeBody = null;
-            var consumer = new AsyncEventingBasicConsumer(_channel);
-            var consumerReceivedTcs =
-                new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-            consumer.ReceivedAsync += (o, a) =>
-            {
-                consumeBody = a.Body.ToArray();
-                consumerReceivedTcs.SetResult(true);
-                return Task.CompletedTask;
-            };
-
-            string consumerTag = await _channel.BasicConsumeAsync(queueName, autoAck: true, consumer: consumer);
-            await _channel.BasicPublishAsync("", q.QueueName, true, sendBody);
-
-            await consumerReceivedTcs.Task.WaitAsync(TimeSpan.FromSeconds(5));
-            Assert.True(await consumerReceivedTcs.Task);
-
-            await _channel.BasicCancelAsync(consumerTag);
-            await Task.Delay(500);
-            AssertActivityData(useRoutingKeyAsOperationName, usePublisherAsParent, queueName, _activities, true);
-        }
-
-        [Theory]
-        [InlineData(true, true)]
-        [InlineData(true, false)]
-        [InlineData(false, true)]
-        [InlineData(false, false)]
-        public async Task TestPublisherWithCachedStringsAndConsumerActivityTags(bool useRoutingKeyAsOperationName, bool usePublisherAsParent)
-        {
-            RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
-            RabbitMQActivitySource.TracingOptions.UsePublisherAsParent = usePublisherAsParent;
-            var _activities = new List<Activity>();
-            using ActivityListener activityListener = StartActivityListener(_activities);
-            await Task.Delay(500);
-            string queueName = $"{Guid.NewGuid()}";
-            QueueDeclareOk q = await _channel.QueueDeclareAsync(queueName);
-            byte[] sendBody = Encoding.UTF8.GetBytes("hi");
-            byte[] consumeBody = null;
-            var consumer = new AsyncEventingBasicConsumer(_channel);
-            var consumerReceivedTcs =
-                new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-            consumer.ReceivedAsync += (o, a) =>
-            {
-                consumeBody = a.Body.ToArray();
-                consumerReceivedTcs.SetResult(true);
-                return Task.CompletedTask;
-            };
-
-            string consumerTag = await _channel.BasicConsumeAsync(queueName, autoAck: true, consumer: consumer);
-            CachedString exchange = new CachedString("");
-            CachedString routingKey = new CachedString(q.QueueName);
-            await _channel.BasicPublishAsync(exchange, routingKey, true, sendBody);
-
-            await consumerReceivedTcs.Task.WaitAsync(TimeSpan.FromSeconds(5));
-            Assert.True(await consumerReceivedTcs.Task);
-
-            await _channel.BasicCancelAsync(consumerTag);
-            await Task.Delay(500);
-            AssertActivityData(useRoutingKeyAsOperationName, usePublisherAsParent, queueName, _activities, true);
-        }
-
-        [Theory]
-        [InlineData(true, true)]
-        [InlineData(true, false)]
-        [InlineData(false, true)]
-        [InlineData(false, false)]
-        public async Task TestPublisherWithPublicationAddressAndConsumerActivityTags(bool useRoutingKeyAsOperationName, bool usePublisherAsParent)
-        {
-            RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
-            RabbitMQActivitySource.TracingOptions.UsePublisherAsParent = usePublisherAsParent;
-            var _activities = new List<Activity>();
-            using ActivityListener activityListener = StartActivityListener(_activities);
-            await Task.Delay(500);
-            string queueName = $"{Guid.NewGuid()}";
-            QueueDeclareOk q = await _channel.QueueDeclareAsync(queueName);
-            byte[] sendBody = Encoding.UTF8.GetBytes("hi");
-            byte[] consumeBody = null;
-            var consumer = new AsyncEventingBasicConsumer(_channel);
-            var consumerReceivedTcs =
-                new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-            consumer.ReceivedAsync += (o, a) =>
-            {
-                consumeBody = a.Body.ToArray();
-                consumerReceivedTcs.SetResult(true);
-                return Task.CompletedTask;
-            };
-
-            string consumerTag = await _channel.BasicConsumeAsync(queueName, autoAck: true, consumer: consumer);
-            PublicationAddress publicationAddress = new PublicationAddress(ExchangeType.Direct, "", q.QueueName);
-            await _channel.BasicPublishAsync(publicationAddress, new BasicProperties(), sendBody);
-
-            await consumerReceivedTcs.Task.WaitAsync(TimeSpan.FromSeconds(5));
-            Assert.True(await consumerReceivedTcs.Task);
-
-            await _channel.BasicCancelAsync(consumerTag);
-            await Task.Delay(500);
-            AssertActivityData(useRoutingKeyAsOperationName, usePublisherAsParent, queueName, _activities, true);
-        }
-
         [Theory]
         [InlineData(true, true)]
         [InlineData(true, false)]
@@ -307,11 +193,15 @@ public async Task TestPublisherWithPublicationAddressAndConsumerActivityTagsAsyn
         }
 
         [Theory]
-        [InlineData(true, true)]
-        [InlineData(true, false)]
-        [InlineData(false, true)]
-        [InlineData(false, false)]
-        public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName, bool usePublisherAsParent)
+        [InlineData(true, true, true)]
+        [InlineData(true, true, false)]
+        [InlineData(true, false, true)]
+        [InlineData(true, false, false)]
+        [InlineData(false, true, true)]
+        [InlineData(false, true, false)]
+        [InlineData(false, false, true)]
+        [InlineData(false, false, false)]
+        public async Task TestPublisherAndBasicGetActivityTagsAsync(bool useRoutingKeyAsOperationName, bool usePublisherAsParent, bool useMessageId)
         {
             RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
             RabbitMQActivitySource.TracingOptions.UsePublisherAsParent = usePublisherAsParent;
@@ -321,18 +211,20 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera
             string queue = $"queue-{Guid.NewGuid()}";
             const string msg = "for basic.get";
 
+            var basicProps = useMessageId ? new BasicProperties() { MessageId = Guid.NewGuid().ToString() } : new BasicProperties();
+
             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, usePublisherAsParent, queue, activities, false);
+                AssertActivityData(useRoutingKeyAsOperationName, usePublisherAsParent, queue, activities, false, basicProps.MessageId);
             }
             finally
             {
@@ -345,7 +237,7 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera
         [InlineData(true, false)]
         [InlineData(false, true)]
         [InlineData(false, false)]
-        public async Task TestPublisherWithCachedStringsAndBasicGetActivityTags(bool useRoutingKeyAsOperationName, bool usePublisherAsParent)
+        public async Task TestPublisherWithCachedStringsAndBasicGetActivityTagsAsync(bool useRoutingKeyAsOperationName, bool usePublisherAsParent)
         {
             RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
             RabbitMQActivitySource.TracingOptions.UsePublisherAsParent = usePublisherAsParent;
@@ -381,7 +273,7 @@ public async Task TestPublisherWithCachedStringsAndBasicGetActivityTags(bool use
         [InlineData(true, false)]
         [InlineData(false, true)]
         [InlineData(false, false)]
-        public async Task TestPublisherWithPublicationAddressAndBasicGetActivityTags(bool useRoutingKeyAsOperationName, bool usePublisherAsParent)
+        public async Task TestPublisherWithPublicationAddressAndBasicGetActivityTagsAsync(bool useRoutingKeyAsOperationName, bool usePublisherAsParent)
         {
             RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
             RabbitMQActivitySource.TracingOptions.UsePublisherAsParent = usePublisherAsParent;
@@ -427,7 +319,7 @@ private static ActivityListener StartActivityListener(List<Activity> activities)
         }
 
         private void AssertActivityData(bool useRoutingKeyAsOperationName, bool usePublisherAsParent, string queueName,
-            List<Activity> activityList, bool isDeliver = false)
+            List<Activity> activityList, bool isDeliver = false, string messageId = null)
         {
             string childName = isDeliver ? "deliver" : "fetch";
             Activity[] activities = activityList.ToArray();
@@ -480,6 +372,12 @@ private void AssertActivityData(bool useRoutingKeyAsOperationName, bool usePubli
             AssertIntTagGreaterThanZero(sendActivity, RabbitMQActivitySource.MessagingEnvelopeSize);
             AssertIntTagGreaterThanZero(sendActivity, RabbitMQActivitySource.MessagingBodySize);
             AssertIntTagGreaterThanZero(receiveActivity, RabbitMQActivitySource.MessagingBodySize);
+
+            if (messageId is not null)
+            {
+                AssertStringTagEquals(sendActivity, RabbitMQActivitySource.MessageId, messageId);
+                AssertStringTagEquals(receiveActivity, RabbitMQActivitySource.MessageId, messageId);
+            }
         }
     }
 }

projects/Test/SequentialIntegration/TestHeartbeats.cs

@@ -59,7 +59,7 @@ public override Task InitializeAsync()
 
         [SkippableFact(Timeout = 35000)]
         [Trait("Category", "LongRunning")]
-        public async Task TestThatHeartbeatWriterUsesConfigurableInterval()
+        public async Task TestThatHeartbeatWriterUsesConfigurableIntervalAsync()
         {
             Skip.IfNot(LongRunningTestsEnabled(), "RABBITMQ_LONG_RUNNING_TESTS is not set, skipping test");
 
@@ -72,7 +72,7 @@ public async Task TestThatHeartbeatWriterUsesConfigurableInterval()
 
         [SkippableFact]
         [Trait("Category", "LongRunning")]
-        public async Task TestThatHeartbeatWriterWithTLSEnabled()
+        public async Task TestThatHeartbeatWriterWithTLSEnabledAsync()
         {
             Skip.IfNot(LongRunningTestsEnabled(), "RABBITMQ_LONG_RUNNING_TESTS is not set, skipping test");
 
@@ -94,7 +94,7 @@ public async Task TestThatHeartbeatWriterWithTLSEnabled()
 
         [SkippableFact(Timeout = 90000)]
         [Trait("Category", "LongRunning")]
-        public async Task TestHundredsOfConnectionsWithRandomHeartbeatInterval()
+        public async Task TestHundredsOfConnectionsWithRandomHeartbeatIntervalAsync()
         {
             Skip.IfNot(LongRunningTestsEnabled(), "RABBITMQ_LONG_RUNNING_TESTS is not set, skipping test");