PAN-3799 Add ConditionWaiter to the messagequeue so that a max queue length can be assigned to put back pressure on the publisher

Author: steelheaddigital

src/Learning.MessageQueue/AsyncConditionWaiter.cs

@@ -0,0 +1,36 @@
+using System;
+using System.Threading.Tasks;
+using System.Threading;
+
+namespace Learning.MessageQueue
+{
+    public class AsyncConditionWaiter
+    {
+        private readonly Func<Task<bool>> _conditionCheck;
+        private readonly int _pollingInterval;
+        private TaskCompletionSource<object> _tcs;
+
+        public AsyncConditionWaiter(Func<Task<bool>> conditionCheck, int pollingInterval = 100)
+        {
+            _conditionCheck = conditionCheck;
+            _pollingInterval = pollingInterval;
+            _tcs = new TaskCompletionSource<object>();
+        }
+
+        public async Task WaitForConditionAsync(CancellationToken cancellationToken = default)
+        {
+            while (!await _conditionCheck())
+            {
+                await Task.Delay(_pollingInterval, cancellationToken); // Wait before checking again to avoid busy-waiting
+            }
+
+            // Condition is met, signal completion
+            _tcs.SetResult(null);
+        }
+
+        public Task WaitForCompletionAsync()
+        {
+            return _tcs.Task;
+        }
+    }
+}

src/Learning.MessageQueue/IMessageQueue.cs

@@ -7,6 +7,8 @@ public interface IMessageQueue
     {
         Task PublishAsync(IMessage message);
 
-        Task PublishAsync(string serializedMessage, string messageId, string messageType);
+        Task PublishAsync(IMessage message, int? capacity = null);
+
+        Task PublishAsync(string serializedMessage, string messageId, string messageType, int? capacity = null);
     }
 }

src/Learning.MessageQueue/RedisMessageQueue.cs

