Master

RedisMessagePipeline.Demo/Program.cs

@@ -7,16 +7,16 @@
 using RedLockNet.SERedis.Configuration;
 using StackExchange.Redis;
 
-ConnectionMultiplexer redis = ConnectionMultiplexer.Connect("localhost:6379");
+ConnectionMultiplexer redis = ConnectionMultiplexer.Connect("");
 RedLockMultiplexer lockMultiplexer = new RedLockMultiplexer(redis);
 IDatabase db = redis.GetDatabase();
 
 var loggerFactory = new LoggerFactory();
 RedLockFactory lockFactory = RedLockFactory.Create(new List<RedLockMultiplexer> { lockMultiplexer });
 RedisPipelineFactory factory = new RedisPipelineFactory(loggerFactory, lockFactory, db);
 
-var consumer = factory.CreateConsumer(new MyMessageHandler(), new RedisPipelineConsumerSettings("my-messages"));
-var admin = factory.CreateAdmin(new RedisPipelineAdminSettings("my-messages"));
+var consumer = factory.CreateConsumer(new MyMessageHandler(), new RedisPipelineConsumerSettings("my-messages") { Type = EnPipelineType.QUEUE_SCHEDULE});
+var admin = factory.CreateAdmin(new RedisPipelineAdminSettings("my-messages") { Type = EnPipelineType.QUEUE_SCHEDULE });
 
 // ----  Administrate the pipeline  -----
 
@@ -26,11 +26,11 @@
 // Push messages
 for (int i = 0; i < 10; i++)
 {
-    await admin.PushAsync($"message:{i}");
+    await admin.AddScheduleAsync($"{i}", DateTime.Now.AddSeconds(i * 10), $"message:{i}");
 }
 
 // Resume the pipeline, skipping problematic messages if necessary
-await admin.ResumeAsync(1, CancellationToken.None);
+await admin.ResumeAsync(0, CancellationToken.None);
 
 
 // ------ Start the consumer to process messages ------
@@ -42,7 +42,7 @@ class MyMessageHandler : IRedisPipelineHandler
     public async Task<bool> HandleAsync(RedisValue redisValue, CancellationToken cancellationToken)
     {                
         bool success = Random.Shared.Next(0, 100) > 50;
-        Console.WriteLine($"Processing task: {redisValue}, success: {success}");
+        Console.WriteLine($"{DateTime.Now} Processing task: {redisValue}, success: {success}");
         await Task.Delay(300, cancellationToken);                
         return success;
     }

RedisMessagePipeline/Admin/IRedisPipelineAdmin.cs

@@ -1,12 +1,14 @@
 using StackExchange.Redis;
+using System;
 using System.Threading;
 using System.Threading.Tasks;
 
 namespace RedisMessagePipeline.Admin
 {
     public interface IRedisPipelineAdmin
     {
-        Task PushAsync(RedisValue redisValue);
+        Task<long> PushQueueAsync(RedisValue redisValue);
+        Task AddScheduleAsync(RedisValue keyValue, DateTimeOffset schedule, RedisValue redisValue);
         Task StopAsync();
         Task CleanAsync(CancellationToken cancellationToken);
         Task ResumeAsync(int skip, CancellationToken cancellationToken);

RedisMessagePipeline/Admin/RedisPipelineAdminSettings.cs

@@ -1,4 +1,6 @@
-namespace RedisMessagePipeline.Admin
+using RedisMessagePipeline.Factory;
+
+namespace RedisMessagePipeline.Admin
 {
     /// <summary>
     /// Settings for RedisPipelineAdmin to manage Redis pipelines.
@@ -10,6 +12,7 @@ public RedisPipelineAdminSettings(string resource)
             Resource = resource;
         }
         public string Resource { get; set; }
+        public EnPipelineType Type { get; set; } = EnPipelineType.QUEUE;
         public RedisPipelineLockSettings LockSettings { get; set; } = new RedisPipelineLockSettings();
     }
 }

RedisMessagePipeline/Admin/RedisPipelineBaseAdmin.cs

