ActiveMQ: deliver messages to subscribers off the session's dispatch thread

.github/workflows/dotnet.yml

@@ -49,6 +49,9 @@ jobs:
         options: >-
           --health-cmd "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092"
           --health-interval 5s --health-timeout 10s --health-retries 20
+      activemq:
+        image: apache/activemq-classic:6.2.0
+        ports: ['61616:61616']
 
     steps:
     - name: Checkout
@@ -82,11 +85,13 @@ jobs:
         RABBITMQ_URI: amqp://localhost:5672
         NATS_URL: nats://localhost:4222
         KAFKA_BOOTSTRAP_SERVERS: localhost:9092
+        ACTIVEMQ_URI: activemq:tcp://localhost:61616
       run: |
         dotnet test --project Obvs.RabbitMQ.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
         dotnet test --project Obvs.NATS.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
         dotnet test --project Obvs.Kafka.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
         dotnet test --project Obvs.NetMQ.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
+        dotnet test --project Obvs.ActiveMQ.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
     - name: Test
       # xUnit v3 runs on Microsoft.Testing.Platform (global.json). Exit code 8 means a project ran no tests,
       # as the transport test projects only contain Explicit tests.

Obvs.ActiveMQ.Tests/Obvs.ActiveMQ.Tests.csproj

@@ -19,5 +19,6 @@
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\Obvs.ActiveMQ\Obvs.ActiveMQ.csproj" />
+    <ProjectReference Include="..\Obvs.Serialization.Json\Obvs.Serialization.Json.csproj" />
   </ItemGroup>
 </Project>

Obvs.ActiveMQ.Tests/TestBroker.cs

@@ -0,0 +1,10 @@
+using System;
+
+namespace Obvs.ActiveMQ.Tests
+{
+    internal static class TestBroker
+    {
+        // Set ACTIVEMQ_URI to run the Explicit tests against a broker other than a default local install
+        public static string Uri => Environment.GetEnvironmentVariable("ACTIVEMQ_URI") ?? "activemq:tcp://localhost:61616";
+    }
+}

Obvs.ActiveMQ.Tests/TestMessageSource.cs

@@ -1,7 +1,9 @@
 using System;
 using System.Collections;
 using System.IO;
+using System.Linq;
 using System.Text;
+using System.Threading;
 using System.Xml.Serialization;
 using Apache.NMS;
 using Apache.NMS.Util;
@@ -58,6 +60,13 @@ public TestMessageSource()
                 _acknowledgementMode);
         }
 
+        // Messages reach subscribers on the task pool, so wait for delivery rather than expecting it on the listener's thread
+        private void AssertDelivered(ITestMessage message)
+        {
+            SpinWait.SpinUntil(() => Fake.GetCalls(_observer).Any(call => call.Method.Name == nameof(_observer.OnNext) && ReferenceEquals(call.Arguments[0], message)), TimeSpan.FromSeconds(5));
+            A.CallTo(() => _observer.OnNext(message)).MustHaveHappened();
+        }
+
         [Fact]
         public void ShouldConnectToBrokerWhenSubscribedTo()
         {
@@ -143,7 +152,7 @@ public void ShouldDeserializeAndPublishMessageWhenReceived()
             _source.Messages.Subscribe(_observer);
             consumer.Listener += Raise.FreeForm.With((Apache.NMS.IMessage) bytesMessage);
 
-            A.CallTo(() => _observer.OnNext(message)).MustHaveHappened();
+            AssertDelivered(message);
         }
 
         [Fact]
@@ -161,7 +170,7 @@ public void ShouldDeserializeByteMessagesAndPublishMessageWhenReceived()
             _source.Messages.Subscribe(_observer);
             consumer.Listener += Raise.FreeForm.With((Apache.NMS.IMessage) bytesMessage);
 
-            A.CallTo(() => _observer.OnNext(message)).MustHaveHappened();
+            AssertDelivered(message);
         }
 
         [Fact]
