From 039054ad9ea5b18bafd7c49a67ae91e23f748803 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Fri, 29 Nov 2024 17:12:44 -0600 Subject: [PATCH] allow CancellationToken --- .../Infrastructures/Events/IEventSubscriber.cs | 6 +++++- .../Infrastructures/Events/RedisSubscriber.cs | 10 +++++++++- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs index fa96bb35..e04c2488 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs @@ -1,8 +1,12 @@ +using System.Threading; + namespace BotSharp.Abstraction.Infrastructures.Events; public interface IEventSubscriber { Task SubscribeAsync(string channel, Func received); - Task SubscribeAsync(string channel, string group, bool priorityEnabled, Func received); + Task SubscribeAsync(string channel, string group, bool priorityEnabled, + Func received, + CancellationToken? stoppingToken = null); } diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs index 89b38a8d..34ff0079 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs @@ -25,7 +25,9 @@ public class RedisSubscriber : IEventSubscriber }); } - public async Task SubscribeAsync(string channel, string group, bool priorityEnabled, Func received) + public async Task SubscribeAsync(string channel, string group, bool priorityEnabled, + Func received, + CancellationToken? stoppingToken = null) { var db = _redis.GetDatabase(); @@ -44,6 +46,12 @@ public class RedisSubscriber : IEventSubscriber { await Task.Delay(100); + if (stoppingToken.HasValue && stoppingToken.Value.IsCancellationRequested) + { + _logger.LogInformation($"Stopping consumer channel & group: [{channel}, {group}]"); + break; + } + if (priorityEnabled) { if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0)