Update for NServiceBus 8

src/NServiceBus.Extensions.IntegrationTesting/AttachIncomingLogicalMessageContextToActivity.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Diagnostics;
+using System.Threading.Tasks;
+using NServiceBus.Pipeline;
+
+namespace NServiceBus.Extensions.IntegrationTesting
+{
+    internal class AttachIncomingLogicalMessageContextToActivity : Behavior<IIncomingLogicalMessageContext>
+    {
+        public override Task Invoke(
+            IIncomingLogicalMessageContext context,
+            Func<Task> next
+        )
+        {
+            Activity.Current?.AddTag("testing.incoming.message.context", context);
+            return next();
+        }
+    }
+}

src/NServiceBus.Extensions.IntegrationTesting/AttachInvokeHandlerContextToActivity.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Diagnostics;
+using System.Threading.Tasks;
+using NServiceBus.Pipeline;
+
+namespace NServiceBus.Extensions.IntegrationTesting
+{
+    internal class AttachInvokeHandlerContextToActivity : Behavior<IInvokeHandlerContext>
+    {
+        public override Task Invoke(
+            IInvokeHandlerContext context,
+            Func<Task> next
+        )
+        {
+            Activity.Current?.AddTag("testing.invoke.handler.context", context);
+            return next();
+        }
+    }
+}

src/NServiceBus.Extensions.IntegrationTesting/AttachOutgoingLogicalMessageContextToActivity.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Diagnostics;
+using System.Threading.Tasks;
+using NServiceBus.Pipeline;
+
+namespace NServiceBus.Extensions.IntegrationTesting
+{
+    internal class AttachOutgoingLogicalMessageContextToActivity : Behavior<IOutgoingLogicalMessageContext>
+    {
+        public override Task Invoke(
+            IOutgoingLogicalMessageContext context,
+            Func<Task> next
+        )
+        {
+            Activity.Current?.AddTag("testing.outgoing.message.context", context);
+            return next();
+        }
+    }
+}

src/NServiceBus.Extensions.IntegrationTesting/EndpointConfigurationExtensions.cs

@@ -35,7 +35,21 @@ public static EndpointConfiguration ConfigureTestEndpoint(this EndpointConfigura
             transportConfigurationAction?.Invoke(transport);
 
             endpoint.PurgeOnStartup(true);
-            endpoint.DisableFeature<Audit>();
+            endpoint.DisableFeature<Features.Audit>();
+            endpoint.EnableOpenTelemetry();
+
+            endpoint.Pipeline.Register(
+                new AttachIncomingLogicalMessageContextToActivity(),
+                "Attach incoming logical message context as OpenTelemetry tags");
+
+            endpoint.Pipeline.Register(
+                new AttachOutgoingLogicalMessageContextToActivity(),
+                "Attach Outgoing Logical Message Context as OpenTelemetry tags");
+
+            endpoint.Pipeline.Register(
+                new AttachInvokeHandlerContextToActivity(),
+                "Attach invoke handler Context as OpenTelemetry tags");
+
             endpoint
                 .Recoverability()
                 .Immediate(i => i.NumberOfRetries(0))

src/NServiceBus.Extensions.IntegrationTesting/EndpointFixture.cs

@@ -4,7 +4,6 @@
 using System.Reactive.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using NServiceBus.Extensions.Diagnostics;
 using NServiceBus.Pipeline;
 using NServiceBus.Sagas;
 
@@ -72,58 +71,55 @@ private static async Task<ObservedMessageContexts> ExecuteAndWait<TMessageContex
 
             var messageReceivingTaskSource = new TaskCompletionSource<object>();
 
-            using var all = DiagnosticListener.AllListeners
-                .Subscribe(listener =>
+            // https://github.com/dotnet/runtime/blob/eccafbac942be9e3b06d48cff735fd6e50c3f25a/src/libraries/System.Diagnostics.DiagnosticSource/tests/ActivitySourceTests.cs#L134
+            using ActivityListener listener = new();
+            listener.ActivityStopped = (activitySource) =>
+            {
+                switch (activitySource.OperationName)
                 {
-                    switch (listener.Name)
-                    {
-                        case ActivityNames.IncomingLogicalMessage:
-                            var incomingObs = listener
-                                .Select(e => e.Value)
-                                .Cast<IIncomingLogicalMessageContext>();
-
-                            subscriptions.Add(incomingObs.Subscribe(e =>
-                            {
-                                incomingMessageContexts.Add(e);
-
-                                if (e is TMessageContext ctx && predicate(ctx))
-                                {
-                                    messageReceivingTaskSource.SetResult(null);
-                                }
-                            }));
-
-                            break;
-                        case ActivityNames.OutgoingLogicalMessage:
-                            var outgoingObs = listener
-                                .Select(e => e.Value)
-                                .Cast<IOutgoingLogicalMessageContext>();
-
-                            subscriptions.Add(outgoingObs.Subscribe((e) =>
-                            {
-                                outgoingMessageContexts.Add(e);
-
-                                if (e is TMessageContext ctx && predicate(ctx))
-                                {
-                                    messageReceivingTaskSource.SetResult(null);
-                                }
-                            }));
-
-                            break;
-                        case ActivityNames.InvokedHandler:
-                            var invokeHandlerObs = listener.Select(e => e.Value).Cast<IInvokeHandlerContext>();
-                            subscriptions.Add(invokeHandlerObs.Subscribe((e) =>
-                            {
-                                invokeHandlerContexts.Add(e);
-
-                                if (e is TMessageContext ctx && predicate(ctx))
-                                {
-                                    messageReceivingTaskSource.SetResult(null);
-                                }
-                            }));
-
-                            break;
-                    }
-                });
+                    case NsbActivityNames.IncomingMessageActivityName:
+                        var context = activitySource.GetTagItem("testing.incoming.message.context") as IIncomingLogicalMessageContext;
+
+                        incomingMessageContexts.Add(context);
+
+                        if (context is TMessageContext ctx && predicate(ctx))
+                        {
+                            messageReceivingTaskSource.SetResult(null);
+                        }
+
+                        break;
+                    case NsbActivityNames.OutgoingMessageActivityName:
+                        var outgoingContext = activitySource.GetTagItem("testing.outgoing.message.context") as IOutgoingLogicalMessageContext;
+
+                        outgoingMessageContexts.Add(outgoingContext);
+
+                        if (outgoingContext is TMessageContext ctx2 && predicate(ctx2))
+                        {
+                            messageReceivingTaskSource.SetResult(null);
+                        }
+
+
+                        break;
+                    case NsbActivityNames.InvokeHandlerActivityName:
+                        var handlerContext = activitySource.Parent.GetTagItem("testing.invoke.handler.context") as IInvokeHandlerContext;
+
+                        invokeHandlerContexts.Add(handlerContext);
+
+                        if (handlerContext is TMessageContext ctx3 && predicate(ctx3))
+                        {
+                            messageReceivingTaskSource.SetResult(null);
+                        }
+
+
+                        break;
+                }
+            };
+            listener.ShouldListenTo = _ => true;
+            listener.Sample = (ref ActivityCreationOptions<ActivityContext> activityOptions) => ActivitySamplingResult.AllData;
+
+            ActivitySource.AddActivityListener(listener);
+
+
 
             await testAction();
 