@@ -0,0 +1,114 @@
+using Microsoft.Extensions.Logging;
+using RedLockNet;
+using StackExchange.Redis;
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace RedisMessagePipeline.Admin
+{
+    /// <summary>
+    /// Admin functionality to manage operations on the Redis pipeline, such as starting, stopping, and cleaning.
+    /// </summary>
+    public abstract class RedisPipelineBaseAdmin : IRedisPipelineAdmin
+    {
+        protected readonly ILogger<RedisPipelineBaseAdmin> logger;
+        protected readonly RedisPipelineAdminSettings settings;
+        protected readonly IDistributedLockFactory lockFactory;
+        protected readonly IDatabase database;
+
+        internal RedisPipelineBaseAdmin(
+            ILogger<RedisPipelineBaseAdmin> logger,
+            RedisPipelineAdminSettings settings,
+            IDistributedLockFactory lockFactory,
+            IDatabase database)
+        {
+            this.logger = logger;
+            this.settings = settings;
+            this.lockFactory = lockFactory;
+            this.database = database;
+        }
+
+        /// <summary>
+        /// Pushes a new message to the Redis pipeline.
+        /// </summary>
+        public virtual Task<long> PushQueueAsync(RedisValue redisValue)
+        {
+            if (this.settings.Type != Factory.EnPipelineType.QUEUE)
+            {
+                throw new Exception("This queue is configured for scheduling; use AddScheduleAsync.");
+            }
+
+            logger.LogDebug("Push a new message '{message}' to '{resource}' redis pipeline", redisValue, settings.Resource);
+
+            return Task.FromResult(-1L);
+
+        }
+
+        /// <summary>
+        /// Stops the Redis pipeline.
+        /// </summary>
+        public Task StopAsync()
+        {
+            logger.LogDebug("Redis pipeline '{resource}' has been stopped", settings.Resource);
+
+            RedisKey key = RedisPipelineExtensions.StateKey(settings.Resource);
+            return database.StringSetAsync(key, RedisPipelineExtensions.STATE_STOPPED);
+        }
+
+        /// <summary>
+        /// Cleans up resources used by the Redis pipeline.
+        /// </summary>
+        public abstract Task CleanAsync(CancellationToken cancellationToken);
+
+        /// <summary>
+        /// Resumes operations of the Redis pipeline after a stop.
+        /// </summary>
+        public abstract Task ResumeAsync(int skip, CancellationToken cancellationToken);
+
+        public virtual Task AddScheduleAsync(RedisValue keyValue, DateTimeOffset schedule, RedisValue redisValue)
+        {
+            if (this.settings.Type != Factory.EnPipelineType.QUEUE_SCHEDULE)
+            {
+                throw new Exception("The queue is not configured for scheduling; use PushQueueAsync.");
+            }
+
+            logger.LogDebug("Redis pipeline '{resource}' has been stopped", settings.Resource);
+
+            return Task.CompletedTask;
+        }
+
+        protected async Task<long> RemoveByPatternInBatchesAsync(string pattern, int batchSize = 500)
+        {
+            var endpoints = database.Multiplexer.GetEndPoints();
+            long totalRemovidas = 0;
+
+            foreach (var endpoint in endpoints)
+            {
+                var server = database.Multiplexer.GetServer(endpoint);
+
+                if (!server.IsConnected || server.IsReplica)
+                    continue;
+
+                var batch = new List<RedisKey>(batchSize);
+
+                foreach (var key in server.Keys(pattern: pattern, pageSize: batchSize))
+                {
+                    batch.Add(key);
+                    if (batch.Count >= batchSize)
+                    {
+                        totalRemovidas += await database.KeyDeleteAsync(batch.ToArray());
+                        batch.Clear();
+                    }
+                }
+                if (batch.Count > 0)
+                {
+                    totalRemovidas += await database.KeyDeleteAsync(batch.ToArray());
+                }
+            }
+
+            return totalRemovidas;
+        }
+    }
+}

RedisMessagePipeline/Admin/RedisPipelineAdmin.cs → RedisMessagePipeline/Admin/RedisPipelineQueueAdmin.cs

@@ -10,51 +10,32 @@ namespace RedisMessagePipeline.Admin
     /// <summary>
     /// Admin functionality to manage operations on the Redis pipeline, such as starting, stopping, and cleaning.
     /// </summary>
