Fix Theading issue on Implicit Clear

This change fixes the following
  - Bulk Clear was using ProducerRegistry.LookupBy not LookupByDefailt
  - In the Implicit clearing Threads the Monitor was never releasing so it would never allow entry after the first run
  - TimedOutboxSweeper would Fail on Async as it can't be RunSynchronously ad the thread was usually already exited

Brighter.sln

@@ -1,7 +1,7 @@
 
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio Version 16
-VisualStudioVersion = 16.0.28922.388
+# Visual Studio Version 17
+VisualStudioVersion = 17.0.32112.339
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "samples", "samples", "{235DE1F1-E71B-4817-8E27-3B34FF006E4C}"
 EndProject
@@ -223,17 +223,17 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "GreetingsWorker", "samples\
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paramore.Brighter.PostgreSql", "src\Paramore.Brighter.PostgreSql\Paramore.Brighter.PostgreSql.csproj", "{08E6D0F8-B6CE-454F-8761-77731D99F743}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paramore.Brighter.PostgreSql.EntityFrameworkCore", "src\Paramore.Brighter.PostgreSql.EntityFrameworkCore\Paramore.Brighter.PostgreSql.EntityFrameworkCore.csproj", "{AA85493A-4120-4DA0-BAA5-CBF34D238A64}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paramore.Brighter.PostgreSql.EntityFrameworkCore", "src\Paramore.Brighter.PostgreSql.EntityFrameworkCore\Paramore.Brighter.PostgreSql.EntityFrameworkCore.csproj", "{AA85493A-4120-4DA0-BAA5-CBF34D238A64}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "GreetingsWeb", "samples\WebAPI_EFCore\GreetingsWeb\GreetingsWeb.csproj", "{BC4C3DF7-52C0-41EA-98DD-70E402AD5116}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "GreetingsWeb", "samples\WebAPI_EFCore\GreetingsWeb\GreetingsWeb.csproj", "{BC4C3DF7-52C0-41EA-98DD-70E402AD5116}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SalutationPorts", "samples\WebAPI_EFCore\SalutationPorts\SalutationPorts.csproj", "{5A6A95C0-F82E-49DD-B4C6-98D7A765ECFF}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "SalutationPorts", "samples\WebAPI_EFCore\SalutationPorts\SalutationPorts.csproj", "{5A6A95C0-F82E-49DD-B4C6-98D7A765ECFF}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SalutationEntities", "samples\WebAPI_EFCore\SalutationEntities\SalutationEntities.csproj", "{F701369D-EDA3-4407-8655-6B81DD6EBCBA}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "SalutationEntities", "samples\WebAPI_EFCore\SalutationEntities\SalutationEntities.csproj", "{F701369D-EDA3-4407-8655-6B81DD6EBCBA}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Salutations_MySqlMigrations", "samples\WebAPI_EFCore\Salutations_MySqlMigrations\Salutations_MySqlMigrations.csproj", "{82E64F30-8D74-4E01-A974-5A78EBAD916C}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Salutations_MySqlMigrations", "samples\WebAPI_EFCore\Salutations_MySqlMigrations\Salutations_MySqlMigrations.csproj", "{82E64F30-8D74-4E01-A974-5A78EBAD916C}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Salutations_SqliteMigrations", "samples\WebAPI_EFCore\Salutations_SqliteMigrations\Salutations_SqliteMigrations.csproj", "{05647D1B-87A3-4440-B468-82866B206E49}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Salutations_SqliteMigrations", "samples\WebAPI_EFCore\Salutations_SqliteMigrations\Salutations_SqliteMigrations.csproj", "{05647D1B-87A3-4440-B468-82866B206E49}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -1229,6 +1229,30 @@ Global
 		{93589653-2B49-4818-BE98-FE6F16EC72EC}.Release|Mixed Platforms.Build.0 = Release|Any CPU
 		{93589653-2B49-4818-BE98-FE6F16EC72EC}.Release|x86.ActiveCfg = Release|Any CPU
 		{93589653-2B49-4818-BE98-FE6F16EC72EC}.Release|x86.Build.0 = Release|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Debug|x86.Build.0 = Debug|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Release|Any CPU.Build.0 = Release|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Release|Mixed Platforms.Build.0 = Release|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Release|x86.ActiveCfg = Release|Any CPU
