From 2a3c4cc5b640bf9caef573da5fca93eef3b9d7f8 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 14:17:51 -0700 Subject: [PATCH] ActiveMQ: deliver messages to subscribers off the session's dispatch thread MessageSource passed each message to subscribers synchronously on the NMS session's dispatch thread. Code that ran after awaiting a message could then run on that thread, and deadlock it by disposing the bus: stopping the connection waits for the dispatch thread, forever. NMS skips that wait only for code in the dispatch's own execution context, which an await continuation isn't. Messages are now handed to subscribers on the task pool, in order. TestServiceBus.ShouldNotDeadlockWhenCodeAfterAwaitingAMessageDisposesTheBus hung before this change. Five TestMessageSource tests now wait for delivery instead of expecting it on the listener's thread. CI runs the ActiveMQ broker tests against apache/activemq-classic. Backport of the fix in v7 PR #121. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/dotnet.yml | 5 ++ .../Obvs.ActiveMQ.Tests.csproj | 1 + Obvs.ActiveMQ.Tests/TestBroker.cs | 10 ++++ Obvs.ActiveMQ.Tests/TestMessageSource.cs | 19 +++++-- Obvs.ActiveMQ.Tests/TestServiceBus.cs | 53 +++++++++++++++++++ Obvs.ActiveMQ/MessageSource.cs | 5 ++ 6 files changed, 88 insertions(+), 5 deletions(-) create mode 100644 Obvs.ActiveMQ.Tests/TestBroker.cs create mode 100644 Obvs.ActiveMQ.Tests/TestServiceBus.cs diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 3b3d815d..daa05a76 100644 --- a/.github/workflows/dotnet.yml +++ b/.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. diff --git a/Obvs.ActiveMQ.Tests/Obvs.ActiveMQ.Tests.csproj b/Obvs.ActiveMQ.Tests/Obvs.ActiveMQ.Tests.csproj index 185f3b46..f6cc2035 100644 --- a/Obvs.ActiveMQ.Tests/Obvs.ActiveMQ.Tests.csproj +++ b/Obvs.ActiveMQ.Tests/Obvs.ActiveMQ.Tests.csproj @@ -19,5 +19,6 @@ + \ No newline at end of file diff --git a/Obvs.ActiveMQ.Tests/TestBroker.cs b/Obvs.ActiveMQ.Tests/TestBroker.cs new file mode 100644 index 00000000..f2c7f64c --- /dev/null +++ b/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"; + } +} diff --git a/Obvs.ActiveMQ.Tests/TestMessageSource.cs b/Obvs.ActiveMQ.Tests/TestMessageSource.cs index b549cefd..f4efe40f 100644 --- a/Obvs.ActiveMQ.Tests/TestMessageSource.cs +++ b/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._)).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._)).MustHaveHappened(); - A.CallTo(() => _observer.OnNext(message)).MustHaveHappened(); + AssertDelivered(message); } } } \ No newline at end of file diff --git a/Obvs.ActiveMQ.Tests/TestServiceBus.cs b/Obvs.ActiveMQ.Tests/TestServiceBus.cs new file mode 100644 index 00000000..12404703 --- /dev/null +++ b/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() + .Named("Obvs.TestService.DisposeAfterAwait") + .ConnectToBroker(TestBroker.Uri) + .SerializedAsJson() + .AsClientAndServer() + .Create(); + // Continuations run inline, wherever the message is received + var received = new TaskCompletionSource(); + using var subscription = serviceBus.Commands.OfType().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); + } + } +} diff --git a/Obvs.ActiveMQ/MessageSource.cs b/Obvs.ActiveMQ/MessageSource.cs index e9876fe0..21f36b13 100644 --- a/Obvs.ActiveMQ/MessageSource.cs +++ b/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 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(() =>