diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 3b3d815..daa05a7 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 185f3b4..f6cc203 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 0000000..f2c7f64 --- /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 b549cef..f4efe40 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 0000000..1240470 --- /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 e9876fe..21f36b1 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(() =>