+		{08E6D0F8-B6CE-454F-8761-77731D99F743}.Release|x86.Build.0 = Release|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Debug|x86.Build.0 = Debug|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Release|Any CPU.Build.0 = Release|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Release|Mixed Platforms.Build.0 = Release|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Release|x86.ActiveCfg = Release|Any CPU
+		{AA85493A-4120-4DA0-BAA5-CBF34D238A64}.Release|x86.Build.0 = Release|Any CPU
 		{BC4C3DF7-52C0-41EA-98DD-70E402AD5116}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{BC4C3DF7-52C0-41EA-98DD-70E402AD5116}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{BC4C3DF7-52C0-41EA-98DD-70E402AD5116}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU

src/Paramore.Brighter.Extensions.Hosting/TimedOutboxSweeper.cs

@@ -36,19 +36,31 @@ private void DoWork(object state)
             s_logger.LogInformation("Outbox Sweeper looking for unsent messages");
 
             var scope = _serviceScopeFactory.CreateScope();
-            IAmACommandProcessor commandProcessor = scope.ServiceProvider.GetService<IAmACommandProcessor>();
+            try
+            {
+                IAmACommandProcessor commandProcessor = scope.ServiceProvider.GetService<IAmACommandProcessor>();
 
-            var outBoxSweeper = new OutboxSweeper(
-                milliSecondsSinceSent: _options.MinimumMessageAge,
-                commandProcessor: commandProcessor, 
-                _options.BatchSize,
-                _options.UseBulk);
-
-            if(_options.UseBulk)
-                outBoxSweeper.SweepAsync(CancellationToken.None).RunSynchronously();
-            else
-                outBoxSweeper.Sweep();
+                var outBoxSweeper = new OutboxSweeper(
+                    milliSecondsSinceSent: _options.MinimumMessageAge,
+                    commandProcessor: commandProcessor,
+                    _options.BatchSize,
+                    _options.UseBulk);
 
+                if (_options.UseBulk)
+                    outBoxSweeper.SweepAsyncOutbox();
+                else
+                    outBoxSweeper.Sweep();
+            }
+            catch (Exception e)
+            {
+                s_logger.LogError(e, "Error while sweeping the outbox.");
+                throw;
+            }
+            finally
+            {
+                scope.Dispose();
+            }
+
             s_logger.LogInformation("Outbox Sweeper sleeping");
         }
 

src/Paramore.Brighter/ExternalBusServices.cs

@@ -26,13 +26,13 @@ internal class ExternalBusServices : IDisposable
 
         internal IAmAProducerRegistry ProducerRegistry { get; set; }
 
-        private static SemaphoreSlim _clearSemaphoreToken = new SemaphoreSlim(1, 1);
+        private static readonly SemaphoreSlim _clearSemaphoreToken = new SemaphoreSlim(1, 1);
+        private static readonly SemaphoreSlim _backgroundClearSemaphoreToken = new SemaphoreSlim(1, 1);
 
         private DateTime _lastOutStandingMessageCheckAt = DateTime.UtcNow;
 
         //Used to checking the limit on outstanding messages for an Outbox. We throw at that point. Writes to the static bool should be made thread-safe by locking the object
         private readonly object _checkOutStandingMessagesObject = new object();
-        private readonly object _implicitClearMessagesObject = new object();
 
         //Uses -1 to indicate no outbox and will thus force a throw on a failed publish
         private int _outStandingCount;
