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.Kafka.Tests/TestMessageSendAndReceive.cs b/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs index f62b1af..b37c055 100644 --- a/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs +++ b/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs @@ -119,5 +119,27 @@ public async Task TestMessagePublishedStraightAfterSubscribingIsReceived() Assert.Contains(messages, message => message.Data == data); } + + [Fact] + [Trait("Category", "Explicit")] + public async Task ShouldNotDeadlockWhenASubscriberDisposesItsSubscription() + { + // Code that runs on receiving a message may shut the app down. If messages arrived on the consume + // thread, disposing the subscription there would wait for that same thread, forever. + var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var disposed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + IDisposable subscription = null; + subscription = _source.Messages.OfType().Subscribe(_ => + { + received.TrySetResult(true); + subscription.Dispose(); + disposed.TrySetResult(true); + }); + + await _publisher.PublishAsync(new TestMessage($"dispose in callback {Guid.NewGuid()}")); + + Assert.True(await received.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); + Assert.True(await disposed.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); + } } } \ No newline at end of file diff --git a/Obvs.RabbitMQ.Tests/TestServiceBus.cs b/Obvs.RabbitMQ.Tests/TestServiceBus.cs index f062580..491e652 100644 --- a/Obvs.RabbitMQ.Tests/TestServiceBus.cs +++ b/Obvs.RabbitMQ.Tests/TestServiceBus.cs @@ -15,6 +15,31 @@ namespace Obvs.RabbitMQ.Tests { public class TestServiceBus { + [Fact, Trait("Category", "Explicit")] + public async Task ShouldNotDeadlockWhenASubscriberDisposesTheBus() + { + // Code that runs on receiving a message (including code after awaiting one) may shut the app down. + // If messages arrived on RabbitMQ's consumer-dispatch thread, closing the connection there would + // wait for that same thread, forever. + IServiceBus serviceBus = ServiceBus.Configure() + .WithRabbitMQEndpoints() + .Named("Obvs.TestService.DisposeInCallback") + .ConnectToBroker(TestBroker.Uri) + .SerializedAsJson() + .AsClientAndServer() + .Create(); + var disposed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var subscription = serviceBus.Commands.OfType().Subscribe(_ => + { + ((IDisposable)serviceBus).Dispose(); + disposed.SetResult(true); + }); + + await serviceBus.SendAsync(new TestCommand { Id = 1 }); + + Assert.True(await disposed.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); + } + [Fact, Trait("Category", "Explicit")] public async Task ShouldSendAndReceiveMessagesOverServiceBus() {