feat: add polling to get the number of active queues

Author: martinhonovais

samples/KafkaFlow.Retry.Sample/Helpers/KafkaClusterConfigurationBuilderHelper.cs

@@ -174,7 +174,18 @@ internal static IClusterConfigurationBuilder SetupRetryDurableSqlServer(
                                             .WithCleanupPollingConfiguration(
                                                 configure => configure
                                                     .Enabled(false)
+                                                    .WithCronExpression("0 0/1 * 1/1 * ? *")
+                                            )
+                                            .WithRetryDurableActiveQueuesCountPollingConfiguration(
+                                                configure => configure
+                                                    .Enabled(true)
+                                                    .WithCronExpression("0 0/1 * 1/1 * ? *")
+                                                    .Do((numberOfActiveQueues) => 
+                                                    {
+                                                        Console.Write($"Number of active queues {numberOfActiveQueues}");
+                                                    })
                                             )
+
                                     ))
                             .AddTypedHandlers(
                                 handlers => handlers

samples/KafkaFlow.Retry.Sample/Program.cs

@@ -21,8 +21,16 @@ private static async Task Main()
         var mongoDbDatabaseName = "kafka_flow_retry_durable_sample";
         var mongoDbRetryQueueCollectionName = "RetryQueues";
         var mongoDbRetryQueueItemCollectionName = "RetryQueueItems";
-        var sqlServerConnectionString =
-            "Server=localhost;Trusted_Connection=True; Pooling=true; Min Pool Size=1; Max Pool Size=100; MultipleActiveResultSets=true; Application Name=KafkaFlow Retry Sample";
+        var sqlServerConnectionString = string.Join(
+            string.Empty,
+            "Server=localhost;",
+            "Trusted_Connection=True;",
+            "Pooling=true;",
+            "Min Pool Size=1;",
+            "Max Pool Size=100;",
+            "MultipleActiveResultSets=true;",
+            "Application Name=KafkaFlow Retry Tests;"
+        ); 
         var sqlServerDatabaseName = "kafka_flow_retry_durable_sample";
         var topics = new[]
         {

src/KafkaFlow.Retry.MongoDb/RetryQueueDataProvider.cs

@@ -105,6 +105,11 @@ public async Task<QueuePendingItemsResult> CheckQueuePendingItemsAsync(QueuePend
         return new QueuePendingItemsResult(QueuePendingItemsResultStatus.NoPendingItems);
     }
 
+    public Task<long> CountQueuesAsync(CountQueuesInput input)
+    {
+        throw new NotImplementedException();
+    }
+
     public async Task<DeleteQueuesResult> DeleteQueuesAsync(DeleteQueuesInput input)
     {
         Guard.Argument(input, nameof(input)).NotNull();

src/KafkaFlow.Retry.Postgres/RetryQueueDataProvider.cs

@@ -107,6 +107,11 @@ public async Task<QueuePendingItemsResult> CheckQueuePendingItemsAsync(QueuePend
         }
     }
 
+    public Task<long> CountQueuesAsync(CountQueuesInput input)
+    {
+        throw new NotImplementedException();
+    }
+
     public async Task<DeleteQueuesResult> DeleteQueuesAsync(DeleteQueuesInput input)
     {
         Guard.Argument(input, nameof(input)).NotNull();

src/KafkaFlow.Retry.SqlServer/Repositories/IRetryQueueRepository.cs

@@ -17,6 +17,8 @@ internal interface IRetryQueueRepository
 
     Task<RetryQueueDbo> GetQueueAsync(IDbConnection dbConnection, string queueGroupKey);
 
+    Task<long> CountQueueAsync(IDbConnection dbConnection, string searchGroupKey, RetryQueueStatus retryQueueStatus);
+
     Task<IList<RetryQueueDbo>> GetTopSortedQueuesOrderedAsync(IDbConnection dbConnection, RetryQueueStatus retryQueueStatus, GetQueuesSortOption sortOption, string searchGroupKey, int top);
 
     Task<int> UpdateAsync(IDbConnection dbConnection, Guid idDomain, RetryQueueStatus retryQueueStatus, DateTime lastExecution);

src/KafkaFlow.Retry.SqlServer/Repositories/RetryQueueRepository.cs

@@ -34,6 +34,24 @@ public async Task<long> AddAsync(IDbConnection dbConnection, RetryQueueDbo retry
         }
     }
 
+    public async Task<long> CountQueueAsync(IDbConnection dbConnection, string searchGroupKey, RetryQueueStatus retryQueueStatus)
+    {
+        using (var command = dbConnection.CreateCommand())
+        {
+            command.CommandType = CommandType.Text;
+            command.CommandText =
+                $@"SELECT COUNT(1)
+                     FROM [{dbConnection.Schema}].[RetryQueues]
+                    WHERE SearchGroupKey = @SearchGroupKey
+                      AND IdStatus = @IdStatus";
+
+            command.Parameters.AddWithValue("SearchGroupKey", searchGroupKey);
+            command.Parameters.AddWithValue("IdStatus", (byte)retryQueueStatus);
+
+            return (Int32) await command.ExecuteScalarAsync().ConfigureAwait(false);
+        }
+    }
+
     public async Task<int> DeleteQueuesAsync(IDbConnection dbConnection, string searchGroupKey,
         RetryQueueStatus retryQueueStatus, DateTime maxLastExecutionDateToBeKept, int maxRowsToDelete)
     {
@@ -83,9 +101,9 @@ public async Task<RetryQueueDbo> GetQueueAsync(IDbConnection dbConnection, strin
             command.CommandType = CommandType.Text;
             command.CommandText =
                 $@"SELECT Id, IdDomain, IdStatus, SearchGroupKey, QueueGroupKey, CreationDate, LastExecution
-                                        FROM [{dbConnection.Schema}].[RetryQueues]
-                                        WHERE QueueGroupKey = @QueueGroupKey
-                                        ORDER BY Id";
+                    FROM [{dbConnection.Schema}].[RetryQueues]
+                    WHERE QueueGroupKey = @QueueGroupKey
+                    ORDER BY Id";
 
             command.Parameters.AddWithValue("QueueGroupKey", queueGroupKey);
 

src/KafkaFlow.Retry.SqlServer/RetryQueueDataProvider.cs

@@ -107,6 +107,20 @@ public async Task<QueuePendingItemsResult> CheckQueuePendingItemsAsync(QueuePend
         }
     }
 
+    public async Task<long> CountQueuesAsync(CountQueuesInput input)
+    {
+        Guard.Argument(input, nameof(input)).NotNull();
+
+        using (var dbConnection = _connectionProvider.Create(_sqlServerDbSettings))
+        {
+            return await _retryQueueRepository.CountQueueAsync(
+                    dbConnection,
+                    input.SearchGroupKey,
+                    input.Status)
+                .ConfigureAwait(false);
+        }
+    }
+
     public async Task<DeleteQueuesResult> DeleteQueuesAsync(DeleteQueuesInput input)
     {
         Guard.Argument(input, nameof(input)).NotNull();

src/KafkaFlow.Retry/Durable/Definitions/Builders/Polling/PollingDefinitionsAggregatorBuilder.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using Dawn;
+using KafkaFlow.Retry.Durable.Definitions.Builders.Polling;
 using KafkaFlow.Retry.Durable.Definitions.Polling;
 
 namespace KafkaFlow.Retry;
@@ -11,12 +12,14 @@ public class PollingDefinitionsAggregatorBuilder
     private readonly CleanupPollingDefinitionBuilder _cleanupPollingDefinitionBuilder;
     private readonly List<PollingDefinition> _pollingDefinitions;
     private readonly RetryDurablePollingDefinitionBuilder _retryDurablePollingDefinitionBuilder;
+    private readonly RetryDurableActiveQueuesCountPollingDefinitionBuilder _retryDurableActiveQueuesCountPollingDefinitionBuilder;
     private string _schedulerId;
 
     public PollingDefinitionsAggregatorBuilder()
     {
         _cleanupPollingDefinitionBuilder = new CleanupPollingDefinitionBuilder();
         _retryDurablePollingDefinitionBuilder = new RetryDurablePollingDefinitionBuilder();
+        _retryDurableActiveQueuesCountPollingDefinitionBuilder = new RetryDurableActiveQueuesCountPollingDefinitionBuilder();
 
         _pollingDefinitions = new List<PollingDefinition>();
     }
@@ -47,6 +50,19 @@ public PollingDefinitionsAggregatorBuilder WithRetryDurablePollingConfiguration(
         return this;
     }
 
+    public PollingDefinitionsAggregatorBuilder WithRetryDurableActiveQueuesCountPollingConfiguration(
+        Action<RetryDurableActiveQueuesCountPollingDefinitionBuilder> configure)
+    {
+        Guard.Argument(configure, nameof(configure)).NotNull();
+
+        configure(_retryDurableActiveQueuesCountPollingDefinitionBuilder);
+        var etryDurableActiveQueuesCountPollingDefinition = _retryDurableActiveQueuesCountPollingDefinitionBuilder.Build();
+
+        _pollingDefinitions.Add(etryDurableActiveQueuesCountPollingDefinition);
+
+        return this;
+    }
+
     public PollingDefinitionsAggregatorBuilder WithSchedulerId(string schedulerId)
     {
         _schedulerId = schedulerId;
@@ -65,6 +81,11 @@ internal PollingDefinitionsAggregator Build()
             ValidateRequiredPollingDefinition(PollingJobType.Cleanup);
         }
 
+        if (_retryDurableActiveQueuesCountPollingDefinitionBuilder.Required)
+        {
+            ValidateRequiredPollingDefinition(PollingJobType.RetryDurableActiveQueuesCount);
+        }
+
         return new PollingDefinitionsAggregator(_schedulerId, _pollingDefinitions);
     }
 

src/KafkaFlow.Retry/Durable/Definitions/Builders/Polling/RetryDurableActiveQueuesCountPollingDefinitionBuilder.cs

@@ -0,0 +1,30 @@
+using System;
+using Dawn;
+using KafkaFlow.Retry.Durable.Definitions.Polling;
+
+namespace KafkaFlow.Retry.Durable.Definitions.Builders.Polling;
+
+public class RetryDurableActiveQueuesCountPollingDefinitionBuilder
+    : PollingDefinitionBuilder<RetryDurableActiveQueuesCountPollingDefinitionBuilder>
+{
+    protected Action<long> ActionToPerform;
+
+    internal override bool Required => false;
+
+    public RetryDurableActiveQueuesCountPollingDefinitionBuilder Do(Action<long> actionToPerform)
+    {
+        Guard.Argument(actionToPerform, nameof(actionToPerform)).NotNull();
+
+        ActionToPerform = actionToPerform;
+        return this;
+    }
+
+    internal RetryDurableActiveQueuesCountPollingDefinition Build()
+    {
+        return new RetryDurableActiveQueuesCountPollingDefinition(
+            IsEnabled,
+            CronExpression,
+            ActionToPerform
+        );
+    }
+}

src/KafkaFlow.Retry/Durable/Definitions/Polling/PollingJobType.cs

@@ -5,5 +5,6 @@ internal enum PollingJobType
     Unknown = 0,
 
     RetryDurable = 1,
-    Cleanup = 2
+    Cleanup = 2,
+    RetryDurableActiveQueuesCount = 3
 }