@@ -188,7 +197,7 @@ public void ShouldDeserializeAndPublishMessageOfRightTypeName()
             _source.Messages.Subscribe(_observer);
             consumer.Listener += Raise.FreeForm.With((Apache.NMS.IMessage) bytesMessage);
 
-            A.CallTo(() => _observer.OnNext(message)).MustHaveHappened();
+            AssertDelivered(message);
         }
 
         [Fact]
@@ -214,7 +223,7 @@ public void ShouldProcessMessagesWithoutTypeNameIfOnlyOneDeserializerSupplied()
             consumer.Listener += Raise.FreeForm.With((Apache.NMS.IMessage) bytesMessage);
 
             A.CallTo(() => _deserializer.Deserialize(A<Stream>._)).MustHaveHappened();
-            A.CallTo(() => _observer.OnNext(message)).MustHaveHappened();
+            AssertDelivered(message);
         }
 
         [Fact]
@@ -248,7 +257,7 @@ public void ShouldProcessMessagesAccordingToFilter()
             consumer.Listener += Raise.FreeForm.With((Apache.NMS.IMessage) bytesMessage);
 
             A.CallTo(() => _deserializer.Deserialize(A<Stream>._)).MustHaveHappened();
-            A.CallTo(() => _observer.OnNext(message)).MustHaveHappened();
+            AssertDelivered(message);
         }
     }
 }

Obvs.ActiveMQ.Tests/TestServiceBus.cs

@@ -0,0 +1,53 @@
+using System;
+using System.Reactive.Linq;
+using System.Threading.Tasks;
+using Obvs.ActiveMQ.Configuration;
+using Obvs.Configuration;
+using Obvs.Serialization.Json.Configuration;
+using Obvs.Types;
+using Xunit;
+
+namespace Obvs.ActiveMQ.Tests
+{
+    public class TestServiceBus
+    {
+        public interface IServiceBusTestMessage : IMessage
+        {
+        }
+
+        public class TestCommand : ICommand, IServiceBusTestMessage
+        {
+            public int Id { get; set; }
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldNotDeadlockWhenCodeAfterAwaitingAMessageDisposesTheBus()
+        {
+            // Code after awaiting a message may shut the app down. If messages arrived on the NMS session's
+            // dispatch thread, that code would run there too, and stopping the connection would wait for that
+            // same thread, forever. (NMS only skips the wait for code in the dispatch's own execution context,
+            // which code after an await is not.)
+            IServiceBus serviceBus = ServiceBus.Configure()
+                    .WithActiveMQEndpoints<IServiceBusTestMessage>()
+                        .Named("Obvs.TestService.DisposeAfterAwait")
+                        .ConnectToBroker(TestBroker.Uri)
+                        .SerializedAsJson()
+                        .AsClientAndServer()
+                    .Create();
+            // Continuations run inline, wherever the message is received
+            var received = new TaskCompletionSource<TestCommand>();
+            using var subscription = serviceBus.Commands.OfType<TestCommand>().Subscribe(command => received.TrySetResult(command));
+
+            async Task DisposeOnReceiving()
+            {
+                await received.Task.ConfigureAwait(false);
+                ((IDisposable)serviceBus).Dispose();
+            }
+
+            var disposed = DisposeOnReceiving();
+            await serviceBus.SendAsync(new TestCommand { Id = 1 });
+
+            await disposed.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
+        }
+    }
+}

Obvs.ActiveMQ/MessageSource.cs

@@ -4,6 +4,7 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
+using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
 using Apache.NMS;
@@ -50,6 +51,10 @@ public MessageSource(Lazy<IConnection> lazyConnection,
                         .Select(message => new { message, deserializer = GetDeserializer(message) })
                         .Where(msg => msg.deserializer != null)
                         .Select(msg => DeserializeAndAcknowledge(msg.message, msg.deserializer))
+                        // Hand messages to subscribers on the task pool, in order. Subscriber code running on the
+                        // session's dispatch thread could otherwise deadlock it, for example by disposing the bus,
+                        // which stops the connection and waits for that thread.
+                        .ObserveOn(TaskPoolScheduler.Default)
                         .Subscribe(observer);
 
                     return Disposable.Create(() =>