init

src/Infrastructure/BotSharp.Core.Crontab/CrontabPlugin.cs

@@ -36,7 +36,7 @@ public class CrontabPlugin : IBotSharpPlugin
         services.AddScoped<ICrontabService, CrontabService>();
         services.AddScoped<ITaskFeeder, CrontabService>();
 
-        services.AddHostedService<CrontabWatcher>();
-        services.AddHostedService<CrontabEventSubscription>();
+        //services.AddHostedService<CrontabWatcher>();
+        //services.AddHostedService<CrontabEventSubscription>();
     }
 }

src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs

@@ -654,4 +654,4 @@ public class ConversationController : ControllerBase
         return jsonOption;
     }
     #endregion
-}
+}

src/Plugins/BotSharp.Plugin.ChatHub/ChatHubMiddleware.cs

@@ -3,11 +3,11 @@ using System.Text.RegularExpressions;
 
 namespace BotSharp.Plugin.ChatHub;
 
-public class WebSocketsMiddleware
+public class ChatHubMiddleware
 {
     private readonly RequestDelegate _next;
 
-    public WebSocketsMiddleware(RequestDelegate next)
+    public ChatHubMiddleware(RequestDelegate next)
     {
         _next = next;
     }

src/Plugins/BotSharp.Plugin.ChatHub/ChatStreamMiddleware.cs

@@ -0,0 +1,155 @@
+using Azure;
+using BotSharp.Abstraction.Realtime;
+using BotSharp.Abstraction.Realtime.Models;
+using Microsoft.AspNetCore.Http;
+using System.Net.WebSockets;
+
+namespace BotSharp.Plugin.ChatHub;
+
+public class ChatStreamMiddleware
+{
+    private readonly RequestDelegate _next;
+    private readonly ILogger<ChatStreamMiddleware> _logger;
+
+    public ChatStreamMiddleware(
+        RequestDelegate next,
+        ILogger<ChatStreamMiddleware> logger)
+    {
+        _next = next;
+        _logger = logger;
+    }
+
+    public async Task Invoke(HttpContext httpContext)
+    {
+        var request = httpContext.Request;
+
+        if (request.Path.StartsWithSegments("/chat/stream"))
+        {
+            if (httpContext.WebSockets.IsWebSocketRequest)
+            {
+                try
+                {
+                    var services = httpContext.RequestServices;
+                    var segments = request.Path.Value.Split("/");
+                    var agentId = segments[segments.Length - 2];
+                    var conversationId = segments[segments.Length - 1];
+
+                    using var webSocket = await httpContext.WebSockets.AcceptWebSocketAsync();
+                    await HandleWebSocket(services, agentId, conversationId, webSocket);
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex, $"Error when connecting Chat stream. ({ex.Message})");
+                }
+                return;
+            }
+        }
+
+        await _next(httpContext);
+    }
+
+    private async Task HandleWebSocket(IServiceProvider services, string agentId, string conversationId, WebSocket webSocket)
+    {
+        var hub = services.GetRequiredService<IRealtimeHub>();
+        var conn = hub.SetHubConnection(conversationId);
+
+        // load conversation and state
+        var convService = services.GetRequiredService<IConversationService>();
+        convService.SetConversationId(conversationId, []);
+        await convService.GetConversationRecordOrCreateNew(agentId);
+
+        var buffer = new byte[1024 * 1024 * 8];
+        WebSocketReceiveResult result;
+
+        do
+        {
+            result = await webSocket.ReceiveAsync(new(buffer), CancellationToken.None);
+            if (result.MessageType != WebSocketMessageType.Text)
+            {
+                continue;
+            }
+
+            var receivedText = Encoding.UTF8.GetString(buffer, 0, result.Count);
+            if (string.IsNullOrEmpty(receivedText))
+            {
+                continue;
+            }
+
+            var (eventType, data) = MapEvents(conn, receivedText);
+
+            if (eventType == "start")
+            {
+                await ConnectToModel(hub, webSocket);
+            }
+            else if (eventType == "media")
+            {
+                if (!string.IsNullOrEmpty(data))
+                {
+                    await hub.Completer.AppenAudioBuffer(data);
+                }
+            }
+            else if (eventType == "disconnect")
+            {
+                await hub.Completer.Disconnect();
+            }
+        }
+        while (!webSocket.CloseStatus.HasValue);
+
+        await webSocket.CloseAsync(result.CloseStatus.Value, result.CloseStatusDescription, CancellationToken.None);
+    }
+
+    private async Task ConnectToModel(IRealtimeHub hub, WebSocket webSocket)
+    {
+        await hub.ConnectToModel(async data =>
+        {
+            await SendEventToUser(webSocket, data);
+        });
+    }
+
+    private async Task SendEventToUser(WebSocket webSocket, string message)
+    {
+        var buffer = Encoding.UTF8.GetBytes(message);
+        await webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);
+    }
+
+    private (string, string) MapEvents(RealtimeHubConnection conn, string receivedText)
+    {
+        var response = JsonSerializer.Deserialize<ChatStreamEventResponse>(receivedText);
+        string data = string.Empty;
+
+        switch (response.Event)
+        {
+            case "start":
+                conn.ResetStreamState();
+                break;
+            case "media":
+                var mediaResponse = JsonSerializer.Deserialize<ChatStreamMediaEventResponse>(receivedText);
+                data = mediaResponse?.Payload ?? string.Empty;
+                break;
+            case "disconnect":
+                break;
+        }
+
+        conn.OnModelMessageReceived = message =>
+            JsonSerializer.Serialize(new
+            {
+                @event = "media",
+                media = new { payload = message }
+            });
+
+        conn.OnModelAudioResponseDone = () =>
+            JsonSerializer.Serialize(new
+            {
+                @event = "mark",
+                mark = new { name = "responsePart" }
+            });
+
+        conn.OnModelUserInterrupted = () =>
+            JsonSerializer.Serialize(new
+            {
+                @event = "clear"
+            });
+
+        return (response.Event, data);
+    }
+}

src/Plugins/BotSharp.Plugin.ChatHub/Models/Stream/ChatStreamEventResponse.cs

@@ -0,0 +1,15 @@
+using System.Text.Json.Serialization;
+
+namespace BotSharp.Plugin.ChatHub.Models.Stream;
+
+internal class ChatStreamEventResponse
+{
+    [JsonPropertyName("event")]
+    public string Event { get; set; }
+}
+
+internal class ChatStreamMediaEventResponse : ChatStreamEventResponse
+{
+    [JsonPropertyName("payload")]
+    public string Payload { get; set; }
+}

src/Plugins/BotSharp.Plugin.ChatHub/Using.cs

@@ -33,4 +33,5 @@ global using BotSharp.Abstraction.Messaging.Enums;
 global using BotSharp.Abstraction.Messaging.Models.RichContent;
 global using BotSharp.Abstraction.Templating;
 global using BotSharp.Plugin.ChatHub.Settings;
-global using BotSharp.Plugin.ChatHub.Enums;
+global using BotSharp.Plugin.ChatHub.Enums;
+global using BotSharp.Plugin.ChatHub.Models.Stream;

src/WebStarter/Program.cs

@@ -44,7 +44,7 @@ var app = builder.Build();
 
 // Enable SignalR
 app.MapHub<SignalRHub>("/chatHub");
-app.UseMiddleware<WebSocketsMiddleware>();
+app.UseMiddleware<ChatHubMiddleware>();
 
 // Use BotSharp
 app.UseBotSharp()