src/NServiceBus.Extensions.IntegrationTesting/NServiceBus.Extensions.IntegrationTesting.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFramework>netstandard2.0</TargetFramework>
+    <TargetFramework>net6.0</TargetFramework>
     <Authors>Jimmy Bogard</Authors>
     <Copyright>Jimmy Bogard</Copyright>
     <PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>
@@ -12,10 +12,11 @@
     <Deterministic>true</Deterministic>
     <ContinuousIntegrationBuild Condition="'$(GITHUB_ACTIONS)' == 'true'">true</ContinuousIntegrationBuild>
     <PackageTags>nservicebus;messaging;testing</PackageTags>
+    <OutputType>Library</OutputType>
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="NServiceBus.Extensions.Diagnostics" Version="2.0.0" />
+    <PackageReference Include="NServiceBus" Version="8.1.3" />
     <PackageReference Include="System.Reactive.Core" Version="5.0.0" />
     <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
     <PackageReference Include="MinVer" Version="2.5.0" PrivateAssets="All" />

src/NServiceBus.Extensions.IntegrationTesting/NsbActivityNames.cs

@@ -0,0 +1,11 @@
+namespace NServiceBus.Extensions.IntegrationTesting
+{
+    internal static class NsbActivityNames
+    {
+        public const string IncomingMessageActivityName = "NServiceBus.Diagnostics.ReceiveMessage";
+
+        public const string OutgoingMessageActivityName = "NServiceBus.Diagnostics.SendMessage";
+
+        public const string InvokeHandlerActivityName = "NServiceBus.Diagnostics.InvokeHandler";
+    }
+}

tests/NServiceBus.Extensions.IntegrationTesting.Tests/NServiceBus.Extensions.IntegrationTesting.Tests.csproj

@@ -9,7 +9,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.0.0" />
     <PackageReference Include="Microsoft.AspNetCore.Mvc.Testing" Version="6.0.2" />
-    <PackageReference Include="NServiceBus.Extensions.Hosting" Version="1.1.0" />
+    <PackageReference Include="NServiceBus.Extensions.Hosting" Version="2.0.0" />
     <PackageReference Include="Shouldly" Version="4.0.3" />
     <PackageReference Include="xunit" Version="2.4.1" />
     <PackageReference Include="xunit.runner.visualstudio" Version="2.4.3">