fix: clean up of dispose and delays in Dispatcher and Dispatcher tests

src/Paramore.Brighter.ServiceActivator/Dispatcher.cs

@@ -215,7 +215,7 @@ private void Start()
 
                     _logger.Value.InfoFormat("Dispatcher: Dispatcher starting {0} performers", _tasks.Count);
 
-                    while (!_tasks.IsEmpty)
+                    while (_tasks.Any())
                     {
                         try
                         {
@@ -228,19 +228,25 @@ private void Start()
                             if (consumer != null)
                             {
                                 _logger.Value.DebugFormat("Dispatcher: Removing a consumer with subscription name {0}", consumer.Name);
-                                consumer.Dispose();
 
-                                _consumers.TryRemove(consumer.Name, out consumer);
+                                if (_consumers.TryRemove(consumer.Name, out consumer))
+                                {
+                                    consumer.Dispose();
+                                }
                             }
 
-                            Task removedTask;
-                            _tasks.TryRemove(stoppingConsumer.Id, out removedTask);
+                            if (_tasks.TryRemove(stoppingConsumer.Id, out Task removedTask))
+                            {
+                                removedTask.Dispose();
+                            }
+
+                            stoppingConsumer.Dispose();
                         }
                         catch (AggregateException ae)
                         {
                             ae.Handle(ex =>
                             {
-                                _logger.Value.ErrorFormat("Dispatcher: Error on consumer; consumer shut down");
+                                _logger.Value.Error(ex, "Dispatcher: Error on consumer; consumer shut down");
                                 return true;
                             });
                         }
@@ -251,6 +257,11 @@ private void Start()
                 }
             },
             TaskCreationOptions.LongRunning);
+
+            while (State != DispatcherState.DS_RUNNING)
+            {
+                Task.Delay(100).Wait();
+            }
         }
 
         private IEnumerable<Consumer> CreateConsumers(IEnumerable<Subscription> connections)

tests/Paramore.Brighter.Core.Tests/MessageDispatch/When_a_message_dispatcher_Is_asked_to_connect_a_channel_and_handler_async.cs

@@ -1,4 +1,4 @@
-using System.Collections.Generic;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 using FluentAssertions;
 using Paramore.Brighter.Core.Tests.CommandProcessors.TestDoubles;
@@ -9,7 +9,7 @@
 
 namespace Paramore.Brighter.Core.Tests.MessageDispatch
 {
-    public class MessageDispatcherRoutingAsyncTests
+    public class MessageDispatcherRoutingAsyncTests : IAsyncLifetime
     {
         private readonly Dispatcher _dispatcher;
         private readonly FakeChannel _channel;
@@ -38,14 +38,12 @@ public MessageDispatcherRoutingAsyncTests()
             _channel.Enqueue(message);
 
             _dispatcher.State.Should().Be(DispatcherState.DS_AWAITING);
-            _dispatcher.Receive();
         }
 
-        [Fact(Timeout = 50000)]
-        public void When_a_message_dispatcher_is_asked_to_connect_a_channel_and_handler_async()
+        [Fact]
+        public async Task When_a_message_dispatcher_is_asked_to_connect_a_channel_and_handler_async()
         {
-            Task.Delay(5000).Wait();
-            _dispatcher.End().Wait();
+            await _dispatcher.End();
 
             //should have consumed the messages in the channel
             _channel.Length.Should().Be(0);
@@ -54,7 +52,23 @@ public void When_a_message_dispatcher_is_asked_to_connect_a_channel_and_handler_
             //should have dispatched a request
             _commandProcessor.Observe<MyEvent>().Should().NotBeNull();
             //should have published async
-            _commandProcessor.Commands.Should().Contain(ctype => ctype == CommandType.PublishAsync);
+            _commandProcessor.Commands.Should().Contain(commandType => commandType == CommandType.PublishAsync);
+        }
+
+        public Task InitializeAsync()
+        {
+            _dispatcher.Receive();
+            var completionSource = new TaskCompletionSource<IDispatcher>();
+            completionSource.SetResult(_dispatcher);
+
+            return completionSource.Task;
+        }
+
+        public Task DisposeAsync()
+        {
+            var completionSource = new TaskCompletionSource<object>();
+            completionSource.SetResult(null);
+            return completionSource.Task;
         }
-   }
+    }
 }

tests/Paramore.Brighter.Core.Tests/MessageDispatch/When_a_message_dispatcher_has_a_new_connection_added_while_running.cs

