From a3157c24707e87686569e39396d827e6eabfca0c Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Tue, 17 Jun 2025 13:17:28 -0500 Subject: [PATCH] init message hub and chathub observers --- Directory.Packages.props | 1 + .../BotSharp.Abstraction.csproj | 1 + .../Observables/Models/HubObserveData.cs | 6 + .../Observables/Models/ObserveDataBase.cs | 6 + .../Conversations/ConversationPlugin.cs | 3 + .../Observables/Queues/MessageHub.cs | 45 +++++++ .../BotSharp.Plugin.ChatHub/ChatHubPlugin.cs | 13 +- .../Observers/ChatHubObserver.cs | 119 ++++++++++++++++++ 8 files changed, 193 insertions(+), 1 deletion(-) create mode 100644 src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs create mode 100644 src/Infrastructure/BotSharp.Abstraction/Observables/Models/ObserveDataBase.cs create mode 100644 src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs create mode 100644 src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index cec2ac88..38e68d3a 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -26,6 +26,7 @@ + diff --git a/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj b/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj index 97e645d0..2008c6a2 100644 --- a/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj +++ b/src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj @@ -36,6 +36,7 @@ + diff --git a/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs b/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs new file mode 100644 index 00000000..a6d10cce --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs @@ -0,0 +1,6 @@ +namespace BotSharp.Abstraction.Observables.Models; + +public class HubObserveData : ObserveDataBase +{ + public RoleDialogModel Data { get; set; } = null!; +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Observables/Models/ObserveDataBase.cs b/src/Infrastructure/BotSharp.Abstraction/Observables/Models/ObserveDataBase.cs new file mode 100644 index 00000000..17773272 --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Observables/Models/ObserveDataBase.cs @@ -0,0 +1,6 @@ +namespace BotSharp.Abstraction.Observables.Models; + +public abstract class ObserveDataBase +{ + public IServiceProvider ServiceProvider { get; set; } = null!; +} diff --git a/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs b/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs index 13ee1de6..e8e18eb8 100644 --- a/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs +++ b/src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs @@ -10,6 +10,7 @@ using BotSharp.Core.Messaging; using BotSharp.Core.Routing.Reasoning; using BotSharp.Core.Templating; using BotSharp.Core.Translation; +using BotSharp.Core.Observables.Queues; using Microsoft.Extensions.Configuration; namespace BotSharp.Core.Conversations; @@ -41,6 +42,8 @@ public class ConversationPlugin : IBotSharpPlugin return settingService.Bind("GoogleApi"); }); + services.AddSingleton(); + services.AddScoped(); services.AddScoped(); services.AddScoped(); diff --git a/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs b/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs new file mode 100644 index 00000000..ed6772b4 --- /dev/null +++ b/src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs @@ -0,0 +1,45 @@ +using BotSharp.Abstraction.Observables.Models; +using System.Reactive.Subjects; + +namespace BotSharp.Core.Observables.Queues; + +public class MessageHub +{ + private readonly ILogger _logger; + private readonly ISubject _observable = new Subject(); + public IObservable Events => _observable; + + public MessageHub(ILogger logger) + { + _logger = logger; + } + + /// + /// Push an item to the observers. + /// + /// + public void Push(HubObserveData item) + { + _logger.LogInformation($"Pushing item to observers: {item.Data.Content}"); + _observable.OnNext(item); + } + + /// + /// Send a complete notification to the observers. + /// This will stop the observers from receiving data. + /// + public void Complete() + { + _observable.OnCompleted(); + } + + /// + /// Send an error notification to the observers. + /// This will stop the observers from receiving data. + /// + /// + public void Error(Exception error) + { + _observable.OnError(error); + } +} diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs b/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs index 725655fc..b1fb144f 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs @@ -1,5 +1,8 @@ using BotSharp.Abstraction.Crontab; +using BotSharp.Core.Observables.Queues; using BotSharp.Plugin.ChatHub.Hooks; +using BotSharp.Plugin.ChatHub.Observers; +using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.Configuration; namespace BotSharp.Plugin.ChatHub; @@ -7,7 +10,7 @@ namespace BotSharp.Plugin.ChatHub; /// /// The dialogue channel connects users, AI assistants and customer service representatives. /// -public class ChatHubPlugin : IBotSharpPlugin +public class ChatHubPlugin : IBotSharpPlugin, IBotSharpAppPlugin { public string Id => "6e52d42d-1e23-406b-8599-36af36c83209"; public string Name => "Chat Hub"; @@ -28,4 +31,12 @@ public class ChatHubPlugin : IBotSharpPlugin services.AddScoped(); services.AddScoped(); } + + public void Configure(IApplicationBuilder app) + { + var services = app.ApplicationServices; + var queue = services.GetRequiredService(); + var logger = services.GetRequiredService>(); + queue.Events.Subscribe(new ChatHubObserver(logger)); + } } diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs new file mode 100644 index 00000000..5699945f --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs @@ -0,0 +1,119 @@ +using BotSharp.Abstraction.Conversations.Dtos; +using BotSharp.Abstraction.Observables.Models; +using BotSharp.Abstraction.SideCar; +using BotSharp.Abstraction.Users.Dtos; +using BotSharp.Plugin.ChatHub.Hooks; +using Microsoft.AspNetCore.SignalR; + +namespace BotSharp.Plugin.ChatHub.Observers; + +public class ChatHubObserver : IObserver +{ + private readonly ILogger _logger; + private IServiceProvider _services; + private IUserIdentity _user; + + private const string RECEIVE_CLIENT_MESSAGE = "OnMessageReceivedFromClient"; + private const string GENERATE_SENDER_ACTION = "OnSenderActionGenerated"; + + public ChatHubObserver(ILogger logger) + { + _logger = logger; + } + + public void OnCompleted() + { + _logger.LogInformation($"{nameof(ChatHubObserver)} receives complete notification."); + } + + public void OnError(Exception error) + { + _logger.LogError(error, $"{nameof(ChatHubObserver)} receives error notification: {error.Message}"); + } + + public void OnNext(HubObserveData value) + { + _services = value.ServiceProvider; + _user = _services.GetRequiredService(); + + ReceiveMessage(value.Data).ConfigureAwait(false).GetAwaiter().GetResult(); + } + + private async Task ReceiveMessage(RoleDialogModel message) + { + if (!AllowSendingMessage()) return; + + var conv = _services.GetRequiredService(); + var userService = _services.GetRequiredService(); + var sender = await userService.GetMyProfile(); + + // Update console conversation UI for CSR + var model = new ChatResponseDto() + { + ConversationId = conv.ConversationId, + MessageId = message.MessageId, + Payload = message.Payload, + Text = !string.IsNullOrEmpty(message.SecondaryContent) ? message.SecondaryContent : message.Content, + Sender = UserDto.FromUser(sender) + }; + await ReceiveClientMessage(conv.ConversationId, model); + + // Send typing-on to client + var action = new ConversationSenderActionModel + { + ConversationId = conv.ConversationId, + SenderAction = SenderActionEnum.TypingOn + }; + + await GenerateSenderAction(conv.ConversationId, action); + } + + private async Task ReceiveClientMessage(string conversationId, ChatResponseDto model) + { + try + { + var settings = _services.GetRequiredService(); + var chatHub = _services.GetRequiredService>(); + + if (settings.EventDispatchBy == EventDispatchType.Group) + { + await chatHub.Clients.Group(conversationId).SendAsync(RECEIVE_CLIENT_MESSAGE, model); + } + else + { + await chatHub.Clients.User(_user.Id).SendAsync(RECEIVE_CLIENT_MESSAGE, model); + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, $"Failed to receive assistant message in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})"); + } + } + + private bool AllowSendingMessage() + { + var sidecar = _services.GetService(); + return sidecar == null || !sidecar.IsEnabled(); + } + + private async Task GenerateSenderAction(string conversationId, ConversationSenderActionModel action) + { + try + { + var settings = _services.GetRequiredService(); + var chatHub = _services.GetRequiredService>(); + if (settings.EventDispatchBy == EventDispatchType.Group) + { + await chatHub.Clients.Group(conversationId).SendAsync(GENERATE_SENDER_ACTION, action); + } + else + { + await chatHub.Clients.User(_user.Id).SendAsync(GENERATE_SENDER_ACTION, action); + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, $"Failed to generate sender action in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})"); + } + } +}