From 11ceb8e013f64c96cf55da127434b60424f2d028 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Fri, 22 Nov 2024 22:38:09 -0600 Subject: [PATCH] Add EventPriority --- .../Infrastructures/Enums/EventPriority.cs | 8 +++ .../Infrastructures/Events/IEventPublisher.cs | 2 +- .../Infrastructures/Events/RedisPublisher.cs | 10 ++- .../Infrastructures/Events/RedisSubscriber.cs | 72 ++++++++++++------- 4 files changed, 63 insertions(+), 29 deletions(-) create mode 100644 src/Infrastructure/BotSharp.Abstraction/Infrastructures/Enums/EventPriority.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Enums/EventPriority.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Enums/EventPriority.cs new file mode 100644 index 00000000..75c4402d --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Enums/EventPriority.cs @@ -0,0 +1,8 @@ +namespace BotSharp.Abstraction.Infrastructures.Enums; + +public enum EventPriority +{ + Low = 1, + Medium = 2, + High = 3 +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs index 68ceebd2..a58adf59 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs @@ -10,7 +10,7 @@ public interface IEventPublisher /// Task BroadcastAsync(string channel, string message); - Task PublishAsync(string channel, string message); + Task PublishAsync(string channel, string message, EventPriority priority = EventPriority.Medium); Task ReDispatchAsync(string channel, int count = 10, string order = "asc"); diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs index 2ff585d6..d991a45c 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs @@ -1,3 +1,4 @@ +using BotSharp.Abstraction.Infrastructures.Enums; using StackExchange.Redis; namespace BotSharp.Core.Infrastructures.Events; @@ -20,10 +21,13 @@ public class RedisPublisher : IEventPublisher await _subscriber.PublishAsync(channel, message); } - public async Task PublishAsync(string channel, string message) + public async Task PublishAsync(string channel, string message, EventPriority priority = EventPriority.Medium) { var db = _redis.GetDatabase(); + // convert to apporiate channel by priority + channel = $"{channel}-{priority}"; + if (CheckMessageExists(db, channel, "message", message)) { _logger.LogError($"The message already exists {channel} {message}"); @@ -41,7 +45,7 @@ public class RedisPublisher : IEventPublisher _logger.LogInformation($"Published message {channel} {message} ({messageId})"); } - private bool CheckMessageExists(IDatabase db, string streamName, string fieldName, string desiredValue) + private bool CheckMessageExists(IDatabase db, string channel, string fieldName, string desiredValue) { // Define the range to fetch all messages RedisValue start = "-"; // Start from the smallest ID @@ -49,7 +53,7 @@ public class RedisPublisher : IEventPublisher int count = 10; // Number of messages to retrieve // Fetch the latest 10 messages - var streamEntries = db.StreamRange(streamName, start, end, count, Order.Descending); + var streamEntries = db.StreamRange(channel, start, end, count, Order.Descending); if (streamEntries.Length == 0) { diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs index bf6652b1..f35113fe 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs @@ -1,3 +1,4 @@ +using BotSharp.Abstraction.Infrastructures.Enums; using StackExchange.Redis; namespace BotSharp.Core.Infrastructures.Events; @@ -28,6 +29,52 @@ public class RedisSubscriber : IEventSubscriber { var db = _redis.GetDatabase(); + await CreateConsumerGroup(db, $"{channel}-{EventPriority.Low}", group); + await CreateConsumerGroup(db, $"{channel}-{EventPriority.Medium}", group); + await CreateConsumerGroup(db, $"{channel}-{EventPriority.High}", group); + + while (true) + { + if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0) + { + continue; + } + + if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0) + { + continue; + } + + await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received); + } + } + + private async Task HandleGroupMessage(IDatabase db, string channel, string group, Func received) + { + var entries = await db.StreamReadGroupAsync(channel, group, Environment.MachineName, count: 1); + foreach (var entry in entries) + { + _logger.LogInformation($"Consumer {Environment.MachineName} received: {channel} {entry.Values[0].Value}"); + await db.StreamAcknowledgeAsync(channel, group, entry.Id); + + try + { + await received(channel, entry.Values[0].Value); + + // Optionally delete the message to save space + await db.StreamDeleteAsync(channel, [entry.Id]); + } + catch (Exception ex) + { + _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}\r\n{ex}"); + } + } + + return entries.Length; + } + + private async Task CreateConsumerGroup(IDatabase db, string channel, string group) + { // Create the consumer group if it doesn't exist try { @@ -43,30 +90,5 @@ public class RedisSubscriber : IEventSubscriber _logger.LogError($"Error creating consumer group: '{group}' {ex.Message}"); throw; } - - while (true) - { - var entries = await db.StreamReadGroupAsync(channel, group, Environment.MachineName, count: 1); - foreach (var entry in entries) - { - _logger.LogInformation($"Consumer {Environment.MachineName} received: {channel} {entry.Values[0].Value}"); - await db.StreamAcknowledgeAsync(channel, group, entry.Id); - - try - { - await received(channel, entry.Values[0].Value); - - // Optionally delete the message to save space - await db.StreamDeleteAsync(channel, [entry.Id]); - } - catch (Exception ex) - { - _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}\r\n{ex}"); - } - } - - await Task.Delay(Random.Shared.Next(1, 11) * 100); - } - } }