-    public class RedisPipelineAdmin : IRedisPipelineAdmin
+    public class RedisPipelineQueueAdmin : RedisPipelineBaseAdmin
     {
-        private readonly ILogger<RedisPipelineAdmin> logger;
-        private readonly RedisPipelineAdminSettings settings;
-        private readonly IDistributedLockFactory lockFactory;
-        private readonly IDatabase database;
-
-        internal RedisPipelineAdmin(
-            ILogger<RedisPipelineAdmin> logger,
+        internal RedisPipelineQueueAdmin(
+            ILogger<RedisPipelineQueueAdmin> logger,
             RedisPipelineAdminSettings settings,
             IDistributedLockFactory lockFactory,
             IDatabase database)
+            : base(logger, settings, lockFactory, database)
         {
-            this.logger = logger;
-            this.settings = settings;
-            this.lockFactory = lockFactory;
-            this.database = database;
-        }
 
-        /// <summary>
-        /// Pushes a new message to the Redis pipeline.
-        /// </summary>
-        public Task PushAsync(RedisValue redisValue)
-        {
-            logger.LogDebug("Push a new message '{message}' to '{resource}' redis pipeline", redisValue, settings.Resource);
-
-            RedisKey key = RedisPipelineExtensions.MessagesKey(settings.Resource);
-            return database.ListRightPushAsync(key, redisValue);
         }
 
         /// <summary>
-        /// Stops the Redis pipeline.
+        /// Pushes a new message to the Redis pipeline.
         /// </summary>
-        public Task StopAsync()
+        public override async Task<long> PushQueueAsync(RedisValue redisValue)
         {
-            logger.LogDebug("Redis pipeline '{resource}' has been stopped", settings.Resource);
-
-            RedisKey key = RedisPipelineExtensions.StateKey(settings.Resource);
-            return database.StringSetAsync(key, RedisPipelineExtensions.STATE_STOPPED);
+            await base.PushQueueAsync(redisValue);
+            RedisKey key = RedisPipelineExtensions.MessagesListKey(settings.Resource);
+            return await database.ListRightPushAsync(key, redisValue);
         }
 
         /// <summary>
         /// Cleans up resources used by the Redis pipeline.
         /// </summary>
-        public async Task CleanAsync(CancellationToken cancellationToken)
+        public override async Task CleanAsync(CancellationToken cancellationToken)
         {
             using (IRedLock locker = await lockFactory.CreateLockAsync(
                 resource: settings.Resource,
@@ -73,7 +54,7 @@ await database.KeyDeleteAsync(new RedisKey[]
                 {
                     RedisPipelineExtensions.FailureKey(settings.Resource),
                     RedisPipelineExtensions.StateKey(settings.Resource),
-                    RedisPipelineExtensions.MessagesKey(settings.Resource),
+                    RedisPipelineExtensions.MessagesListKey(settings.Resource),
                 });
 
                 logger.LogDebug("Redis pipeline '{resource}' has been cleaned up", settings.Resource);
@@ -83,7 +64,7 @@ await database.KeyDeleteAsync(new RedisKey[]
         /// <summary>
         /// Resumes operations of the Redis pipeline after a stop.
         /// </summary>
-        public async Task ResumeAsync(int skip, CancellationToken cancellationToken)
+        public override async Task ResumeAsync(int skip, CancellationToken cancellationToken)
         {
             using (IRedLock locker = await lockFactory.CreateLockAsync(
                 resource: settings.Resource,
@@ -106,10 +87,10 @@ public async Task ResumeAsync(int skip, CancellationToken cancellationToken)
                 }
 
                 ITransaction transaction = database.CreateTransaction();
-                RedisKey messagesKey = RedisPipelineExtensions.MessagesKey(settings.Resource);
+                RedisKey messagesKey = RedisPipelineExtensions.MessagesListKey(settings.Resource);
                 RedisKey stateKey = RedisPipelineExtensions.StateKey(settings.Resource);
                 Task[] transactionTasks = new Task[] {
-                    transaction.ListLeftPopAsync(messagesKey, count: skip),
+                    skip > 0 ? transaction.ListLeftPopAsync(messagesKey, count: skip) : Task.CompletedTask,
                     transaction.StringSetAsync(stateKey, 0)
                 };
                 await transaction.ExecuteAsync();