108 lines
4.2 KiB
C#
108 lines
4.2 KiB
C#
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<DataStreamSubscriptionController> logger) : ControllerBase {
|
|
readonly ILogger<DataStreamSubscriptionController> 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<byte>(buffer), CancellationToken.None);
|
|
|
|
while (!receiveResult.CloseStatus.HasValue) {
|
|
await webSocket.SendAsync(
|
|
new ArraySegment<byte>(buffer, 0, receiveResult.Count),
|
|
receiveResult.MessageType,
|
|
receiveResult.EndOfMessage,
|
|
CancellationToken.None);
|
|
|
|
receiveResult = await webSocket.ReceiveAsync(
|
|
new ArraySegment<byte>(buffer), CancellationToken.None);
|
|
}
|
|
|
|
await webSocket.CloseAsync(
|
|
receiveResult.CloseStatus.Value,
|
|
receiveResult.CloseStatusDescription,
|
|
CancellationToken.None);
|
|
}
|
|
|
|
async Task<int> 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<byte>(buffer), CancellationToken.None);
|
|
|
|
try {
|
|
while (!receiveResult.CloseStatus.HasValue) {
|
|
receiveResult = await webSocket.ReceiveAsync(
|
|
new ArraySegment<byte>(buffer), CancellationToken.None);
|
|
var msg = Encoding.UTF8.GetString(buffer, 0, receiveResult.Count);
|
|
var rpcReq = JsonSerializer.Deserialize<JsonRpcRequest<DataStreamSubscription>>(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<string, StatementExpression>(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<byte>(encoded, 0, encoded.Length);
|
|
await webSocket.SendAsync(
|
|
payload,
|
|
receiveResult.MessageType,
|
|
receiveResult.EndOfMessage,
|
|
CancellationToken.None);
|
|
}
|
|
}
|
|
}
|