@@ -198,7 +198,7 @@ internal void ClearOutbox(int amountToClear, int minimumAge, bool useAsync, bool
 
         private async Task BackgroundDispatchUsingSync(int amountToClear, int minimumAge)
         {
-            if (Monitor.TryEnter(_implicitClearMessagesObject))
+            if (await _backgroundClearSemaphoreToken.WaitAsync(TimeSpan.Zero))
             {
                 await _clearSemaphoreToken.WaitAsync(CancellationToken.None);
                 try
@@ -209,44 +209,61 @@ private async Task BackgroundDispatchUsingSync(int amountToClear, int minimumAge
                     Dispatch(messages);
                     s_logger.LogInformation("Messages have been cleared");
                 }
+                catch (Exception e)
+                {
+                    s_logger.LogError(e, "Error while dispatching from outbox");
+                }
                 finally
                 {
                     _clearSemaphoreToken.Release();
-                    Monitor.Exit(_implicitClearMessagesObject);
+                    _backgroundClearSemaphoreToken.Release();
                 }
 
                 CheckOutstandingMessages();
             }
+            else
+            {
+                s_logger.LogInformation("Skipping dispatch of messages as another thread is running");
+            }
         }
 
         private async Task BackgroundDispatchUsingAsync(int amountToClear, int minimumAge, bool useBulk)
         {
-            if (Monitor.TryEnter(_implicitClearMessagesObject))
+
+            if (await _backgroundClearSemaphoreToken.WaitAsync(TimeSpan.Zero))
             {
                 await _clearSemaphoreToken.WaitAsync(CancellationToken.None);
                 try
                 {
                     var messages =
                         await AsyncOutbox.OutstandingMessagesAsync(minimumAge, amountToClear);
-                    
+
                     s_logger.LogInformation("Found {NumberOfMessages} to clear out of amount {AmountToClear}",
                         messages.Count(), amountToClear);
 
                     if (useBulk)
                         await BulkDispatchAsync(messages, CancellationToken.None);
                     else
                         await DispatchAsync(messages, false, CancellationToken.None);
-                    
+
                     s_logger.LogInformation("Messages have been cleared");
                 }
+                catch (Exception e)
+                {
+                    s_logger.LogError(e, "Error while dispatching from outbox");
+                }
                 finally
                 {
                     _clearSemaphoreToken.Release();
-                    Monitor.Exit(_implicitClearMessagesObject);
+                    _backgroundClearSemaphoreToken.Release();
                 }
 
                 CheckOutstandingMessages();
             }
+            else
+            {
+                s_logger.LogInformation("Skipping dispatch of messages as another thread is running");
+            }
         }
 
         private void Dispatch(IEnumerable<Message> posts)
@@ -323,7 +340,7 @@ private async Task BulkDispatchAsync(IEnumerable<Message> posts, CancellationTok
 
             foreach (var topicBatch in messagesByTopic)
             {
-                var producer = ProducerRegistry.LookupBy(topicBatch.Key);
+                var producer = ProducerRegistry.LookupByOrDefault(topicBatch.Key);
 
                 if (producer is IAmABulkMessageProducerAsync bulkMessageProducer)
                 {

src/Paramore.Brighter/OutboxSweeper.cs

@@ -39,5 +39,10 @@ public Task SweepAsync(CancellationToken cancellationToken = default)
 
             return Task.CompletedTask;
         }
+
+        public void SweepAsyncOutbox()
+        {
+            _commandProcessor.ClearAsyncOutbox(_batchSize, _milliSecondsSinceSent, _useBulk);
+        }
     }
 }

tests/Paramore.Brighter.Core.Tests/CommandProcessors/TestDoubles/FakeMessageProducer.cs

@@ -30,15 +30,14 @@ THE SOFTWARE. */
 
 namespace Paramore.Brighter.Core.Tests.CommandProcessors.TestDoubles
 {
-    public class FakeMessageProducer : IAmAMessageProducerSync, IAmAMessageProducerAsync, ISupportPublishConfirmation, IAmABulkMessageProducerAsync
+    public class FakeMessageProducer : IAmAMessageProducerSync, IAmAMessageProducerAsync, IAmABulkMessageProducerAsync
     {
-        public event Action<bool, Guid> OnMessagePublished;
         public int MaxOutStandingMessages { get; set; } = -1;
         public int MaxOutStandingCheckIntervalMilliSeconds { get; set; } = 0;
 
         public List<Message> SentMessages = new List<Message>();
         public bool MessageWasSent { get; set; }
-        
+
         public void Dispose() { }
 
         public Task SendAsync(Message message)
@@ -52,7 +51,7 @@ public async IAsyncEnumerable<Guid[]> SendAsync(IEnumerable<Message> messages, [
         {
             foreach (var msg in messages)
             {
-                yield return new[] {msg.Id};
+                yield return new[] { msg.Id };
             }
             MessageWasSent = true;
             SentMessages.AddRange(messages);
@@ -62,13 +61,11 @@ public void Send(Message message)
         {
             MessageWasSent = true;
             SentMessages.Add(message);
-            OnMessagePublished?.Invoke(true, message.Id);
         }
-        
+
         public void SendWithDelay(Message message, int delayMilliseconds = 0)
         {
             Send(message);
         }
-
-   }
+    }
 }

tests/Paramore.Brighter.Core.Tests/CommandProcessors/TestDoubles/FakeMessageProducerWithPublishConfirmation.cs

@@ -0,0 +1,40 @@
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2015 Toby Henderson <hendersont@gmail.com>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the “Software”), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE. */
+
+#endregion
+
+using System;
+
+namespace Paramore.Brighter.Core.Tests.CommandProcessors.TestDoubles
+{
+    public class FakeMessageProducerWithPublishConfirmation : FakeMessageProducer, ISupportPublishConfirmation
+    {
+        public event Action<bool, Guid> OnMessagePublished;
+
+        public new void Send(Message message)
+        {
+            MessageWasSent = true;
+            SentMessages.Add(message);
+            OnMessagePublished?.Invoke(true, message.Id);
+        }
+    }
+}

tests/Paramore.Brighter.Core.Tests/CommandProcessors/TestDoubles/FakeOutboxSync.cs

@@ -122,10 +122,13 @@ public Task MarkDispatchedAsync(Guid id, DateTime? dispatchedAt = null, Dictiona
             return tcs.Task;
         }
 
-        public Task MarkDispatchedAsync(IEnumerable<Guid> ids, DateTime? dispatchedAt = null, Dictionary<string, object> args = null,
+        public async Task MarkDispatchedAsync(IEnumerable<Guid> ids, DateTime? dispatchedAt = null, Dictionary<string, object> args = null,
             CancellationToken cancellationToken = default(CancellationToken))
         {
-            throw new NotImplementedException();
+            foreach (var id in ids)
+            {
+                await MarkDispatchedAsync(id, dispatchedAt, args, cancellationToken);
+            }
         }
 
         public Task<IEnumerable<Message>> OutstandingMessagesAsync(double millSecondsSinceSent, int pageSize = 100, int pageNumber = 1,

tests/Paramore.Brighter.Core.Tests/CommandProcessors/When_Bulk_Clearing_The_PostBox_On_The_Command_Processor _Async.cs

@@ -43,15 +43,15 @@ public class CommandProcessorPostBoxBulkClearAsyncTests : IDisposable
         private readonly Message _message;
         private readonly Message _message2;
         private readonly FakeOutboxSync _fakeOutboxSync;
-        private readonly FakeMessageProducer _fakeMessageProducer;
+        private readonly FakeMessageProducerWithPublishConfirmation _fakeMessageProducerWithPublishConfirmation;
 
         public CommandProcessorPostBoxBulkClearAsyncTests()
         {
             var myCommand = new MyCommand{ Value = "Hello World"};
             var myCommand2 = new MyCommand { Value = "Hello World 2" };
 
             _fakeOutboxSync = new FakeOutboxSync();
-            _fakeMessageProducer = new FakeMessageProducer();
+            _fakeMessageProducerWithPublishConfirmation = new FakeMessageProducerWithPublishConfirmation();
 
             var topic = "MyCommand";
             var topic2 = "MyCommand2";
@@ -82,7 +82,7 @@ public CommandProcessorPostBoxBulkClearAsyncTests()
                 new PolicyRegistry { { CommandProcessor.RETRYPOLICYASYNC, retryPolicy }, { CommandProcessor.CIRCUITBREAKERASYNC, circuitBreakerPolicy } },
                 messageMapperRegistry,
                 _fakeOutboxSync,
-                new ProducerRegistry(new Dictionary<string, IAmAMessageProducer>() { { topic, _fakeMessageProducer }, { topic2, _fakeMessageProducer } }));
+                new ProducerRegistry(new Dictionary<string, IAmAMessageProducer>() { { topic, _fakeMessageProducerWithPublishConfirmation }, { topic2, _fakeMessageProducerWithPublishConfirmation } }));
         }
 
         [Fact]
@@ -96,15 +96,15 @@ public async Task When_Clearing_The_PostBox_On_The_Command_Processor_Async()
             await Task.Delay(1000);
 
             //_should_send_a_message_via_the_messaging_gateway
-            _fakeMessageProducer.MessageWasSent.Should().BeTrue();
+            _fakeMessageProducerWithPublishConfirmation.MessageWasSent.Should().BeTrue();
 
-            var sentMessage = _fakeMessageProducer.SentMessages[0];
+            var sentMessage = _fakeMessageProducerWithPublishConfirmation.SentMessages[0];
             sentMessage.Should().NotBeNull();
             sentMessage.Id.Should().Be(_message.Id);
             sentMessage.Header.Topic.Should().Be(_message.Header.Topic);
             sentMessage.Body.Value.Should().Be(_message.Body.Value);
 
-            var sentMessage2 = _fakeMessageProducer.SentMessages[1];
+            var sentMessage2 = _fakeMessageProducerWithPublishConfirmation.SentMessages[1];
             sentMessage2.Should().NotBeNull();
             sentMessage2.Id.Should().Be(_message2.Id);
             sentMessage2.Header.Topic.Should().Be(_message2.Header.Topic);