namespace Bts.DataStreamWebApp.Server.Controllers; using BCG; using bts.api.core; using bts.websocket_core.json_rpc; using Bts.DataStreamExpressions; using Microsoft.AspNetCore.Mvc; using Spark.Core.Disposables; using System.Net.WebSockets; using System.Text; using System.Text.Json; [Route("/ws")] public class DataStreamSubscriptionController(ILogger logger) : ControllerBase { readonly ILogger logger = logger; IDisposable? streamSubscription; [Route("/ws")] public async Task Get() { if (!HttpContext.WebSockets.IsWebSocketRequest) { logger.LogError("Expected websocket request on path '{}'", HttpContext.Request.Path); return; } Console.WriteLine("Establishing websocket connection..."); using var webSocket = await HttpContext.WebSockets.AcceptWebSocketAsync(); Console.WriteLine($"Websocket connection established."); await runWebSocketServer(null, "bts", string.Empty, webSocket); } public static async Task Echo(WebSocket webSocket) { var buffer = new byte[1024 * 4]; var receiveResult = await webSocket.ReceiveAsync( new ArraySegment(buffer), CancellationToken.None); while (!receiveResult.CloseStatus.HasValue) { await webSocket.SendAsync( new ArraySegment(buffer, 0, receiveResult.Count), receiveResult.MessageType, receiveResult.EndOfMessage, CancellationToken.None); receiveResult = await webSocket.ReceiveAsync( new ArraySegment(buffer), CancellationToken.None); } await webSocket.CloseAsync( receiveResult.CloseStatus.Value, receiveResult.CloseStatusDescription, CancellationToken.None); } async Task runWebSocketServer(string? sparkd, string user, string password, WebSocket webSocket) { var context = ServiceContext.create(sparkd); await context.login(user, password); var buffer = new byte[1024 * 4]; var receiveResult = await webSocket.ReceiveAsync( new ArraySegment(buffer), CancellationToken.None); try { while (!receiveResult.CloseStatus.HasValue) { receiveResult = await webSocket.ReceiveAsync( new ArraySegment(buffer), CancellationToken.None); var msg = Encoding.UTF8.GetString(buffer, 0, receiveResult.Count); var rpcReq = JsonSerializer.Deserialize>(msg); var query = rpcReq?.Params?.Query ?? throw new InvalidOperationException( "Failed to parse JsonRpcRequest.Params"); var exp = DataStreamExpressionsInterpreter.Evaluate(query); if (exp is null) { Console.WriteLine($"Invalid query expression: {query}."); continue; } var vs = exp.getView(new Dictionary(0)); Disposable.dispose(ref streamSubscription); vs.onChange += OnStreamChanged; streamSubscription = Disposable.Create(() => vs.onChange -= OnStreamChanged); } } catch (Exception ex) { Console.WriteLine($"Exception on WebsocketServer thread: {ex}"); return 1; } return 0; async void OnStreamChanged(object newValue) { if (newValue is not DataPacket dp) { Console.WriteLine( $"OnStreamChanged delegate expected DataPacket (actually received {newValue?.GetType()}."); return; } Console.WriteLine("Received stream update."); var added_rows_str = dp.addedRows.get_printed_rows(val => val.Length); var msg = $"Received new rows:\n{added_rows_str}"; var encoded = Encoding.UTF8.GetBytes(msg); var payload = new ArraySegment(encoded, 0, encoded.Length); await webSocket.SendAsync( payload, receiveResult.MessageType, receiveResult.EndOfMessage, CancellationToken.None); } } }