Merge master into v7 (deadlock backport tests)

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.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<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var disposed = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            IDisposable subscription = null;
+            subscription = _source.Messages.OfType<TestMessage>().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));
+        }
     }
 }

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<ITestMessage>()
+                        .Named("Obvs.TestService.DisposeInCallback")
+                        .ConnectToBroker(TestBroker.Uri)
+                        .SerializedAsJson()
+                        .AsClientAndServer()
+                    .Create();
+            var disposed = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            using var subscription = serviceBus.Commands.OfType<TestCommand>().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()
         {