Skip to content

Closes#1849: include a message id tag in rabbitmqactivitysource for published messages #1852

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Aug 19, 2025
Merged

Closes#1849: include a message id tag in rabbitmqactivitysource for published messages #1852

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
4 changes: 2 additions & 2 deletions projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand All @@ -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;
Expand Down
146 changes: 22 additions & 124 deletions projects/Test/SequentialIntegration/TestActivitySource.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)]
Expand Down Expand Up @@ -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;
Expand All @@ -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
{
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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);
}
}
}
}
6 changes: 3 additions & 3 deletions projects/Test/SequentialIntegration/TestHeartbeats.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");

Expand All @@ -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");

Expand All @@ -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");

Expand Down
Loading
Loading