@@ -1,4 +1,4 @@
-#region Licence
+#region Licence
 /* The MIT License (MIT)
 Copyright © 2014 Ian Cooper <[email protected]>
 
@@ -34,7 +34,7 @@ THE SOFTWARE. */
 
 namespace Paramore.Brighter.Core.Tests.MessageDispatch
 {
-    public class DispatcherAddNewConnectionTests : IDisposable
+    public class DispatcherAddNewConnectionTests : IAsyncLifetime
     {
         private readonly Dispatcher _dispatcher;
         private readonly FakeChannel _channel;
@@ -59,20 +59,20 @@ public DispatcherAddNewConnectionTests()
             _channel.Enqueue(message);
 
             _dispatcher.State.Should().Be(DispatcherState.DS_AWAITING);
-            _dispatcher.Receive();
         }
 
         [Fact]
-        public void When_A_Message_Dispatcher_Has_A_New_Connection_Added_While_Running()
+        public async Task When_A_Message_Dispatcher_Has_A_New_Connection_Added_While_Running()
         {
             _dispatcher.Open(_newSubscription);
 
             var @event = new MyEvent();
             var message = new MyEventMessageMapper().MapToMessage(@event);
             _channel.Enqueue(message);
 
-            Task.Delay(1000).Wait();
 
+            await Task.Delay(500);
+
             //_should_have_consumed_the_messages_in_the_event_channel
             _channel.Length.Should().Be(0);
             //_should_have_a_running_state
@@ -83,10 +83,19 @@ public void When_A_Message_Dispatcher_Has_A_New_Connection_Added_While_Running()
             _dispatcher.Connections.Should().HaveCount(2);
         }
 
-        public void Dispose()
+
+        public Task InitializeAsync()
+        {
+            _dispatcher.Receive();
+            var completionSource = new TaskCompletionSource<IDispatcher>();
+            completionSource.SetResult(_dispatcher);
+
+            return completionSource.Task;
+        }
+
+        public Task DisposeAsync()
         {
-            if (_dispatcher?.State == DispatcherState.DS_RUNNING)
-                _dispatcher.End().Wait();
+            return _dispatcher?.End();
         }
     }
 }

tests/Paramore.Brighter.Core.Tests/MessageDispatch/When_a_message_dispatcher_is_asked_to_connect_a_channel_and_handler.cs

@@ -1,4 +1,4 @@
-#region Licence
+#region Licence
 /* The MIT License (MIT)
 Copyright © 2014 Ian Cooper <[email protected]>
 
@@ -33,7 +33,7 @@ THE SOFTWARE. */
 
 namespace Paramore.Brighter.Core.Tests.MessageDispatch
 {
-    public class MessageDispatcherRoutingTests
+    public class MessageDispatcherRoutingTests : IAsyncLifetime
     {
         private readonly Dispatcher _dispatcher;
         private readonly FakeChannel _channel;
@@ -61,14 +61,13 @@ public MessageDispatcherRoutingTests()
             _channel.Enqueue(message);
 
             _dispatcher.State.Should().Be(DispatcherState.DS_AWAITING);
-            _dispatcher.Receive();
         }
 
         [Fact]
-        public void When_A_Message_Dispatcher_Is_Asked_To_Connect_A_Channel_And_Handler()
+        public async Task When_A_Message_Dispatcher_Is_Asked_To_Connect_A_Channel_And_Handler()
         {
-            Task.Delay(1000).Wait();
-            _dispatcher.End().Wait();
+
+            await _dispatcher.End();
 
             //_should_have_consumed_the_messages_in_the_channel
             _channel.Length.Should().Be(0);
@@ -79,5 +78,21 @@ public void When_A_Message_Dispatcher_Is_Asked_To_Connect_A_Channel_And_Handler(
             //_should_have_published_async
             _commandProcessor.Commands.Should().Contain(ctype => ctype == CommandType.Publish);
         }
+
+        public Task InitializeAsync()
+        {
+            _dispatcher.Receive();
+            var completionSource = new TaskCompletionSource<IDispatcher>();
+            completionSource.SetResult(_dispatcher);
+
+            return completionSource.Task;
+        }
+
+        public Task DisposeAsync()
+        {
+            var completionSource = new TaskCompletionSource<object>();
+            completionSource.SetResult(null);
+            return completionSource.Task;
+        }
     }
 }

tests/Paramore.Brighter.Core.Tests/MessageDispatch/When_a_message_dispatcher_restarts_a_connection.cs

