diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs new file mode 100644 index 00000000..471d810e --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs @@ -0,0 +1,14 @@ +namespace BotSharp.Abstraction.Infrastructures.Events; + +public interface IEventPublisher +{ + /// + /// Boardcast message to all subscribers + /// + /// + /// + /// + Task BroadcastAsync(string channel, string message); + + Task PublishAsync(string channel, string message); +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs new file mode 100644 index 00000000..f295e54e --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs @@ -0,0 +1,8 @@ +namespace BotSharp.Abstraction.Infrastructures.Events; + +public interface IEventSubscriber +{ + Task SubscribeAsync(string channel, Func received); + + Task SubscribeAsync(string channel, string group, Func received); +} diff --git a/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs b/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs index 4bdfe0d7..812d89a8 100644 --- a/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs +++ b/src/Infrastructure/BotSharp.Core/BotSharpCoreExtensions.cs @@ -9,6 +9,8 @@ using BotSharp.Abstraction.Users.Settings; using BotSharp.Abstraction.Interpreters.Settings; using BotSharp.Abstraction.Infrastructures; using BotSharp.Core.Processors; +using StackExchange.Redis; +using BotSharp.Core.Infrastructures.Events; namespace BotSharp.Core; @@ -34,10 +36,12 @@ public static class BotSharpCoreExtensions services.AddSingleton(x => cacheSettings); services.AddSingleton(); + AddRedisEvents(services, config); + services.AddMemoryCache(); RegisterPlugins(services, config); - ConfigureBotSharpOptions(services, configOptions); + AddBotSharpOptions(services, configOptions); return services; } @@ -79,7 +83,7 @@ public static class BotSharpCoreExtensions return app; } - private static void ConfigureBotSharpOptions(IServiceCollection services, Action? configure) + private static void AddBotSharpOptions(IServiceCollection services, Action? configure) { var options = new BotSharpOptions(); if (configure != null) @@ -91,6 +95,17 @@ public static class BotSharpCoreExtensions services.AddSingleton(options); } + private static void AddRedisEvents(IServiceCollection services, IConfiguration config) + { + // Add Redis connection as a singleton + var dbSettings = new BotSharpDatabaseSettings(); + config.Bind("Database", dbSettings); + + services.AddSingleton(ConnectionMultiplexer.Connect(dbSettings.Redis)); + services.AddSingleton(); + services.AddSingleton(); + } + private static void AddDefaultJsonConverters(BotSharpOptions options) { options.JsonSerializerOptions.Converters.Add(new RichContentJsonConverter()); diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs index 25c3a047..e37145ff 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/DistributedLocker.cs @@ -5,21 +5,18 @@ namespace BotSharp.Core.Infrastructures; public class DistributedLocker { - private readonly BotSharpDatabaseSettings _settings; - private static ConnectionMultiplexer connection; + private readonly IConnectionMultiplexer _redis; - public DistributedLocker(BotSharpDatabaseSettings settings) + public DistributedLocker(IConnectionMultiplexer redis) { - _settings = settings; + _redis = redis; } public async Task Lock(string resource, Func> action, int timeoutInSeconds = 30) { - await ConnectToRedisAsync(); - var timeout = TimeSpan.FromSeconds(timeoutInSeconds); - var @lock = new RedisDistributedLock(resource, connection.GetDatabase()); + var @lock = new RedisDistributedLock(resource, _redis.GetDatabase()); await using (var handle = await @lock.TryAcquireAsync(timeout)) { if (handle == null) @@ -33,11 +30,9 @@ public class DistributedLocker public void Lock(string resource, Action action, int timeoutInSeconds = 30) { - ConnectToRedis(); - var timeout = TimeSpan.FromSeconds(timeoutInSeconds); - var @lock = new RedisDistributedLock(resource, connection.GetDatabase()); + var @lock = new RedisDistributedLock(resource, _redis.GetDatabase()); using (var handle = @lock.TryAcquire(timeout)) { if (handle == null) @@ -50,20 +45,4 @@ public class DistributedLocker } } } - - private void ConnectToRedis() - { - if (connection == null) - { - connection = ConnectionMultiplexer.Connect(_settings.Redis); - } - } - - private async Task ConnectToRedisAsync() - { - if (connection == null) - { - connection = await ConnectionMultiplexer.ConnectAsync(_settings.Redis); - } - } } diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs new file mode 100644 index 00000000..4f3ad405 --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs @@ -0,0 +1,32 @@ +using StackExchange.Redis; + +namespace BotSharp.Core.Infrastructures.Events; + +public class RedisPublisher : IEventPublisher +{ + private readonly IConnectionMultiplexer _redis; + private readonly ISubscriber _subscriber; + private readonly ILogger _logger; + + public RedisPublisher(IConnectionMultiplexer redis, ILogger logger) + { + _redis = redis; + _logger = logger; + _subscriber = _redis.GetSubscriber(); + } + + public async Task BroadcastAsync(string channel, string message) + { + await _subscriber.PublishAsync(channel, message); + } + + public async Task PublishAsync(string channel, string message) + { + var db = _redis.GetDatabase(); + // Add a message to the stream, keeping only the latest 1 million messages + await db.StreamAddAsync(channel, "message", message, + maxLength: 1000 * 10000); + + _logger.LogInformation($"Published message {channel} {message}"); + } +} diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs new file mode 100644 index 00000000..39bb1f91 --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs @@ -0,0 +1,73 @@ +using StackExchange.Redis; +using System.Threading.Channels; + +namespace BotSharp.Core.Infrastructures.Events; + +public class RedisSubscriber : IEventSubscriber +{ + private readonly IConnectionMultiplexer _redis; + private readonly ISubscriber _subscriber; + private readonly ILogger _logger; + + public RedisSubscriber(IConnectionMultiplexer redis, ILogger logger) + { + _redis = redis; + _logger = logger; + _subscriber = _redis.GetSubscriber(); + } + + public async Task SubscribeAsync(string channel, Func received) + { + await _subscriber.SubscribeAsync(channel, async (ch, message) => + { + _logger.LogInformation($"Received event from channel: {ch} message: {message}"); + await received(ch, message); + }); + } + + public async Task SubscribeAsync(string channel, string group, Func received) + { + var db = _redis.GetDatabase(); + + // Create the consumer group if it doesn't exist + try + { + await db.StreamCreateConsumerGroupAsync(channel, group, StreamPosition.NewMessages, createStream: true); + } + catch (RedisServerException ex) when (ex.Message.Contains("BUSYGROUP")) + { + // Group already exists, ignore the error + _logger.LogWarning($"Consumer group '{group}' already exists (caught exception)."); + } + catch (Exception ex) + { + _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}"); + } + } + + await Task.Delay(Random.Shared.Next(1, 11) * 100); + } + + } +} diff --git a/src/Infrastructure/BotSharp.Core/Using.cs b/src/Infrastructure/BotSharp.Core/Using.cs index 8a0ca2af..9e9177ba 100644 --- a/src/Infrastructure/BotSharp.Core/Using.cs +++ b/src/Infrastructure/BotSharp.Core/Using.cs @@ -40,4 +40,5 @@ global using BotSharp.Core.Agents.Services; global using BotSharp.Core.Conversations.Services; global using BotSharp.Core.Infrastructures; global using BotSharp.Core.Users.Services; -global using Aspects.Cache; \ No newline at end of file +global using Aspects.Cache; +global using BotSharp.Abstraction.Infrastructures.Events; \ No newline at end of file diff --git a/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs b/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs index efad6e6c..4d85167f 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/BotSharpOpenApiExtensions.cs @@ -209,7 +209,7 @@ public static class BotSharpOpenApiExtensions app.UseSwagger(); - // if (env.IsDevelopment()) + if (env.IsDevelopment()) { IdentityModelEventSource.ShowPII = true; app.UseSwaggerUI();