@@ -35,23 +35,28 @@ public RedisMessageQueue(IRedisClient redis, RedisEventStoreSettings settings, s
 
         public async Task PublishAsync(IMessage message)
         {
-            if(message.TimeStamp == default(DateTimeOffset))
+            await PublishAsync(message, null).ConfigureAwait(false);
+        }
+
+        public async Task PublishAsync(IMessage message, int? capacity)
+        {
+            if (message.TimeStamp == default(DateTimeOffset))
             {
                 message.TimeStamp = DateTimeOffset.UtcNow;
             }
 
             var serializedEvent = JsonConvert.SerializeObject(message, JsonSerializerSettings);
             var messageType = message.GetType().Name;
 
-            await PublishAsync(serializedEvent, message.Id, messageType).ConfigureAwait(false);
+            await PublishAsync(serializedEvent, message.Id, messageType, capacity).ConfigureAwait(false);
         }
 
-        public async Task PublishAsync(string serializedMessage, string messageId, string messageType)
+        public async Task PublishAsync(string serializedMessage, string messageId, string messageType, int? capacity = null)
         {
             //Publish the event
             for (var i = 0; i < _settings.TransactionRetryCount; i++)
             {
-                var publishTran = await GeneratePublishTransaction(serializedMessage, messageType).ConfigureAwait(false);
+                var publishTran = await GeneratePublishTransaction(serializedMessage, messageType, capacity).ConfigureAwait(false);
                 if (await publishTran.ExecuteAsync().ConfigureAwait(false))
                 {
                     return;
@@ -63,7 +68,8 @@ public async Task PublishAsync(string serializedMessage, string messageId, strin
             throw new MessagePublishFailedException(messageId, _settings.TransactionRetryCount);
         }
 
-        private async Task<ITransaction> GeneratePublishTransaction(string serializedEvent, string messageType)
+
+        private async Task<ITransaction> GeneratePublishTransaction(string serializedEvent, string messageType, int? capacity)
         {
             var eventKey = $"{_environment}:{messageType}";
 
@@ -82,6 +88,13 @@ private async Task<ITransaction> GeneratePublishTransaction(string serializedEve
             foreach (var subscriber in subscribers)
             {
                 var listKey = $"{subscriber}:{{{eventKey}}}:PublishedEvents";
+
+                if (capacity.HasValue)
+                {
+                    var waiter = new AsyncConditionWaiter(async () => await _redis.Database.ListLengthAsync(listKey) < capacity);
+                    await waiter.WaitForConditionAsync().ConfigureAwait(false);
+                }
+
                 tran.ListLeftPushAsync(listKey, serializedEvent).ConfigureAwait(false);
             }
 

test/Learning.Cqrs.Test/Learning.Cqrs.Test.csproj

@@ -15,7 +15,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.4.0" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.9.4" />
     <PackageReference Include="MSTest.TestAdapter" Version="2.0.0" />
     <PackageReference Include="MSTest.TestFramework" Version="2.0.0" />
     <PackageReference Include="System.Diagnostics.TraceSource" Version="4.3.0" />

test/Learning.EventStore.Test/Learning.EventStore.Test.csproj

@@ -17,7 +17,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.4.0" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.9.4" />
     <PackageReference Include="FakeItEasy" Version="5.5.0" />
     <PackageReference Include="MSTest.TestAdapter" Version="2.0.0" />
     <PackageReference Include="MSTest.TestFramework" Version="2.0.0" />

test/Learning.EventStore.Test/RedisEventStore/SaveTest.cs

@@ -85,7 +85,7 @@ public async Task CallsMessageQueuePublish()
             A.CallTo(() => _trans.ExecuteAsync(CommandFlags.None)).Returns(Task.Run(() => true));
             await _redisEventStore.SaveAsync(_eventList);
 
-            A.CallTo(() => _messageQueue.PublishAsync(A<string>._, "12345", A<string>._ ))
+            A.CallTo(() => _messageQueue.PublishAsync(A<string>._, "12345", A<string>._, A<int?>._))
                 .MustHaveHappened();
         }
 
@@ -112,7 +112,7 @@ public async Task ThrowsExceptionIfSaveTransactionFails()
         public async Task DeletesFromEventStoreHashAndCommitListIfPublishThrowsException()
         {
             A.CallTo(() => _trans.ExecuteAsync(CommandFlags.None)).Returns(Task.Run(() => true));
-            A.CallTo(() => _messageQueue.PublishAsync(A<string>._, A<string>._, A<string>._)).Throws(new MessagePublishFailedException("12345", 10));
+            A.CallTo(() => _messageQueue.PublishAsync(A<string>._, A<string>._, A<string>._, A<int?>._)).Throws(new MessagePublishFailedException("12345", 10));
 
             try
             {

test/Learning.EventStore.Test/RedisMessageQueue/PublishTest.cs

@@ -108,5 +108,18 @@ public async Task ThrowsExeptionIfPublishTransactionFails()
                 A.CallTo(() => _trans.PublishAsync("test:TestEvent", true, CommandFlags.None)).MustHaveHappened(Repeated.Exactly.Times(10));
             }
         }
+
+        [TestMethod]
+        public async Task PublishesEventsWithCapacity()
+        {
+            A.CallTo(() => _trans.ExecuteAsync(CommandFlags.None)).Returns(Task.Run(() => true));
+            A.CallTo(() => _redis.Database.ListLengthAsync("Subscriber1:{test:TestEvent}:PublishedEvents", CommandFlags.None)).ReturnsNextFromSequence(new[] { (long)100, (long)99 });
+            
+            await _messageQueue.PublishAsync(_event, 100);
+
+
+            A.CallTo(() => _trans.PublishAsync("test:TestEvent", true, CommandFlags.None)).MustHaveHappened(Repeated.Exactly.Once);
+            A.CallTo(() => _redis.Database.ListLengthAsync("Subscriber1:{test:TestEvent}:PublishedEvents", CommandFlags.None)).MustHaveHappened(Repeated.Exactly.Twice);
+        }
     }
 }

test/Learning.EventStore.Test/SqlServerEventStore/SaveTest.cs

@@ -50,15 +50,15 @@ public async Task CallsMessageQueuePublish()
             await _sqlEventStore.SaveAsync(_eventList);
 
             A.CallTo(() => _dapper.ExecuteAsync(_writeDbConnection, A<string>._, A<EventDto>._, CommandType.StoredProcedure, _transaction)).MustHaveHappened();
-            A.CallTo(() => _messageQueue.PublishAsync(_serializedEvent, "12345", A<string>._ ))
+            A.CallTo(() => _messageQueue.PublishAsync(_serializedEvent, "12345", A<string>._, A<int?>._))
                 .MustHaveHappened();
             A.CallTo(() => _transaction.Commit()).MustHaveHappened();
         }
 
         [TestMethod]
         public async Task RollsBackTransactionIfPublishFails()
         {
-            A.CallTo(() => _messageQueue.PublishAsync(_serializedEvent, "12345", A<string>._)).Throws(new Exception("Publish Failed"));
+            A.CallTo(() => _messageQueue.PublishAsync(_serializedEvent, "12345", A<string>._, A<int?>._)).Throws(new Exception("Publish Failed"));
 
             try
             {
@@ -67,7 +67,7 @@ public async Task RollsBackTransactionIfPublishFails()
             catch (Exception)
             {
                 A.CallTo(() => _dapper.ExecuteAsync(_writeDbConnection, A<string>._, A<EventDto>._, CommandType.StoredProcedure, _transaction)).MustHaveHappened();
-                A.CallTo(() => _messageQueue.PublishAsync(_serializedEvent, "12345", A<string>._))
+                A.CallTo(() => _messageQueue.PublishAsync(_serializedEvent, "12345", A<string>._, A<int?>._))
                     .MustHaveHappened();
                 A.CallTo(() => _transaction.Commit()).MustNotHaveHappened();
                 A.CallTo(() => _transaction.Rollback()).MustHaveHappened();