@@ -1,4 +1,4 @@
-#region Licence
+#region Licence
 /* The MIT License (MIT)
 Copyright © 2014 Ian Cooper <[email protected]>
 
@@ -33,14 +33,14 @@ THE SOFTWARE. */
 
 namespace Paramore.Brighter.Core.Tests.MessageDispatch
 {
-    public class MessageDispatcherResetConnection
+    public class MessageDispatcherResetConnection : IAsyncLifetime
     {
         private readonly Dispatcher _dispatcher;
         private readonly FakeChannel _channel;
         private readonly IAmACommandProcessor _commandProcessor;
         private readonly Subscription _subscription;
 
-        public MessageDispatcherResetConnection()
+        public MessageDispatcherResetConnection() 
         {
             _channel = new FakeChannel();
             _commandProcessor = new SpyCommandProcessor();
@@ -57,25 +57,34 @@ public MessageDispatcherResetConnection()
 
             _dispatcher.State.Should().Be(DispatcherState.DS_AWAITING);
             _dispatcher.Receive();
-            Task.Delay(1000).Wait();
-            _dispatcher.Shut(_subscription);
+
         }
 
-        [Fact(Skip = "Breaks test runner on Rider")]
-        public void When_A_Message_Dispatcher_Restarts_A_Connection()
+        [Fact]
+        public async Task When_A_Message_Dispatcher_Restarts_A_Connection()
         {
             _dispatcher.Open(_subscription);
 
             var @event = new MyEvent();
             var message = new MyEventMessageMapper().MapToMessage(@event);
             _channel.Enqueue(message);
 
-            _dispatcher.End().Wait();
+            await _dispatcher.End();
 
             //_should_have_consumed_the_messages_in_the_event_channel
             _channel.Length.Should().Be(0);
             //_should_have_a_stopped_state
             _dispatcher.State.Should().Be(DispatcherState.DS_STOPPED);
         }
+
+        public Task InitializeAsync()
+        {
+            return Task.Delay(500).ContinueWith(task =>   _dispatcher.Shut(_subscription));
+        }
+
+        public Task DisposeAsync()
+        {
+            return _dispatcher?.End();
+        }
     }
 }

tests/Paramore.Brighter.Core.Tests/MessageDispatch/When_a_message_dispatcher_restarts_a_connection_after_all_connections_have_stopped.cs

@@ -34,7 +34,7 @@ THE SOFTWARE. */
 
 namespace Paramore.Brighter.Core.Tests.MessageDispatch
 {
-    public class DispatcherRestartConnectionTests : IDisposable
+    public class DispatcherRestartConnectionTests : IAsyncLifetime
     {
         private readonly Dispatcher _dispatcher;
         private readonly FakeChannel _channel;
@@ -63,18 +63,19 @@ public DispatcherRestartConnectionTests()
             Task.Delay(250).Wait();
             _dispatcher.Shut("test");
             _dispatcher.Shut("newTest");
-            Task.Delay(1000).Wait();
+            Task.Delay(500).Wait();
             _dispatcher.Consumers.Should().HaveCount(0);
         }
 
         [Fact]
-        public void When_A_Message_Dispatcher_Restarts_A_Connection_After_All_Connections_Have_Stopped()
+        public async Task When_A_Message_Dispatcher_Restarts_A_Connection_After_All_Connections_Have_Stopped()
         {
             _dispatcher.Open("newTest");
             var @event = new MyEvent();
             var message = new MyEventMessageMapper().MapToMessage(@event);
             _channel.Enqueue(message);
-            Task.Delay(1000).Wait();
+
+            await Task.Delay(500);
 
             //_should_have_consumed_the_messages_in_the_event_channel
             _channel.Length.Should().Be(0);
@@ -85,11 +86,17 @@ public void When_A_Message_Dispatcher_Restarts_A_Connection_After_All_Connection
             //_should_have_two_connections
             _dispatcher.Connections.Should().HaveCount(2);
         }
+        public Task InitializeAsync()
+        {
+            var completionSource = new TaskCompletionSource<IDispatcher>();
+            completionSource.SetResult(_dispatcher);
+
+            return completionSource.Task;
+        }
 
-        public void Dispose()
+        public Task DisposeAsync()
         {
-            if (_dispatcher?.State == DispatcherState.DS_RUNNING)
-                _dispatcher.End().Wait();
+            return _dispatcher.End();
         }
     }
 }