check point (bc i don't trust googledrive sync yet)

dev: Add gitpod.yml for cloud workspaces
externals: Change repo url to use read-only access token
2024-02-09 23:06:26 -06:00 · 2024-02-09 23:06:26 -06:00 · 2024-02-09 23:06:26 -06:00 · 2024-02-09 23:06:26 -06:00 · 2024-02-09 23:06:26 -06:00 · 2024-02-09 23:06:26 -06:00
204 changed files with 12230 additions and 2333 deletions
@@ -207,17 +207,23 @@ dotnet_naming_symbols.parameters.applicable_accessibilities = public,internal,pr
 dotnet_naming_symbols.private_symbols.applicable_kinds = property,method,field,event,delegate
 dotnet_naming_symbols.private_symbols.applicable_accessibilities = protected,private

+# local symbols
+dotnet_naming_symbols.local_symbols.applicable_kinds = local
+
+# parameter symbols
+dotnet_naming_symbols.parameter_symbols.applicable_kinds = parameter
+
 # Naming styles

-dotnet_naming_style.snake_case.required_prefix =
-dotnet_naming_style.snake_case.required_suffix =
-dotnet_naming_style.snake_case.word_separator = _
-dotnet_naming_style.snake_case.capitalization = all_lower
+dotnet_naming_style.pascal_case.required_prefix =
+dotnet_naming_style.pascal_case.required_suffix =
+dotnet_naming_style.pascal_case.word_separator =
+dotnet_naming_style.pascal_case.capitalization = pascal_case

-dotnet_naming_style.private_snake_case.required_prefix = _
-dotnet_naming_style.private_snake_case.required_suffix =
-dotnet_naming_style.private_snake_case.word_separator = _
-dotnet_naming_style.private_snake_case.capitalization = all_lower
+dotnet_naming_style.camel_case.required_prefix =
+dotnet_naming_style.camel_case.required_suffix =
+dotnet_naming_style.camel_case.word_separator =
+dotnet_naming_style.camel_case.capitalization = camel_case

 dotnet_naming_style.user_types.required_prefix =
 dotnet_naming_style.user_types.required_suffix =
@@ -229,6 +235,10 @@ dotnet_naming_style.interface.required_suffix =
 dotnet_naming_style.interface.word_separator =
 dotnet_naming_style.interface.capitalization = pascal_case

+dotnet_naming_style.local.capitalization = camel_case
+
+dotnet_naming_style.parameter.capitalization = camel_case
+
 # rules
 dotnet_naming_rule.user_types.symbols = user_types
 dotnet_naming_rule.user_types.style = user_types
@@ -238,14 +248,18 @@ dotnet_naming_rule.interfaces.symbols = interface
 dotnet_naming_rule.interfaces.style = interface_style
 dotnet_naming_rule.interfaces.severity = suggestion

-dotnet_naming_rule.public_members_snake_case.symbols = public_symbols
-dotnet_naming_rule.public_members_snake_case.style = snake_case
-dotnet_naming_rule.public_members_snake_case.severity = suggestion
+dotnet_naming_rule.public_members.symbols = public_symbols
+dotnet_naming_rule.public_members.style = pascal_case
+dotnet_naming_rule.public_members.severity = suggestion

-dotnet_naming_rule.private_members_snake_case.symbols = private_symbols
-dotnet_naming_rule.private_members_snake_case.style = private_snake_case
-dotnet_naming_rule.private_members_snake_case.severity = suggestion
+dotnet_naming_rule.private_members.symbols = private_symbols
+dotnet_naming_rule.private_members.style = camel_case
+dotnet_naming_rule.private_members.severity = suggestion

-dotnet_naming_rule.parameters.symbols = parameters
-dotnet_naming_rule.parameters.style = snake_case
+dotnet_naming_rule.variables_are_camel_case.severity = suggestion
+dotnet_naming_rule.variables_are_camel_case.symbols = local_symbols
+dotnet_naming_rule.variables_are_camel_case.style = local
+
+dotnet_naming_rule.parameters.symbols = parameter_symbols
+dotnet_naming_rule.parameters.style = parameter
 dotnet_naming_rule.parameters.severity = suggestion
@@ -1,6 +1,6 @@
 [submodule "externals"]
 	path = externals
-	url = git@bitbucket.org:bluetradesys/externals.git
+	url = https://x-token-auth:ATCTT3xFfGN0h_ENnUSfEAIzbfVzkQPLG26cIli50MtcTwZ7VEc8lin5tJYqibjlBJ9PM9R642Ra5GjiL52TBfbO8KeUBT0UoMT7gLQHL2Y1TkW9yZr6NkvsoDkSl49DeD4Ijiu4_0u4qEm4oMVDDg2_fB059-uOWOo7ZL94tGZET8CSVpFIGvs=B0E03660@bitbucket.org/bluetradesys/externals.git
 [submodule "vcpkg"]
 	path = vcpkg
 	url = https://github.com/microsoft/vcpkg.git
@@ -0,0 +1,49 @@
+# This configuration file was automatically generated by Gitpod.
+# Please adjust to your needs (see https://www.gitpod.io/docs/introduction/learn-gitpod/gitpod-yaml)
+# and commit this file to your remote git repository to share the goodness with others.
+
+# Learn more from ready-to-use templates: https://www.gitpod.io/docs/introduction/getting-started/quickstart
+
+image: gitpod/workspace-dotnet-8
+tasks:
+  - name: Install Deps
+    init: |
+      (
+        set -e
+        sudo install-packages rapidjson-dev
+        sudo install-packages libssh2-1-dev
+        sudo install-packages openssh-server
+        sudo install-packages liblttng-ust-dev
+        sudo install-packages libpcap0.8-dev
+        sudo install-packages ccache
+        sudo ln -s $(which ccache) /usr/local/bin/clang
+        sudo ln -s $(which ccache) /usr/local/bin/clang++
+        sudo ln -s $(which ccache) /usr/local/bin/clang-17
+        sudo ln -s $(which ccache) /usr/local/bin/clang++-17
+        sudo ln -s $(which ccache) /usr/local/bin/gcc
+        sudo ln -s $(which ccache) /usr/local/bin/g++
+        sudo ln -s $(which ccache) /usr/local/bin/gcc-12
+        sudo ln -s $(which ccache) /usr/local/bin/g++-12
+        if [ -e externals/src/emumba/install_emumba-deribit-connector.sh ]; then
+          externals/src/emumba/install_emumba-deribit-connector.sh
+        fi
+      )
+  - name: Setup Python venv
+    init: |
+      python -m pip install --upgrade pip
+      python -m venv .venv
+      . .venv/bin/activate
+      python -m pip install --upgrade pip
+      pip install -r requirements.txt
+
+ports:
+  - name: sparkd config
+    port: 8000
+    onOpen: notify
+  - name: sparkd bus
+    port: 8001
+
+gitConfig:
+  alias.st: "status"
+  status.short: "true"
+  core.pager: "less -FXRS"
@@ -29,8 +29,8 @@
        "CMAKE_TOOLCHAIN_FILE": "${sourceDir}/vcpkg/scripts/buildsystems/vcpkg.cmake",
        "EXTERNALS_ROOT": "${sourceDir}/externals/windows",
        "CONFIG_DEFAULTS_PATH": "config/profiles/default/*.conf",
-        "DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net7.0",
-        "DOTNET_WINDOWS_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net7.0-windows"
+        "DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net8.0",
+        "DOTNET_WINDOWS_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net8.0-windows"
      }
    },

@@ -197,7 +197,7 @@
        "CMAKE_CXX_FLAGS_RELEASE": "-fPIC -Wall -pedantic -Wextra -Werror -Wno-missing-braces -Wno-ignored-qualifiers --std=c++20 -fdiagnostics-color -fno-omit-frame-pointer -DNDEBUG -O3",
        "CMAKE_INSTALL_PREFIX": "${sourceDir}/install/${presetName}",
        "CONFIG_DEFAULTS_PATH": "config/profiles/default/*.conf",
-        "DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net7.0"
+        "DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net8.0"
      }
    },

@@ -1,65 +1,65 @@
-FROM ubuntu:latest
+FROM ubuntu:22.04

 ARG DEBIAN_FRONTEND=noninteractive
-ENV TZ=America/New_York

-# light deps (should be quick to download/install)
+# Foundational Deps
 RUN apt-get update && apt-get install -y \
+    build-essential \
    ca-certificates \
    curl \
    gnupg \
    liblttng-ust-dev \
    libpcap0.8-dev \
+    libssh2-1-dev \
    locales \
-    ninja-build \
+    openssh-server \
    software-properties-common \
-    sqlite3 \
+    sudo \
    tzdata

-# clang
-RUN apt-get update && apt-get install -y \
-    clang-13 \
-    clang-format-13 \
-    clangd-13 \
-    clang-15 \
-    clang-format-15 \
-    clangd-15
+# Clang 17
+RUN curl -fsSL https://apt.llvm.org/llvm-snapshot.gpg.key 2> /dev/null \
+    | gpg --dearmor - \
+    | tee /usr/share/keyrings/llvm-snapshot.gpg > /dev/null \
+    && echo "deb [signed-by=/usr/share/keyrings/llvm-snapshot.gpg] http://apt.llvm.org/jammy/ llvm-toolchain-jammy-17 main" \
+    | tee -a "/etc/apt/sources.list.d/llvm.list" \
+    && apt-get update \
+    && apt-get install -y \
+    clang-17 \
+    clang-format-17 \
+    clangd-17

-# g++
+# G++
 RUN apt-get update && apt-get install -y \
-    g++-11 \
-    g++-12
+    g++

-# python
+# Ninja
 RUN apt-get update && apt-get install -y \
-    python3-pip \
-    python3-venv \
-    && python3 -m pip install --upgrade pip
+    ninja-build

-# dotnet
-RUN apt-get update && apt-get install -y \
-    dotnet7
-
-# ssh
-RUN apt-get update && apt-get install -y \
-    libssh2-1-dev \
-    openssh-server
-
-# cmake
-RUN \
-    curl -fsSL https://apt.kitware.com/keys/kitware-archive-latest.asc 2> /dev/null \
+# CMake
+RUN curl -fsSL https://apt.kitware.com/keys/kitware-archive-latest.asc 2> /dev/null \
    | gpg --dearmor - \
    | tee /usr/share/keyrings/kitware-archive-keyring.gpg > /dev/null \
    && echo "deb [signed-by=/usr/share/keyrings/kitware-archive-keyring.gpg] https://apt.kitware.com/ubuntu/ $(lsb_release -cs) main" \
    | tee /etc/apt/sources.list.d/kitware.list > /dev/null \
    && apt-get update \
-    && rm /usr/share/keyrings/kitware-archive-keyring.gpg \
-    && apt-get install -y kitware-archive-keyring \
-    && echo "deb [signed-by=/usr/share/keyrings/kitware-archive-keyring.gpg] https://apt.kitware.com/ubuntu/ $(lsb_release -cs)-rc main" \
-    | tee -a /etc/apt/sources.list.d/kitware.list >/dev/null \
    && apt-get install -y cmake

-# node
+# Dotnet
+RUN curl -fsSLO "https://packages.microsoft.com/config/ubuntu/$(lsb_release -rs)/packages-microsoft-prod.deb" \
+    && dpkg -i packages-microsoft-prod.deb \
+    && rm packages-microsoft-prod.deb \
+    && apt-get update \
+    && apt-get install -y dotnet-sdk-8.0
+
+# Python
+RUN apt-get update && apt-get install -y \
+    python3-pip \
+    python3-venv \
+    && python3 -m pip install --upgrade pip
+
+# Node
 ARG node_major=18
 RUN mkdir -p /etc/apt/keyrings \
    && curl -fsSL https://deb.nodesource.com/gpgkey/nodesource-repo.gpg.key \
@@ -69,54 +69,35 @@ RUN mkdir -p /etc/apt/keyrings \
    && apt-get update \
    && apt-get install -y nodejs

-# ccache is done here so that ccache sees clang++ installed, not clang-10
+# Sqlite3
 RUN apt-get update && apt-get install -y \
-    ccache 
+    sqlite3

-# emumba (light deps)
-RUN apt-get update && apt-get install -y \
-    build-essential \
-    doxygen \
-    python3-sphinx \
-    graphviz \
-    libgtest-dev \
-    libssl-dev \
-    rapidjson-dev \
-    zlib1g-dev \
-    pkg-config
-
-# emumba (boost)
-RUN apt-get update && apt-get install -y \
-    libboost-filesystem-dev \
-    libboost-program-options-dev \
-    libboost-test-dev \
-    libboost-date-time-dev
-
-# dev utils
+# Dev Utils (install last so they can be added/removed with minimal rebuild)
 RUN apt-get update && apt-get install -y \ 
+    ccache \
    fish \
+    gdb \
    git-all \
-    sudo \
-    vim \
-    wget
+    vim

-# set up default executables (update-alternatives)
-RUN \
-    update-alternatives --install /usr/bin/clang++ clang++ /usr/bin/clang++-13 100; \
-    update-alternatives --install /usr/bin/clang clang /usr/bin/clang-13 100; \
-    update-alternatives --install /usr/bin/clang-format clang-format /usr/bin/clang-format-13 100; \
-    update-alternatives --install /usr/bin/clangd clangd /usr/bin/clangd-13 100; \
-    update-alternatives --install /usr/bin/g++ g++ /usr/bin/g++-11 100; \
-    update-alternatives --install /usr/bin/gcc gcc /usr/bin/gcc-11 100
+# Set up default executables (update-alternatives)
+RUN update-alternatives --install /usr/bin/clang++ clang++ /usr/bin/clang++-17 100; \
+    update-alternatives --install /usr/bin/clang clang /usr/bin/clang-17 100; \
+    update-alternatives --install /usr/bin/clang-format clang-format /usr/bin/clang-format-17 100; \
+    update-alternatives --install /usr/bin/clangd clangd /usr/bin/clangd-17 100; \
+    update-alternatives --install /usr/bin/g++ g++ /usr/bin/g++-12 100; \
+    update-alternatives --install /usr/bin/gcc gcc /usr/bin/gcc-12 100

 # Add compiler symlinks for ccache 
-RUN \
-    ln -s /usr/bin/ccache /usr/local/bin/clang; \
+RUN ln -s /usr/bin/ccache /usr/local/bin/clang; \
    ln -s /usr/bin/ccache /usr/local/bin/clang++; \
    ln -s /usr/bin/ccache /usr/local/bin/clang-13; \
    ln -s /usr/bin/ccache /usr/local/bin/clang++-13; \
    ln -s /usr/bin/ccache /usr/local/bin/clang-15; \
    ln -s /usr/bin/ccache /usr/local/bin/clang++-15; \
+    ln -s /usr/bin/ccache /usr/local/bin/clang-17; \
+    ln -s /usr/bin/ccache /usr/local/bin/clang++-17; \
    ln -s /usr/bin/ccache /usr/local/bin/gcc; \
    ln -s /usr/bin/ccache /usr/local/bin/g++; \
    ln -s /usr/bin/ccache /usr/local/bin/gcc-11; \
@@ -130,8 +111,7 @@ ARG docker_user="bts"
 ARG docker_uid="1000"
 ARG docker_pass="bts"

-RUN \
-    groupadd -g "$docker_uid" "$docker_user" \
+RUN groupadd -g "$docker_uid" "$docker_user" \
    && useradd \
    -m \
    -u "$docker_uid" \
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <TargetFramework>net7.0</TargetFramework>
+    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
  </PropertyGroup>
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <TargetFramework>net7.0</TargetFramework>
+    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
  </PropertyGroup>
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <TargetFramework>net7.0-windows</TargetFramework>
+    <TargetFramework>net8.0-windows</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
    <UseWindowsForms>true</UseWindowsForms>
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <TargetFramework>net7.0</TargetFramework>
+    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
  </PropertyGroup>
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <TargetFramework>net7.0</TargetFramework>
+    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
  </PropertyGroup>
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <TargetFramework>net7.0</TargetFramework>
+    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
  </PropertyGroup>
@@ -0,0 +1,15 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\client\spark_websocket\spark_websocket.csproj" />
+    <ProjectReference Include="..\DataStreamExpressions\DataStreamExpressions.csproj" />
+    <ProjectReference Include="..\..\client\spark_core\spark_core.csproj" />
+  </ItemGroup>
+
+</Project>
@@ -0,0 +1,9 @@
+namespace Bts.DataStream.Websocket;
+
+class ClientSubscription(WebsocketClientID id, SubscriptionParams @params, IDisposable dispose) : IDisposable {
+   public SubscriptionParams Params => @params;
+   public WebsocketClientID Id => id;
+
+   public void Dispose() =>
+      dispose.Dispose();
+}
@@ -0,0 +1,301 @@
+namespace Bts.DataStream.Websocket;
+
+using BCG;
+
+using bts;
+using bts.api.core;
+using bts.collections;
+
+using Bts.DataStreamExpressions;
+using Bts.Websocket;
+
+using mbts.utils;
+
+using Spark.Core.Disposables;
+using Spark.Core.Json;
+
+using System.Buffers;
+using System.Reactive.Subjects;
+using System.Text;
+using System.Text.Json;
+using System.Text.Json.Nodes;
+
+/// <summary>
+/// Represents an I/O model (<see cref="ISubject{TSource, TResult}"/>), that accepts websocket
+/// requests (from clients) as input, and emits data/notifications (to clients) as output.
+/// This class's chief role is managing data subscriptions to sparkd on the clients' behalf.
+/// </summary>
+public class DataSync(JsonSerializerOptions? serializerOpts = null) :
+   ISubject<WebsocketConnectionUpdate, IEnumerable<WebsocketOutgoingMessage>> {
+   protected readonly object lockObj = new();
+   readonly JsonSerializerOptions serializerOpts = serializerOpts ?? JsonSerializerOptions.Default;
+   readonly Dictionary<WebsocketClientID, List<ClientSubscription>> clients = [];
+
+   event Action<IEnumerable<WebsocketOutgoingMessage>>? requestSend;
+
+   public void OnCompleted() =>
+      Log.Api.info("DataSync incoming request stream has completed.");
+
+   public void OnError(Exception error) =>
+      Log.Api.error($"DataSync error: {error}");
+
+   public void OnNext(WebsocketConnectionUpdate value) {
+      switch (value.Type) {
+         case WebsocketConnectionUpdateType.IncomingMessage:
+            onIncomingMessage(value.IncomingMessage);
+            break;
+         case WebsocketConnectionUpdateType.Connect:
+            onConnect(value.Connect);
+            break;
+         case WebsocketConnectionUpdateType.Disconnect:
+            onDisconnect(value.Disconnect);
+            break;
+         default:
+            Log.assert(false, $"Unexpected WebsocketConnectionUpdate type: {value.Type}");
+            break;
+      }
+   }
+
+   public IDisposable Subscribe(IObserver<IEnumerable<WebsocketOutgoingMessage>> observer) {
+      lock (lockObj) {
+         requestSend += observer.OnNext;
+         return Disposable.Create(Dispose);
+      }
+      void Dispose() {
+         lock (lockObj) {
+            requestSend -= observer.OnNext;
+         }
+      }
+   }
+
+   static JsonRpcId? extractRpcId(string messageBody) {
+      try {
+         return JsonNode.Parse(messageBody)
+            ?.AsObject()
+            ?.GetValueOrDefault("id")
+            ?.Deserialize<JsonRpcId>();
+      } catch {
+         return null;
+      }
+   }
+
+   static string getRpcNotification(DataPacket dp) {
+      using (var buffer = new MemoryStream())
+      using (var writer = new Utf8JsonWriter(buffer))
+      using (writer.WriteObject()) {
+         writer.WriteString("method", "subscription");
+         using (writer.WriteObject("params")) {
+            writer.Write("tableName", dp.schema.tableName);
+            writer.Write("columns", dp.schema.columnInfo);
+            var cellJsonFactories = dp.schema.columnInfo.Select(getCellJsonFactory).ToList();
+            using (writer.WriteArray("added")) {
+               foreach (IBlueRow row in (dp.addedRows?.Cast<IBlueRow>()).empty_if_null()) {
+                  using (writer.WriteObject()) {
+                     for (int i = 0; i < cellJsonFactories.Count; i++) {
+                        writer.Write(dp.schema.columnInfo[i].colName, cellJsonFactories[i](row));
+                     }
+                  }
+               }
+            }
+            using (writer.WriteArray("deleted")) {
+               foreach (IBlueRow row in (dp.deletedRows?.Cast<IBlueRow>()).empty_if_null()) {
+                  using (writer.WriteObject()) {
+                     for (int i = 0; i < cellJsonFactories.Count; i++) {
+                        writer.Write(dp.schema.columnInfo[i].colName, cellJsonFactories[i](row));
+                     }
+                  }
+               }
+            }
+         }
+         writer.WriteString("jsonrpc", "2.0");
+         writer.Flush();
+         return Encoding.UTF8.GetString(buffer.ToArray());
+      }
+   }
+
+   static Func<IBlueRow, JsonNode?> getCellJsonFactory(ColumnInfo colInfo) {
+      var t = colInfo.colType;
+      if (t == typeof(int)) {
+         return NullCheckWrapper(v => (int)v);
+      }
+      if (t == typeof(uint)) {
+         return NullCheckWrapper(v => (uint)v);
+      }
+      if (t == typeof(long)) {
+         return NullCheckWrapper(v => (long)v);
+      }
+      if (t == typeof(ulong)) {
+         return NullCheckWrapper(v => (ulong)v);
+      }
+      if (t == typeof(short)) {
+         return NullCheckWrapper(v => (short)v);
+      }
+      if (t == typeof(ushort)) {
+         return NullCheckWrapper(v => (ushort)v);
+      }
+      if (t == typeof(double)) {
+         return NullCheckWrapper(v => (double)v);
+      }
+      if (t == typeof(float)) {
+         return NullCheckWrapper(v => (float)v);
+      }
+      if (t == typeof(bool)) {
+         return NullCheckWrapper(v => (bool)v);
+      }
+      return NullCheckWrapper(v => (string)v);
+
+      Func<IBlueRow, JsonNode?> NullCheckWrapper(Func<object, JsonNode> toJson) =>
+         r => {
+            var obj = r[colInfo.colNo];
+            if (obj is null) {
+               return null;
+            }
+            return toJson(obj);
+         };
+   }
+
+   void onConnect(WebsocketConnectNotification notification) {
+      var id = notification.Id;
+      lock (lockObj) {
+         clients.Add(id, []);
+      }
+      Log.Api.info($"DataStreamWebsocketDataSync client connected. clientId: {notification.Id}");
+      if (notification.Message is not null) {
+         onIncomingMessage(notification.Message);
+      }
+   }
+
+   void onDisconnect(WebsocketDisconnectNotification notification) {
+      Log.Api.warn($"DataStreamWebsocketDataSync client disconnected. clientId: {notification.Id}");
+      unsubscribeAll(notification.Id);
+      clients.Remove(notification.Id);
+   }
+
+   void onIncomingMessage(WebsocketIncomingMessage message) {
+      var id = message.sender_client_id;
+      if (!JsonRpcRequest.TryDeserialize(message, out var req, out var err)) {
+         onBadRequest(message, err);
+         return;
+      }
+      switch (req.Method!.ToLower()) {
+         case "subscribe":
+            subscribe(id, req);
+            break;
+         case "publish":
+            Log.Api.error("DataStreamWebsocket api 'publish' not implemented.");
+            // TODO
+            break;
+         case "unsubscribe":
+            unsubscribe(id, req);
+            break;
+         case "unsubscribe_all":
+            unsubscribeAll(id);
+            break;
+         default:
+            Log.Api.error($"Received request of unknown type: {req}");
+            break;
+      }
+   }
+
+   void sendResponse<T>(WebsocketClientID id, JsonRpcId rpcId, T rpcParams) {
+      var rpcRes = JsonRpcResponse.Create(rpcId, rpcParams);
+      var rpcResStr = JsonSerializer.Serialize(rpcRes, serializerOpts);
+      send([new(id, rpcResStr)]);
+   }
+
+   void sendError<T>(WebsocketClientID id, JsonRpcId rpcId, JsonRpcError<T> error) {
+      var rpcRes = JsonRpcResponse.CreateError(rpcId, error);
+      var rpcResStr = JsonSerializer.Serialize(rpcRes, serializerOpts);
+      send([new(id, rpcResStr)]);
+   }
+
+   bool send(IEnumerable<WebsocketOutgoingMessage> messages) {
+      lock (lockObj) {
+         if (requestSend is null) {
+            Log.Api.warn(
+               $"Unable to send message(s) because '{nameof(requestSend)}' handler is null");
+            return false;
+         }
+         requestSend(messages);
+         return true;
+      }
+   }
+
+   void subscribe(WebsocketClientID id, JsonRpcRequest<JsonObject?> req) {
+      var subParams = req.Params!.Deserialize<SubscriptionParams>()!;
+      var exp = DataStreamExpressionsInterpreter.Evaluate(subParams.Query!);
+      var stream = exp.getView(new Dictionary<string, StatementExpression>(0));
+      var sub = new ClientSubscription(id, subParams, Disposable.Create(Dispose));
+      lock (lockObj) {
+         clients[id].Add(sub);
+         sendResponse(id, req.Id!.Value, req.Params);
+         stream.onChange += OnChange;
+      }
+
+      void OnChange(object? newValue) {
+         var dp = ExceptionAssert.of_type<DataPacket>(newValue);
+         var notif = getRpcNotification(dp);
+         send([new(id, notif)]);
+      }
+
+      void Dispose() =>
+         stream.onChange -= OnChange;
+   }
+
+   void unsubscribe(WebsocketClientID id, JsonRpcRequest<JsonObject?> req) {
+      var subParams = req.Params!.Deserialize<SubscriptionParams>();
+      var subParamsStr = JsonSerializer.Serialize(subParams, serializerOpts);
+      // Minimize amount of code inside lock (just need it to find/remove the subscription)
+      List<ClientSubscription>? subs = null;
+      ClientSubscription? sub = null;
+      lock (lockObj) {
+         subs = clients.GetValueOrDefault(id);
+         subs?.TryRemoveFirst(s => s!.Params.Equals(subParams), out sub);
+      }
+      if (subs is null) {
+         // Didn't find client (log/throw).
+         var msg = $"Unable to unsubscribe: did not find client. Id: {id}.\nParams:\n{subParamsStr}";
+         Log.Api.error(msg);
+         throw new Exception(msg);
+      }
+      if (sub is null) {
+         // Found client, but didn't find subscription (log/throw).
+         var msg = $"Unable to unsubscribe: did not find the specified subscription. " +
+            $"Id: {id}. Request:\n{subParams}";
+         Log.Api.error(msg);
+         throw new Exception(msg);
+      }
+      // Found client and subscription (dispose/remove subscription).
+      Log.Api.info($"Unsubscribing client. Id: {id}.\nParams:\n{subParamsStr}");
+      sub.Dispose();
+   }
+
+   void unsubscribeAll(WebsocketClientID id) {
+      lock (lockObj) {
+         if (!clients.TryGetValue(id, out var subs)) {
+            throw new Exception(
+               $"Unable to unsubscribe: did not find client with id {id}");
+         }
+         subs.DisposeAll();
+         subs.Clear();
+      }
+   }
+
+   void onBadRequest(WebsocketIncomingMessage req, string? errorMsg) {
+      var rpcId = extractRpcId(req.message_body);
+      var logMsg = $"Recieved invalid request:\n{req.message_body}";
+      if (errorMsg is not null) {
+         logMsg += $"\n\n**Exception**\n{errorMsg}";
+      }
+      Log.Api.error(logMsg);
+      if (rpcId is null) {
+         return;
+      }
+      // Send error back to client
+      var rpcErr = JsonRpcError.Create(
+         (long)JsonRpcErrorCode.InvalidRequest,
+         $"Received invalid request: {errorMsg ?? "unknown error"}",
+         req.message_body);
+      sendError(req.sender_client_id, rpcId.Value, rpcErr);
+   }
+}
@@ -0,0 +1,34 @@
+namespace Bts.DataStream.Websocket;
+
+using Newtonsoft.Json.Linq;
+
+using System.Diagnostics.CodeAnalysis;
+
+public static class JObjectExtensions {
+   public static bool TryParse<T>(
+      this JObject filter, string key, [MaybeNull] out T obj, [NotNullWhen(false)] out string? error) {
+      obj = default;
+      var value = filter.GetValueOrDefault(key)?.ToString();
+      if (value == null) {
+         error = $"{key} not present";
+         return true;
+      }
+      return WebsocketJsonParsers.TryParse(value, key, out obj, out error);
+   }
+
+   public static bool TryParse<T>(
+     this JObject filter, string key, out T result, [NotNullWhen(false)] out string? error, T defaultValue) {
+      var value = filter.GetValueOrDefault(key)?.ToString();
+      if (value == null) {
+         result = defaultValue;
+         error = $"{key} not present";
+         return true;
+      }
+      if (WebsocketJsonParsers.TryParse<T>(value, key, out var maybeRes, out error)) {
+         result = maybeRes ?? defaultValue;
+         return true;
+      }
+      result = defaultValue;
+      return false;
+   }
+}
@@ -0,0 +1,24 @@
+namespace Bts.DataStream.Websocket;
+
+using Newtonsoft.Json.Linq;
+
+class JsonParseResult {
+   public string? MessageId { get; init; }
+   public RequestType Type { get; init; }
+   public JObject? Parameters { get; init; }
+   public string? Error { get; init; }
+   public bool Successful => Error is not null;
+
+   public static JsonParseResult CreateError(string error) =>
+      new() {
+         Error = error,
+         MessageId = "missing"
+      };
+
+   public static JsonParseResult Create(string messageId, RequestType type, JObject parameters) =>
+      new() {
+         MessageId = messageId,
+         Parameters = parameters,
+         Type = type
+      };
+}
@@ -0,0 +1,6 @@
+namespace Bts.DataStream.Websocket;
+
+public enum JsonRpcErrorCode {
+   None = 0,
+   InvalidRequest = 1,
+}
@@ -0,0 +1,48 @@
+namespace Bts.DataStream.Websocket;
+
+using Spark.Core;
+
+using System;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+class MultilineStringJsonConverter : JsonConverter<string> {
+   public override string? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) {
+      if (typeToConvert != typeof(string)) {
+         throw new ArgumentException($"Unable to parse {typeToConvert}", nameof(typeToConvert));
+      }
+      return reader.TokenType switch {
+         JsonTokenType.String => StringImpl(ref reader),
+         JsonTokenType.StartArray => ArrayImpl(ref reader),
+         _ => throw new ArgumentException(
+            $"Unable to parse {typeof(SubscriptionParams)} from {reader.TokenType}"),
+      };
+
+      static string? StringImpl(ref Utf8JsonReader reader) =>
+         new(reader.GetString()!);
+
+      static string? ArrayImpl(ref Utf8JsonReader reader) {
+         var lines = new List<string>(2);
+         while (reader.Read()) {
+            switch (reader.TokenType) {
+               case JsonTokenType.String:
+                  lines.Add(reader.GetString()!);
+                  break;
+               case JsonTokenType.EndArray:
+                  return string.Join("\n", lines);
+            }
+         }
+         throw new JsonException(
+            $"Failed to parse {typeof(SubscriptionParams)} (did not reach end of array)");
+      }
+   }
+
+   public override void Write(Utf8JsonWriter writer, string value, JsonSerializerOptions options) {
+      var lines = value.Split("\n");
+      if (lines.Length < 2) {
+         writer.WriteRawValue(value);
+         return;
+      }
+      writer.WriteRawValue($"[{", ".join(lines)}]");
+   }
+}
@@ -0,0 +1,8 @@
+namespace Bts.DataStream.Websocket;
+
+public enum RequestType {
+   Unspecified,
+   Subscribe,
+   Unsubscribe,
+   UnsubscribeAll,
+}
@@ -0,0 +1,27 @@
+namespace Bts.DataStream.Websocket;
+
+using bts;
+
+using Bts.Websocket;
+
+using Google.Protobuf;
+
+using System.Text.Json;
+
+public static class Session {
+   static int sessionIdCounter;
+
+   public static async Task Start(
+      WebsocketServerConfig config, JsonSerializerOptions? opts) {
+      opts ??= JsonSerializerOptions.Default;
+      var sessionId = Interlocked.Increment(ref sessionIdCounter);
+      var gateway = new WebsocketGatewayObservable(config);
+      var sync = new DataSync(opts);
+      using (gateway.Subscribe(sync))
+      using (sync.Subscribe(gateway)) {
+         Log.Api.info($"Starting DataStream Websocket Session {sessionId}");
+         Log.Api.info($"DataStream Websocket Session {sessionId} config:\n{JsonFormatter.Default.Format(config)}");
+         await gateway.ConnectAsync();
+      }
+   }
+}
@@ -0,0 +1,110 @@
+namespace Bts.DataStream.Websocket;
+
+using Spark.Core;
+
+using System;
+using System.Diagnostics.CodeAnalysis;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+public class SubscriptionParams : IEquatable<SubscriptionParams> {
+   static class PropNames {
+      public const string QUERY = "query";
+   }
+
+   public SubscriptionParams(string? query) {
+      Query = query;
+   }
+
+   // Serializer supports plain string, or an array of lines for convenience
+   [MaybeNull]
+   [JsonPropertyName(PropNames.QUERY)]
+   [JsonConverter(typeof(MultilineStringJsonConverter))]
+   public string? Query { get; set; }
+
+   public bool Equals(SubscriptionParams? other) =>
+      other is not null &&
+      Query == other.Query;
+
+   class SubscriptionParamsConverter : JsonConverter<SubscriptionParams> {
+      public override bool CanConvert(Type typeToConvert) =>
+         typeToConvert == typeof(SubscriptionParams);
+
+      public override SubscriptionParams Read(
+         ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) {
+         if (typeToConvert != typeof(SubscriptionParams)) {
+            throw new ArgumentException($"Unable to parse {typeToConvert}", nameof(typeToConvert));
+         }
+         return read(ref reader);
+      }
+
+      SubscriptionParams read(ref Utf8JsonReader reader) {
+         return reader.TokenType switch {
+            JsonTokenType.String => stringImpl(ref reader),
+            JsonTokenType.StartArray => arrayImpl(ref reader),
+            JsonTokenType.StartObject => objectImpl(ref reader),
+            _ => throw new ArgumentException(
+               $"Unable to parse {typeof(SubscriptionParams)} from {reader.TokenType}"),
+         };
+
+         static SubscriptionParams stringImpl(ref Utf8JsonReader reader) =>
+            new(reader.GetString()!);
+
+         static SubscriptionParams arrayImpl(ref Utf8JsonReader reader) {
+            var lines = new List<string>(2);
+            while (reader.Read()) {
+               switch (reader.TokenType) {
+                  case JsonTokenType.String:
+                     lines.Add(reader.GetString()!);
+                     break;
+                  case JsonTokenType.EndArray:
+                     return new("\n".join(lines));
+               }
+            }
+            throw new JsonException(
+               $"Failed to parse {typeof(SubscriptionParams)} (did not reach end of array)");
+         }
+
+         static SubscriptionParams objectImpl(ref Utf8JsonReader reader) {
+            var lines = new List<string>(2);
+            while (reader.Read()) {
+               switch (reader.TokenType) {
+                  case JsonTokenType.PropertyName:
+                     var propName = reader.GetString()!;
+                     if (propName == PropNames.QUERY) {
+                        switch (reader.TokenType) {
+                           case JsonTokenType.StartArray:
+                              return arrayImpl(ref reader);
+                           case JsonTokenType.String:
+                              return stringImpl(ref reader);
+                        }
+                     }
+                     break;
+                  case JsonTokenType.String:
+                     lines.Add(reader.GetString()!);
+                     break;
+                  case JsonTokenType.EndArray:
+                     return new("\n".join(lines));
+               }
+            }
+            throw new JsonException(
+               $"Failed to parse {typeof(SubscriptionParams)} (did not reach end of array)");
+         }
+      }
+
+      public override void Write(
+         Utf8JsonWriter writer, SubscriptionParams value, JsonSerializerOptions options) {
+         var q = value.Query;
+         var lines = q?.Split("\n");
+         if (lines is null || lines.Length < 2) {
+            writer.WriteString(PropNames.QUERY, value.Query);
+            return;
+         }
+         writer.WriteStartArray(PropNames.QUERY);
+         foreach (var l in lines) {
+            writer.WriteStringValue(l);
+         }
+         writer.WriteEndArray();
+      }
+   }
+}
@@ -0,0 +1,5 @@
+global using bts.websocket_core;
+
+global using System;
+
+global using System.Collections.Immutable;
@@ -0,0 +1,10 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>net7.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <EnableDefaultContentItems>True</EnableDefaultContentItems>
+  </PropertyGroup>
+
+</Project>
@@ -0,0 +1,26 @@
+@page
+@model ErrorModel
+@{
+    ViewData["Title"] = "Error";
+}
+
+<h1 class="text-danger">Error.</h1>
+<h2 class="text-danger">An error occurred while processing your request.</h2>
+
+@if (Model.ShowRequestId)
+{
+    <p>
+        <strong>Request ID:</strong> <code>@Model.RequestId</code>
+    </p>
+}
+
+<h3>Development Mode</h3>
+<p>
+    Swapping to the <strong>Development</strong> environment displays detailed information about the error that occurred.
+</p>
+<p>
+    <strong>The Development environment shouldn't be enabled for deployed applications.</strong>
+    It can result in displaying sensitive information from exceptions to end users.
+    For local debugging, enable the <strong>Development</strong> environment by setting the <strong>ASPNETCORE_ENVIRONMENT</strong> environment variable to <strong>Development</strong>
+    and restarting the app.
+</p>
@@ -0,0 +1,27 @@
+using System.Diagnostics;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.Mvc.RazorPages;
+
+namespace Bts.DataStream.WebsocketClient.Pages;
+
+[ResponseCache(Duration = 0, Location = ResponseCacheLocation.None, NoStore = true)]
+[IgnoreAntiforgeryToken]
+public class ErrorModel : PageModel
+{
+    public string? RequestId { get; set; }
+
+    public bool ShowRequestId => !string.IsNullOrEmpty(RequestId);
+
+    private readonly ILogger<ErrorModel> _logger;
+
+    public ErrorModel(ILogger<ErrorModel> logger)
+    {
+        _logger = logger;
+    }
+
+    public void OnGet()
+    {
+        RequestId = Activity.Current?.Id ?? HttpContext.TraceIdentifier;
+    }
+}
+
@@ -0,0 +1,15 @@
+@page
+@model IndexModel
+@{
+  ViewData["Title"] = "Home page";
+}
+
+<div class="text-center">
+  <h1 class="display-4">Welcome</h1>
+  <p>Learn about <a href="https://docs.microsoft.com/aspnet/core">building Web apps with ASP.NET Core</a>.</p>
+  <button id="connectButton">Connect to Websocket</button>
+  <div id="output"></div>
+  <input type="text" id="methodInput" placeholder="Enter method name" />
+  <input type="text" id="paramsInput" placeholder="Enter method parameters (if any)" />
+  <button id="sendMessageButton">Send JSON-RPC Request</button>
+</div>
@@ -0,0 +1,16 @@
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.Mvc.RazorPages;
+
+namespace Bts.DataStream.WebsocketClient.Pages;
+
+public class IndexModel : PageModel {
+   private readonly ILogger<IndexModel> _logger;
+
+   public IndexModel(ILogger<IndexModel> logger) {
+      _logger = logger;
+   }
+
+   public void OnGet() {
+
+   }
+}
@@ -0,0 +1,8 @@
+@page
+@model PrivacyModel
+@{
+    ViewData["Title"] = "Privacy Policy";
+}
+<h1>@ViewData["Title"]</h1>
+
+<p>Use this page to detail your site's privacy policy.</p>
@@ -0,0 +1,19 @@
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.Mvc.RazorPages;
+
+namespace Bts.DataStream.WebsocketClient.Pages;
+
+public class PrivacyModel : PageModel
+{
+    private readonly ILogger<PrivacyModel> _logger;
+
+    public PrivacyModel(ILogger<PrivacyModel> logger)
+    {
+        _logger = logger;
+    }
+
+    public void OnGet()
+    {
+    }
+}
+
@@ -0,0 +1,51 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="utf-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>@ViewData["Title"] - Bts.DataStream.WebsocketClient</title>
+    <link rel="stylesheet" href="~/lib/bootstrap/dist/css/bootstrap.min.css" />
+    <link rel="stylesheet" href="~/css/site.css" asp-append-version="true" />
+    <link rel="stylesheet" href="~/Bts.DataStream.WebsocketClient.styles.css" asp-append-version="true" />
+</head>
+<body>
+    <header>
+        <nav class="navbar navbar-expand-sm navbar-toggleable-sm navbar-light bg-white border-bottom box-shadow mb-3">
+            <div class="container">
+                <a class="navbar-brand" asp-area="" asp-page="/Index">Bts.DataStream.WebsocketClient</a>
+                <button class="navbar-toggler" type="button" data-bs-toggle="collapse" data-bs-target=".navbar-collapse" aria-controls="navbarSupportedContent"
+                        aria-expanded="false" aria-label="Toggle navigation">
+                    <span class="navbar-toggler-icon"></span>
+                </button>
+                <div class="navbar-collapse collapse d-sm-inline-flex justify-content-between">
+                    <ul class="navbar-nav flex-grow-1">
+                        <li class="nav-item">
+                            <a class="nav-link text-dark" asp-area="" asp-page="/Index">Home</a>
+                        </li>
+                        <li class="nav-item">
+                            <a class="nav-link text-dark" asp-area="" asp-page="/Privacy">Privacy</a>
+                        </li>
+                    </ul>
+                </div>
+            </div>
+        </nav>
+    </header>
+    <div class="container">
+        <main role="main" class="pb-3">
+            @RenderBody()
+        </main>
+    </div>
+
+    <footer class="border-top footer text-muted">
+        <div class="container">
+            &copy; 2023 - Bts.DataStream.WebsocketClient - <a asp-area="" asp-page="/Privacy">Privacy</a>
+        </div>
+    </footer>
+
+    <script src="~/lib/jquery/dist/jquery.min.js"></script>
+    <script src="~/lib/bootstrap/dist/js/bootstrap.bundle.min.js"></script>
+    <script src="~/js/site.js" asp-append-version="true"></script>
+
+    @await RenderSectionAsync("Scripts", required: false)
+</body>
+</html>
@@ -0,0 +1,48 @@
+/* Please see documentation at https://docs.microsoft.com/aspnet/core/client-side/bundling-and-minification
+for details on configuring this project to bundle and minify static web assets. */
+
+a.navbar-brand {
+  white-space: normal;
+  text-align: center;
+  word-break: break-all;
+}
+
+a {
+  color: #0077cc;
+}
+
+.btn-primary {
+  color: #fff;
+  background-color: #1b6ec2;
+  border-color: #1861ac;
+}
+
+.nav-pills .nav-link.active, .nav-pills .show > .nav-link {
+  color: #fff;
+  background-color: #1b6ec2;
+  border-color: #1861ac;
+}
+
+.border-top {
+  border-top: 1px solid #e5e5e5;
+}
+.border-bottom {
+  border-bottom: 1px solid #e5e5e5;
+}
+
+.box-shadow {
+  box-shadow: 0 .25rem .75rem rgba(0, 0, 0, .05);
+}
+
+button.accept-policy {
+  font-size: 1rem;
+  line-height: inherit;
+}
+
+.footer {
+  position: absolute;
+  bottom: 0;
+  width: 100%;
+  white-space: nowrap;
+  line-height: 60px;
+}
@@ -0,0 +1,2 @@
+<script src="~/lib/jquery-validation/dist/jquery.validate.min.js"></script>
+<script src="~/lib/jquery-validation-unobtrusive/jquery.validate.unobtrusive.min.js"></script>
@@ -0,0 +1,3 @@
+@using Bts.DataStream.WebsocketClient
+@namespace Bts.DataStream.WebsocketClient.Pages
+@addTagHelper *, Microsoft.AspNetCore.Mvc.TagHelpers
@@ -0,0 +1,3 @@
+@{
+    Layout = "_Layout";
+}
@@ -0,0 +1,26 @@
+using Microsoft.AspNetCore.Mvc;
+
+var builder = WebApplication.CreateBuilder(args);
+
+// Add services to the container.
+builder.Services.AddRazorPages();
+
+var app = builder.Build();
+
+// Configure the HTTP request pipeline.
+if (!app.Environment.IsDevelopment()) {
+   app.UseExceptionHandler("/Error");
+   // The default HSTS value is 30 days. You may want to change this for production scenarios, see https://aka.ms/aspnetcore-hsts.
+   app.UseHsts();
+}
+
+app.UseHttpsRedirection();
+app.UseStaticFiles();
+
+app.UseRouting();
+
+app.UseAuthorization();
+
+app.MapRazorPages();
+
+app.Run();
@@ -0,0 +1,37 @@
+{
+  "iisSettings": {
+    "windowsAuthentication": false,
+    "anonymousAuthentication": true,
+    "iisExpress": {
+      "applicationUrl": "http://localhost:29312",
+      "sslPort": 44390
+    }
+  },
+  "profiles": {
+    "http": {
+      "commandName": "Project",
+      "dotnetRunMessages": true,
+      "launchBrowser": true,
+      "applicationUrl": "http://localhost:5214",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    },
+    "https": {
+      "commandName": "Project",
+      "dotnetRunMessages": true,
+      "launchBrowser": true,
+      "applicationUrl": "https://localhost:7176;http://localhost:5214",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    },
+    "IIS Express": {
+      "commandName": "IISExpress",
+      "launchBrowser": true,
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    }
+  }
+}
@@ -0,0 +1,9 @@
+{
+  "DetailedErrors": true,
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  }
+}
@@ -0,0 +1,9 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  },
+  "AllowedHosts": "*"
+}
@@ -0,0 +1,22 @@
+html {
+  font-size: 14px;
+}
+
+@media (min-width: 768px) {
+  html {
+    font-size: 16px;
+  }
+}
+
+.btn:focus, .btn:active:focus, .btn-link.nav-link:focus, .form-control:focus, .form-check-input:focus {
+  box-shadow: 0 0 0 0.1rem white, 0 0 0 0.25rem #258cfb;
+}
+
+html {
+  position: relative;
+  min-height: 100%;
+}
+
+body {
+  margin-bottom: 60px;
+}
@@ -0,0 +1,67 @@
+// Please see documentation at https://docs.microsoft.com/aspnet/core/client-side/bundling-and-minification
+// for details on configuring this project to bundle and minify static web assets.
+
+// Write your JavaScript code.
+const connectButton = document.getElementById('connectButton');
+const output = document.getElementById('output');
+const methodInput = document.getElementById('methodInput');
+const paramsInput = document.getElementById('paramsInput');
+const sendMessageButton = document.getElementById('sendMessageButton');
+
+let socket;
+
+connectButton.addEventListener('click', () => {
+    if (socket && socket.readyState === WebSocket.OPEN) {
+        return;
+    }
+
+    socket = new WebSocket('ws://localhost:8080'); // Replace with your WebSocket server URL
+
+    socket.onopen = (event) => {
+        output.innerHTML += 'Connected to WebSocket server<br>';
+    };
+
+    socket.onmessage = (event) => {
+        const message = JSON.parse(event.data);
+        output.innerHTML += `Received JSON-RPC response: ${JSON.stringify(message)}<br>`;
+    };
+
+    socket.onclose = (event) => {
+        if (event.wasClean) {
+            output.innerHTML += `Connection closed cleanly, code=${event.code}, reason=${event.reason}<br>`;
+        } else {
+            output.innerHTML += 'Connection abruptly closed<br>';
+        }
+    };
+
+    socket.onerror = (error) => {
+        output.innerHTML += `WebSocket Error: ${error.message}<br>`;
+    };
+});
+
+sendMessageButton.addEventListener('click', () => {
+    // const method = methodInput.value;
+    // const params = paramsInput.value;
+    const method = 'subscribe';
+    const params = `{
+"expression": "literal("n", "2.0")"
+}'
+`
+    if (socket && socket.readyState === WebSocket.OPEN && method) {
+        // const reqParams = {
+        //   'expression': message
+        // }
+        const request = {
+            jsonrpc: '2.0',
+            method: method,
+            params: {
+              query: `literal("n", "2.0")`
+            },
+            id: 1, // You can set a unique ID for each request
+        };
+
+        socket.send(JSON.stringify(request));
+        methodInput.value = ''; // Clear the input fields
+        paramsInput.value = '';
+    }
+});
@@ -0,0 +1,18 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net7.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="System.CommandLine" Version="2.0.0-beta4.22272.1" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Bts.DataStream.Websocket\Bts.DataStream.Websocket.csproj" />
+  </ItemGroup>
+
+</Project>
@@ -0,0 +1,7 @@
+{
+  "host": "localhost",
+  "port": 8080,
+  "scheme": "ws",
+  "default_end_point": "/",
+  "keep_alive_interval_secs": 1
+}
@@ -0,0 +1,76 @@
+using bts.api.core;
+
+using Bts.DataStream.Websocket;
+
+using Google.Protobuf;
+
+using System.CommandLine.Parsing;
+using System.Text.Json;
+
+static WebsocketServerConfig readConfigFile(string path) {
+   using var fs = File.OpenRead(path);
+   using var sr = new StreamReader(fs);
+   return JsonParser.Default.Parse<WebsocketServerConfig>(sr);
+}
+
+static WebsocketServerConfig? readConfigArg(ArgumentResult result) {
+   var path = result.Tokens.Select(t => t.Value).FirstOrDefault() ?? string.Empty;
+   return readConfigFile(path);
+}
+
+static WebsocketServerConfig getDefaultConfig() {
+   const string defaultCfgPath = "DataStreamWebsocketServer.conf";
+   if (File.Exists(defaultCfgPath)) {
+      return readConfigFile(defaultCfgPath);
+   }
+   return new() {
+      Scheme = "http",
+      Host = "localhost",
+      Port = 8080,
+      DefaultEndPoint = "/",
+      KeepAliveIntervalSecs = 1
+   };
+}
+   
+static async Task run(
+   WebsocketServerConfig? wssConfig, string user, string password, string? sparkdConfigSource) {
+   var opts = new JsonSerializerOptions(JsonSerializerDefaults.Web);
+   wssConfig ??= getDefaultConfig();
+   Console.WriteLine("Starting DataStreamWebsocketServer...");
+   var serviceCtx = ServiceContext.create(sparkdConfigSource);
+   await serviceCtx.login(user, password);
+   await Session.Start(wssConfig, opts);
+   // if (sparkdConfigSource is not null && sparkdConfigSource.Length > 9999) {
+   // } else {
+   //    await ChatGptSession.Start();
+   // }
+}
+
+var userOpt = new Option<string>("--user", () => "bts");
+var passOpt = new Option<string>("--password", () => string.Empty);
+var sparkdConfigSourceOpt = new Option<string>(
+   "--sparkd", description: "The sparkd config server url");
+// var bling = new ParseArgument<WebsocketServerConfig?>(parseWsConfig);
+
+// WebsocketServerConfig? parseWsConfig(ArgumentResult result) {
+//    Token t;
+//    TokenType tt;
+//    Symbol s;
+//    throw new NotImplementedException();
+// }
+
+var configOpt = new Option<WebsocketServerConfig?>(
+   "--config", readConfigArg, description: "WebsocketServerConfig file path");
+configOpt.SetDefaultValue(new());
+var rootCommand = new RootCommand("Starts a DataStream Websocket Server");
+rootCommand.AddOption(configOpt);
+rootCommand.AddOption(userOpt);
+rootCommand.AddOption(passOpt);
+rootCommand.SetHandler(run, configOpt, userOpt, passOpt, sparkdConfigSourceOpt);
+
+try {
+   return await rootCommand.InvokeAsync(args);
+} catch (Exception ex) {
+   Console.WriteLine($"Error: {ex}");
+}
+return 1;
@@ -0,0 +1,7 @@
+global using Bts.Websocket;
+
+global using bts.websocket_core;
+
+global using System.CommandLine;
+
+global using System.Collections.Immutable;
@@ -0,0 +1,20 @@
+module.exports = {
+  root: true,
+  env: { browser: true, es2020: true },
+  extends: [
+    'eslint:recommended',
+    'plugin:react/recommended',
+    'plugin:react/jsx-runtime',
+    'plugin:react-hooks/recommended',
+  ],
+  ignorePatterns: ['dist', '.eslintrc.cjs'],
+  parserOptions: { ecmaVersion: 'latest', sourceType: 'module' },
+  settings: { react: { version: '18.2' } },
+  plugins: ['react-refresh'],
+  rules: {
+    'react-refresh/only-export-components': [
+      'warn',
+      { allowConstantExport: true },
+    ],
+  },
+}
@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?
@@ -0,0 +1,11 @@
+<Project Sdk="Microsoft.VisualStudio.JavaScript.Sdk/0.5.128-alpha">
+  <PropertyGroup>
+    <StartupCommand>npm run dev</StartupCommand>
+    <JavaScriptTestRoot>src\</JavaScriptTestRoot>
+    <JavaScriptTestFramework>Jest</JavaScriptTestFramework>
+    <!-- Allows the build (or compile) script located on package.json to run on Build -->
+    <ShouldRunBuildScript>false</ShouldRunBuildScript>
+    <!-- Folder where production build objects will be placed -->
+    <PublishAssetsDirectory>$(MSBuildProjectDirectory)\dist</PublishAssetsDirectory>
+  </PropertyGroup>
+</Project>
@@ -0,0 +1,8 @@
+# React + Vite
+
+This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
+
+Currently, two official plugins are available:
+
+- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react/README.md) uses [Babel](https://babeljs.io/) for Fast Refresh
+- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react-swc) uses [SWC](https://swc.rs/) for Fast Refresh
@@ -0,0 +1,13 @@
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="UTF-8" />
+    <link rel="icon" type="image/svg+xml" href="/vite.svg" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>Vite + React</title>
+  </head>
+  <body>
+    <div id="root"></div>
+    <script type="module" src="/src/main.jsx"></script>
+  </body>
+</html>
@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="utf-8"?>
+<configuration>
+  <packageSources>
+    <clear />
+    <add key="nuget.org" value="https://api.nuget.org/v3/index.json" />
+  </packageSources>
+  <disabledPackageSources>
+    <clear />
+  </disabledPackageSources>
+</configuration>
@@ -0,0 +1,27 @@
+{
+  "name": "bts.datastreamwebapp.client",
+  "private": true,
+  "version": "0.0.0",
+  "type": "module",
+  "scripts": {
+    "dev": "vite",
+    "build": "vite build",
+    "lint": "eslint . --ext js,jsx --report-unused-disable-directives --max-warnings 0",
+    "preview": "vite preview"
+  },
+  "dependencies": {
+    "react": "^18.2.0",
+    "react-dom": "^18.2.0",
+    "react-use-websocket": "^4.5.0"
+  },
+  "devDependencies": {
+    "@types/react": "^18.2.43",
+    "@types/react-dom": "^18.2.17",
+    "@vitejs/plugin-react": "^4.2.1",
+    "eslint": "^8.55.0",
+    "eslint-plugin-react": "^7.33.2",
+    "eslint-plugin-react-hooks": "^4.6.0",
+    "eslint-plugin-react-refresh": "^0.4.5",
+    "vite": "^5.0.8"
+  }
+}
@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" aria-hidden="true" role="img" class="iconify iconify--logos" width="31.88" height="32" preserveAspectRatio="xMidYMid meet" viewBox="0 0 256 257"><defs><linearGradient id="IconifyId1813088fe1fbc01fb466" x1="-.828%" x2="57.636%" y1="7.652%" y2="78.411%"><stop offset="0%" stop-color="#41D1FF"></stop><stop offset="100%" stop-color="#BD34FE"></stop></linearGradient><linearGradient id="IconifyId1813088fe1fbc01fb467" x1="43.376%" x2="50.316%" y1="2.242%" y2="89.03%"><stop offset="0%" stop-color="#FFEA83"></stop><stop offset="8.333%" stop-color="#FFDD35"></stop><stop offset="100%" stop-color="#FFA800"></stop></linearGradient></defs><path fill="url(#IconifyId1813088fe1fbc01fb466)" d="M255.153 37.938L134.897 252.976c-2.483 4.44-8.862 4.466-11.382.048L.875 37.958c-2.746-4.814 1.371-10.646 6.827-9.67l120.385 21.517a6.537 6.537 0 0 0 2.322-.004l117.867-21.483c5.438-.991 9.574 4.796 6.877 9.62Z"></path><path fill="url(#IconifyId1813088fe1fbc01fb467)" d="M185.432.063L96.44 17.501a3.268 3.268 0 0 0-2.634 3.014l-5.474 92.456a3.268 3.268 0 0 0 3.997 3.378l24.777-5.718c2.318-.535 4.413 1.507 3.936 3.838l-7.361 36.047c-.495 2.426 1.782 4.5 4.151 3.78l15.304-4.649c2.372-.72 4.652 1.36 4.15 3.788l-11.698 56.621c-.732 3.542 3.979 5.473 5.943 2.437l1.313-2.028l72.516-144.72c1.215-2.423-.88-5.186-3.54-4.672l-25.505 4.922c-2.396.462-4.435-1.77-3.759-4.114l16.646-57.705c.677-2.35-1.37-4.583-3.769-4.113Z"></path></svg>
@@ -0,0 +1,44 @@
+#root {
+    max-width: 1280px;
+    margin: 0 auto;
+    padding: 2rem;
+    text-align: center;
+    --cell-background-color: #293136;
+    --cell-color: #cbe3ef;
+}
+
+tr:nth-child(even) {
+    background: #F2F2F2;
+}
+
+tr:nth-child(odd) {
+    background: #FFF;
+}
+
+th, td {
+    padding-left: 1rem;
+    padding-right: 1rem;
+    background-color: var(--cell-background-color);
+    color: var(--cell-color);
+}
+
+textarea#query-input {
+  height: 5em;
+  width: 50em;
+}
+
+button#query-button {
+  height: 5em;
+  vertical-align: top;
+}
+
+table#query-output {
+  float: none;
+  clear: both;
+}
+
+p#footer {
+  float: none;
+  clear: both;
+  vertical-align: bottom;
+}
@@ -0,0 +1,88 @@
+import { Component } from "react";
+import "./App.css";
+
+export class Home extends Component {
+  static displayName = Home.name;
+  static sparkdHost = "localhost";
+
+  render() {
+    let socket = null;
+    const wsUrl = `ws://localhost:5268/ws`;
+
+    function connectToSparkd() {
+      if (socket == null) {
+        socket = new WebSocket(wsUrl);
+      }
+      if (socket && socket.readyState === WebSocket.OPEN) {
+        return;
+      }
+
+      const output = document.getElementById("query-output");
+      socket.onopen = (event) => {
+        console.log(`Connected to ${wsUrl}`);
+      };
+
+      socket.onmessage = (event) => {
+        console.log(`Received message: ${event.data}`);
+        output.innerHTML = event.data;
+      };
+
+      socket.onerror = (error) => {
+        console.log(`Error on websocket ${wsUrl}: ${error}`);
+      };
+
+      socket.onclose = (event) => {
+        if (event.wasClean) {
+          const msg = `Connection to ${wsUrl} closed cleanly, code=${event.code}, reason=${event.reason}`;
+          console.log(msg);
+        } else {
+          const msg = `Connection to ${wsUrl} abruptly closed`;
+          console.log(msg);
+        }
+      };
+    }
+
+    function executeQuery() {
+      if (socket == null || !socket) {
+        connectToSparkd();
+      }
+      if (socket == null || !socket) {
+        console.log("Failed to connect to websocket server");
+        return;
+      }
+      if (socket.readyState === WebSocket.OPEN) {
+        const queryInput = document.getElementById("query-input");
+        const rpcReq = {
+          jsonrpc: "2.0",
+          method: "subscribe",
+          params: {
+            query: queryInput.value,
+          },
+          id: 1,
+        };
+        socket.send(JSON.stringify(rpcReq));
+      }
+    }
+
+    return (
+      <div>
+        <h1>Data Stream Query Analyzer</h1>
+        <p>Run queries against the BTS Cloud</p>
+        <div>
+          <textarea id="query-input" placeholder="Enter Query"></textarea>
+          <button id="query-button" onClick={executeQuery}>
+            Execute
+          </button>
+        </div>
+        <table id="query-output"></table>
+        <p id="footer">
+          The <code>ClientApp</code> subdirectory is a standard React
+          application based on the <code>create-react-app</code> template. If
+          you open a command prompt in that directory, you can run{" "}
+          <code>npm</code> commands such as <code>npm test</code> or{" "}
+          <code>npm install</code>.
+        </p>
+      </div>
+    );
+  }
+}
@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" aria-hidden="true" role="img" class="iconify iconify--logos" width="35.93" height="32" preserveAspectRatio="xMidYMid meet" viewBox="0 0 256 228"><path fill="#00D8FF" d="M210.483 73.824a171.49 171.49 0 0 0-8.24-2.597c.465-1.9.893-3.777 1.273-5.621c6.238-30.281 2.16-54.676-11.769-62.708c-13.355-7.7-35.196.329-57.254 19.526a171.23 171.23 0 0 0-6.375 5.848a155.866 155.866 0 0 0-4.241-3.917C100.759 3.829 77.587-4.822 63.673 3.233C50.33 10.957 46.379 33.89 51.995 62.588a170.974 170.974 0 0 0 1.892 8.48c-3.28.932-6.445 1.924-9.474 2.98C17.309 83.498 0 98.307 0 113.668c0 15.865 18.582 31.778 46.812 41.427a145.52 145.52 0 0 0 6.921 2.165a167.467 167.467 0 0 0-2.01 9.138c-5.354 28.2-1.173 50.591 12.134 58.266c13.744 7.926 36.812-.22 59.273-19.855a145.567 145.567 0 0 0 5.342-4.923a168.064 168.064 0 0 0 6.92 6.314c21.758 18.722 43.246 26.282 56.54 18.586c13.731-7.949 18.194-32.003 12.4-61.268a145.016 145.016 0 0 0-1.535-6.842c1.62-.48 3.21-.974 4.76-1.488c29.348-9.723 48.443-25.443 48.443-41.52c0-15.417-17.868-30.326-45.517-39.844Zm-6.365 70.984c-1.4.463-2.836.91-4.3 1.345c-3.24-10.257-7.612-21.163-12.963-32.432c5.106-11 9.31-21.767 12.459-31.957c2.619.758 5.16 1.557 7.61 2.4c23.69 8.156 38.14 20.213 38.14 29.504c0 9.896-15.606 22.743-40.946 31.14Zm-10.514 20.834c2.562 12.94 2.927 24.64 1.23 33.787c-1.524 8.219-4.59 13.698-8.382 15.893c-8.067 4.67-25.32-1.4-43.927-17.412a156.726 156.726 0 0 1-6.437-5.87c7.214-7.889 14.423-17.06 21.459-27.246c12.376-1.098 24.068-2.894 34.671-5.345a134.17 134.17 0 0 1 1.386 6.193ZM87.276 214.515c-7.882 2.783-14.16 2.863-17.955.675c-8.075-4.657-11.432-22.636-6.853-46.752a156.923 156.923 0 0 1 1.869-8.499c10.486 2.32 22.093 3.988 34.498 4.994c7.084 9.967 14.501 19.128 21.976 27.15a134.668 134.668 0 0 1-4.877 4.492c-9.933 8.682-19.886 14.842-28.658 17.94ZM50.35 144.747c-12.483-4.267-22.792-9.812-29.858-15.863c-6.35-5.437-9.555-10.836-9.555-15.216c0-9.322 13.897-21.212 37.076-29.293c2.813-.98 5.757-1.905 8.812-2.773c3.204 10.42 7.406 21.315 12.477 32.332c-5.137 11.18-9.399 22.249-12.634 32.792a134.718 134.718 0 0 1-6.318-1.979Zm12.378-84.26c-4.811-24.587-1.616-43.134 6.425-47.789c8.564-4.958 27.502 2.111 47.463 19.835a144.318 144.318 0 0 1 3.841 3.545c-7.438 7.987-14.787 17.08-21.808 26.988c-12.04 1.116-23.565 2.908-34.161 5.309a160.342 160.342 0 0 1-1.76-7.887Zm110.427 27.268a347.8 347.8 0 0 0-7.785-12.803c8.168 1.033 15.994 2.404 23.343 4.08c-2.206 7.072-4.956 14.465-8.193 22.045a381.151 381.151 0 0 0-7.365-13.322Zm-45.032-43.861c5.044 5.465 10.096 11.566 15.065 18.186a322.04 322.04 0 0 0-30.257-.006c4.974-6.559 10.069-12.652 15.192-18.18ZM82.802 87.83a323.167 323.167 0 0 0-7.227 13.238c-3.184-7.553-5.909-14.98-8.134-22.152c7.304-1.634 15.093-2.97 23.209-3.984a321.524 321.524 0 0 0-7.848 12.897Zm8.081 65.352c-8.385-.936-16.291-2.203-23.593-3.793c2.26-7.3 5.045-14.885 8.298-22.6a321.187 321.187 0 0 0 7.257 13.246c2.594 4.48 5.28 8.868 8.038 13.147Zm37.542 31.03c-5.184-5.592-10.354-11.779-15.403-18.433c4.902.192 9.899.29 14.978.29c5.218 0 10.376-.117 15.453-.343c-4.985 6.774-10.018 12.97-15.028 18.486Zm52.198-57.817c3.422 7.8 6.306 15.345 8.596 22.52c-7.422 1.694-15.436 3.058-23.88 4.071a382.417 382.417 0 0 0 7.859-13.026a347.403 347.403 0 0 0 7.425-13.565Zm-16.898 8.101a358.557 358.557 0 0 1-12.281 19.815a329.4 329.4 0 0 1-23.444.823c-7.967 0-15.716-.248-23.178-.732a310.202 310.202 0 0 1-12.513-19.846h.001a307.41 307.41 0 0 1-10.923-20.627a310.278 310.278 0 0 1 10.89-20.637l-.001.001a307.318 307.318 0 0 1 12.413-19.761c7.613-.576 15.42-.876 23.31-.876H128c7.926 0 15.743.303 23.354.883a329.357 329.357 0 0 1 12.335 19.695a358.489 358.489 0 0 1 11.036 20.54a329.472 329.472 0 0 1-11 20.722Zm22.56-122.124c8.572 4.944 11.906 24.881 6.52 51.026c-.344 1.668-.73 3.367-1.15 5.09c-10.622-2.452-22.155-4.275-34.23-5.408c-7.034-10.017-14.323-19.124-21.64-27.008a160.789 160.789 0 0 1 5.888-5.4c18.9-16.447 36.564-22.941 44.612-18.3ZM128 90.808c12.625 0 22.86 10.235 22.86 22.86s-10.235 22.86-22.86 22.86s-22.86-10.235-22.86-22.86s10.235-22.86 22.86-22.86Z"></path></svg>
@@ -0,0 +1,68 @@
+:root {
+  font-family: Inter, system-ui, Avenir, Helvetica, Arial, sans-serif;
+  line-height: 1.5;
+  font-weight: 400;
+
+  color-scheme: light dark;
+  color: rgba(255, 255, 255, 0.87);
+  background-color: #242424;
+
+  font-synthesis: none;
+  text-rendering: optimizeLegibility;
+  -webkit-font-smoothing: antialiased;
+  -moz-osx-font-smoothing: grayscale;
+}
+
+a {
+  font-weight: 500;
+  color: #646cff;
+  text-decoration: inherit;
+}
+a:hover {
+  color: #535bf2;
+}
+
+body {
+  margin: 0;
+  display: flex;
+  place-items: center;
+  min-width: 320px;
+  min-height: 100vh;
+}
+
+h1 {
+  font-size: 3.2em;
+  line-height: 1.1;
+}
+
+button {
+  border-radius: 8px;
+  border: 1px solid transparent;
+  padding: 0.6em 1.2em;
+  font-size: 1em;
+  font-weight: 500;
+  font-family: inherit;
+  background-color: #1a1a1a;
+  cursor: pointer;
+  transition: border-color 0.25s;
+}
+button:hover {
+  border-color: #646cff;
+}
+button:focus,
+button:focus-visible {
+  outline: 4px auto -webkit-focus-ring-color;
+}
+
+@media (prefers-color-scheme: light) {
+  :root {
+    color: #213547;
+    background-color: #ffffff;
+  }
+  a:hover {
+    color: #747bff;
+  }
+  button {
+    background-color: #f9f9f9;
+  }
+}
@@ -0,0 +1,10 @@
+import React from 'react'
+import ReactDOM from 'react-dom/client'
+import {Home} from './App.jsx'
+import './index.css'
+
+ReactDOM.createRoot(document.getElementById('root')).render(
+  <React.StrictMode>
+    <Home />
+  </React.StrictMode>,
+)
@@ -0,0 +1,60 @@
+import { fileURLToPath, URL } from 'node:url';
+
+import { defineConfig } from 'vite';
+import plugin from '@vitejs/plugin-react';
+import fs from 'fs';
+import path from 'path';
+import child_process from 'child_process';
+
+const baseFolder =
+    process.env.APPDATA !== undefined && process.env.APPDATA !== ''
+        ? `${process.env.APPDATA}/ASP.NET/https`
+        : `${process.env.HOME}/.aspnet/https`;
+
+const certificateArg = process.argv.map(arg => arg.match(/--name=(?<value>.+)/i)).filter(Boolean)[0];
+const certificateName = certificateArg ? certificateArg.groups.value : "bts.datastreamwebapp.client";
+
+if (!certificateName) {
+    console.error('Invalid certificate name. Run this script in the context of an npm/yarn script or pass --name=<<app>> explicitly.')
+    process.exit(-1);
+}
+
+const certFilePath = path.join(baseFolder, `${certificateName}.pem`);
+const keyFilePath = path.join(baseFolder, `${certificateName}.key`);
+
+if (!fs.existsSync(certFilePath) || !fs.existsSync(keyFilePath)) {
+    if (0 !== child_process.spawnSync('dotnet', [
+        'dev-certs',
+        'https',
+        '--export-path',
+        certFilePath,
+        '--format',
+        'Pem',
+        '--no-password',
+    ], { stdio: 'inherit', }).status) {
+        throw new Error("Could not create certificate.");
+    }
+}
+
+// https://vitejs.dev/config/
+export default defineConfig({
+    plugins: [plugin()],
+    resolve: {
+        alias: {
+            '@': fileURLToPath(new URL('./src', import.meta.url))
+        }
+    },
+    server: {
+        proxy: {
+            '^/weatherforecast': {
+                target: 'https://localhost:7128/',
+                secure: false
+            }
+        },
+        port: 5173,
+        https: {
+            key: fs.readFileSync(keyFilePath),
+            cert: fs.readFileSync(certFilePath),
+        }
+    }
+})
@@ -0,0 +1,35 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <InvariantGlobalization>true</InvariantGlobalization>
+    <SpaRoot>..\Bts.DataStreamWebApp.Client</SpaRoot>
+    <SpaProxyLaunchCommand>npm run dev</SpaProxyLaunchCommand>
+    <SpaProxyServerUrl>https://localhost:5173</SpaProxyServerUrl>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="CommandLineParser" Version="2.9.1" />
+    <PackageReference Include="Microsoft.AspNetCore.SpaProxy">
+      <Version>8.*-*</Version>
+    </PackageReference>
+    <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
+    <PackageReference Include="Swashbuckle.AspNetCore" Version="6.4.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Bts.DataStreamWebApp.Client\Bts.DataStreamWebApp.Client.esproj">
+      <ReferenceOutputAssembly>false</ReferenceOutputAssembly>
+    </ProjectReference>
+    <ProjectReference Include="..\DataStream\DataStream.csproj" />
+    <ProjectReference Include="..\SharedStreams\SharedStreams.csproj" />
+    <ProjectReference Include="..\..\client\spark_api\spark_api.csproj" />
+    <ProjectReference Include="..\..\client\spark_websocket\spark_websocket.csproj" />
+  </ItemGroup>
+
+  <!-- <ItemGroup>
+    <Content Update="wwwroot\alt-index.html" CopyToPublishDirectory="Never" />
+  </ItemGroup> -->
+</Project>
@@ -0,0 +1,6 @@
+@Bts.DataStreamWebApp.Server_HostAddress = http://localhost:5268
+
+GET {{Bts.DataStreamWebApp.Server_HostAddress}}/weatherforecast/
+Accept: application/json
+
+###
@@ -0,0 +1,106 @@
+namespace Bts.DataStreamWebApp.Server.Controllers;
+
+using BCG;
+
+using bts.api.core;
+
+using Bts.DataStreamExpressions;
+using Bts.Websocket;
+
+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 queryArgs = rpcReq!.Params;
+            var exp = DataStreamExpressionsInterpreter.Evaluate(queryArgs.Query!);
+            if (exp is null) {
+               Console.WriteLine($"Invalid query expression: {queryArgs.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);
+      }
+   }
+}
@@ -0,0 +1,22 @@
+namespace Bts.DataStreamWebApp.Server.Controllers;
+using Microsoft.AspNetCore.Mvc;
+
+[ApiController]
+[Route("[controller]")]
+public class WeatherForecastController(ILogger<WeatherForecastController> logger) : ControllerBase {
+   static readonly string[] Summaries = [
+      "Freezing", "Bracing", "Chilly", "Cool", "Mild", "Warm", "Balmy", "Hot", "Sweltering", "Scorching"
+   ];
+
+   readonly ILogger<WeatherForecastController> _logger = logger;
+
+   [HttpGet(Name = "GetWeatherForecast")]
+   public IEnumerable<WeatherForecast> Get() {
+      return Enumerable.Range(1, 5).Select(index => new WeatherForecast {
+         Date = DateOnly.FromDateTime(DateTime.Now.AddDays(index)),
+         TemperatureC = Random.Shared.Next(-20, 55),
+         Summary = Summaries[Random.Shared.Next(Summaries.Length)]
+      })
+      .ToArray();
+   }
+}
@@ -0,0 +1,8 @@
+namespace Bts.DataStreamWebApp.Server;
+
+using System.Text.Json.Serialization;
+
+public class DataStreamSubscription {
+   [JsonPropertyName("query")]
+   public string? Query { get; set; }
+}
@@ -0,0 +1,8 @@
+namespace Bts.DataStreamWebApp.Server;
+
+using BCG;
+
+class DataStreamUpdate {
+   public List<IBlueRow> addedRows { get; set; } = new(0);
+   public List<IBlueRow> deletedRows { get; set; } = new(0);
+}
@@ -0,0 +1,14 @@
+namespace Bts.DataStreamWebApp.Server;
+
+using CommandLine;
+
+class DataStreamWebAppServerOptions {
+   [Option("sparkd")]
+   public string? sparkd { get; set; }
+
+   [Option("user", Default = "bts")]
+   public string? user { get; set; }
+
+   [Option("password", Default = "")]
+   public string? password { get; set; }
+}
@@ -0,0 +1,58 @@
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.Logging;
+
+using System;
+using System.Net.WebSockets;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Bts.DataStreamWebApp.Server {
+   // public class DataStreamWebSocketHandler {
+   //    readonly WebSocketConnectionManager _connectionManager = new();
+   //    readonly ILogger<DataStreamWebSocketHandler> _logger;
+
+   //    public DataStreamWebSocketHandler(
+   //       WebSocketConnectionManager connectionManager,
+   //       ILogger<DataStreamWebSocketHandler> logger
+   //    ) {
+   //       _connectionManager = connectionManager;
+   //       _logger = logger;
+   //    }
+
+   //    public async Task HandleWebSocket(HttpContext context, WebSocket webSocket) {
+   //       var socketId = _connectionManager.AddSocket(webSocket);
+
+   //       _logger.LogInformation($"WebSocket connection established with ID {socketId}");
+
+   //       while (webSocket.State == WebSocketState.Open) {
+   //          var message = await ReceiveMessageAsync(webSocket);
+   //          if (message != null) {
+   //             _logger.LogInformation($"Received message from ID {socketId}: {message}");
+   //             await BroadcastMessageAsync(message);
+   //          }
+   //       }
+
+   //       _connectionManager.RemoveSocket(socketId);
+   //       _logger.LogInformation($"WebSocket connection closed with ID {socketId}");
+   //    }
+
+   //    async Task<string?> ReceiveMessageAsync(WebSocket webSocket) {
+   //       var buffer = new byte[1024];
+   //       var result = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
+
+   //       if (result.CloseStatus.HasValue) {
+   //          return null;
+   //       }
+
+   //       return System.Text.Encoding.UTF8.GetString(buffer, 0, result.Count);
+   //    }
+
+   //    async Task BroadcastMessageAsync(string message) {
+   //       foreach (var socket in _connectionManager.GetAllSockets()) {
+   //          if (socket.Value.State == WebSocketState.Open) {
+   //             await socket.Value.SendAsync(System.Text.Encoding.UTF8.GetBytes(message), WebSocketMessageType.Text, true, CancellationToken.None);
+   //          }
+   //       }
+   //    }
+   // }
+}
@@ -0,0 +1,158 @@
+using BCG;
+
+using bts;
+using bts.api.core;
+using bts.utils;
+
+using Bts.DataStreamExpressions;
+using Bts.DataStreamWebApp.Server;
+using Bts.DataStreamWebApp.Server.Controllers;
+
+using CommandLine;
+
+using mbts.config;
+
+using System.Net.WebSockets;
+using System.Reactive.Disposables;
+using System.Reactive.Linq;
+using System.Text;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+static string fmtReq(HttpRequest req) =>
+   $"{req.Protocol}://{req.Host}/{req.Path}?{req.QueryString}";
+
+static string fmtDataPacket(DataPacket dp) {
+   var sb = new StringBuilder();
+   var rows = dp.addedRows.Cast<IBlueRow>().ToList();
+   var colWidths = dp.schema.columnInfo.Select(c => rows.Max(r => r[c.colNo]?.ToString()?.Length ?? 0)).ToArray();
+   foreach (IBlueRow r in dp.addedRows) {
+      for (int i = 0; i < colWidths.Length; i++) {
+         var str = r[i]?.ToString() ?? string.Empty;
+         if (i > 0) {
+            sb.Append(" | ");
+         }
+         sb.Append(str);
+         var width = colWidths[i];
+         if (str.Length < width) {
+            sb.Append(new string(' ', width - str.Length));
+         }
+      }
+      sb.AppendLine();
+   }
+   return sb.ToString();
+}
+
+var parser = new Parser(ps => {
+   ps.CaseInsensitiveEnumValues = true;
+   ps.HelpWriter = Console.Out;
+});
+
+await parser.ParseArguments<DataStreamWebAppServerOptions>(args).
+   MapResult(run, errors => Task.FromResult(1));
+
+async Task<int> run(DataStreamWebAppServerOptions opts) {
+   BTSNativeMethods.provide_command_line(args, args.Length);
+   Configuration.initialize();
+   var context = ServiceContext.create(opts.sparkd);
+   await context.login(opts.user!, opts.password!);
+   var vsWorkLoop = Observable.Interval(TimeSpan.FromMilliseconds(200)).Subscribe(_ => {
+      ViewStreamDatabase.instance.doWork();
+      MasterWorker.getInstance(false).doWork();
+   });
+   var builder = WebApplication.CreateBuilder(args);
+   builder.Services.AddDirectoryBrowser();
+   var app = builder.Build();
+   var wsOpts = new WebSocketOptions {
+      KeepAliveInterval = TimeSpan.FromMinutes(2)
+   };
+   app.UseWebSockets(wsOpts);
+   app.UseDefaultFiles();
+   app.UseStaticFiles();
+   app.MapFallbackToFile("/index.html");
+
+   app.Use(async (context, next) => {
+      var reqDesc = fmtReq(context.Request);
+      Console.WriteLine($"Webserver received: {reqDesc}");
+      if (context.Request.Path == "/ws") {
+         if (context.WebSockets.IsWebSocketRequest) {
+            Console.WriteLine("Accepting websocket connection...");
+            using var webSocket = await context.WebSockets.AcceptWebSocketAsync();
+            Console.WriteLine("Websocket connection accepted.");
+            await runWebSocketServer(
+               opts.sparkd, opts.user!, opts.password!, webSocket);
+         } else {
+            context.Response.StatusCode = StatusCodes.Status400BadRequest;
+         }
+      } else {
+         Console.WriteLine($"Webserver ignoring req: {reqDesc}");
+         await next(context);
+      }
+   });
+
+   try {
+      await app.RunAsync();
+   } catch (Exception ex) {
+      Console.WriteLine($"Exception on DataStreamWebsocketServer thread: {ex}");
+      return 1;
+   }
+   Spark.Core.Disposables.Disposable.dispose(ref vsWorkLoop);
+   return 0;
+}
+
+async Task<int> runWebSocketServer(string? sparkd, string user, string password, WebSocket webSocket) {
+   var buffer = new byte[1024 * 4];
+   var receiveResult = await webSocket.ReceiveAsync(
+       new ArraySegment<byte>(buffer), CancellationToken.None);
+
+   try {
+      while (!receiveResult.CloseStatus.HasValue) {
+         Console.WriteLine("Awaiting websocket message from client");
+         receiveResult = await webSocket.ReceiveAsync(
+             new ArraySegment<byte>(buffer), CancellationToken.None);
+         var msg = Encoding.UTF8.GetString(buffer, 0, receiveResult.Count);
+         var rpcReq = JsonSerializer.Deserialize<Bts.Websocket.JsonRpcRequest<DataStreamSubscription>>(msg);
+         var queryArgs = rpcReq!.Params!;
+         // var queryArgs = new DataStreamSubscription { Query = msg };
+         var exp = DataStreamExpressionsInterpreter.Evaluate(queryArgs.Query!);
+         if (exp is null) {
+            Console.WriteLine($"Invalid query expression: {queryArgs.Query}.");
+            continue;
+         }
+         var vs = exp.getView(new Dictionary<string, StatementExpression>(0));
+         Console.WriteLine("Subscribing to stream...");
+         vs.onChange += onStreamChanged;
+
+         await webSocket.SendAsync(
+             new ArraySegment<byte>(buffer, 0, receiveResult.Count),
+             receiveResult.MessageType,
+             receiveResult.EndOfMessage,
+             CancellationToken.None);
+      }
+   } 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 sb = new StringBuilder();
+      var updateDesc = fmtDataPacket(dp);
+      var msg = $"Received new rows:\n{updateDesc}";
+      // var payload = new ReadOnlyMemory<byte>([.. msg]);
+      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);
+   }
+}
@@ -0,0 +1,48 @@
+{
+  "profiles": {
+    "default": {
+      "commandName": "Project",
+      "launchBrowser": true,
+      "applicationUrl": "http://localhost:5400"
+    },
+    "http": {
+      "commandName": "Project",
+      "launchBrowser": true,
+      "launchUrl": "swagger",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development",
+        "ASPNETCORE_HOSTINGSTARTUPASSEMBLIES": "Microsoft.AspNetCore.SpaProxy"
+      },
+      "dotnetRunMessages": true,
+      "applicationUrl": "http://localhost:5268"
+    },
+    "https": {
+      "commandName": "Project",
+      "launchUrl": "swagger",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development",
+        "ASPNETCORE_HOSTINGSTARTUPASSEMBLIES": "Microsoft.AspNetCore.SpaProxy"
+      },
+      "dotnetRunMessages": true,
+      "applicationUrl": "https://localhost:7128;http://localhost:5268"
+    },
+    "IIS Express": {
+      "commandName": "IISExpress",
+      "launchBrowser": true,
+      "launchUrl": "swagger",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development",
+        "ASPNETCORE_HOSTINGSTARTUPASSEMBLIES": "Microsoft.AspNetCore.SpaProxy"
+      }
+    }
+  },
+  "$schema": "http://json.schemastore.org/launchsettings.json",
+  "iisSettings": {
+    "windowsAuthentication": false,
+    "anonymousAuthentication": true,
+    "iisExpress": {
+      "applicationUrl": "http://localhost:2083",
+      "sslPort": 44383
+    }
+  }
+}
@@ -0,0 +1,12 @@
+namespace Bts.DataStreamWebApp.Server;
+
+public class WeatherForecast
+{
+   public DateOnly Date { get; set; }
+
+   public int TemperatureC { get; set; }
+
+   public int TemperatureF => 32 + (int)(TemperatureC / 0.5556);
+
+   public string? Summary { get; set; }
+}
@@ -0,0 +1,37 @@
+using System;
+using System.Collections.Concurrent;
+using System.Net.WebSockets;
+
+namespace Bts.DataStreamWebApp.Server {
+   // public class WebSocketConnectionManager {
+   //    private readonly ConcurrentDictionary<Guid, WebSocket> _sockets = new ConcurrentDictionary<Guid, WebSocket>();
+
+   //    public Guid AddSocket(WebSocket socket) {
+   //       var socketId = Guid.NewGuid();
+   //       _sockets.TryAdd(socketId, socket);
+   //       return socketId;
+   //    }
+
+   //    public WebSocket? GetSocket(Guid socketId) {
+   //       _sockets.TryGetValue(socketId, out var socket);
+   //       return socket;
+   //    }
+
+   //    public ConcurrentDictionary<Guid, WebSocket>.ValueCollection GetAllSockets() {
+   //       return _sockets.Values;
+   //    }
+
+   //    public Guid? GetSocketId(WebSocket socket) {
+   //       foreach (var (key, value) in _sockets) {
+   //          if (value == socket) {
+   //             return key;
+   //          }
+   //       }
+   //       return null;
+   //    }
+
+   //    public void RemoveSocket(Guid socketId) {
+   //       _sockets.TryRemove(socketId, out _);
+   //    }
+   // }
+}
@@ -0,0 +1,8 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  }
+}
@@ -0,0 +1,9 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  },
+  "AllowedHosts": "*"
+}
@@ -0,0 +1,168 @@
+<!DOCTYPE html>
+<html>
+<head>
+    <meta charset="utf-8" />
+    <title></title>
+    <style>
+        table {
+            border: 0
+        }
+
+        .commslog-data {
+            font-family: Consolas, Courier New, Courier, monospace;
+        }
+
+        .commslog-server {
+            background-color: red;
+            color: white
+        }
+
+        .commslog-client {
+            background-color: green;
+            color: white
+        }
+    </style>
+</head>
+<body>
+    <h1>ALT WebSocket Sample Application</h1>
+    <p id="stateLabel">Ready to connect...</p>
+    <div>
+        <label for="connectionUrl">WebSocket Server URL:</label>
+        <input id="connectionUrl" />
+        <button id="connectButton" type="submit">Connect</button>
+    </div>
+    <p></p>
+    <div>
+        <label for="sendMessage">Message to send:</label>
+        <input id="sendMessage" disabled />
+        <button id="sendButton" type="submit" disabled>Send</button>
+        <button id="closeButton" disabled>Close Socket</button>
+    </div>
+
+    <h2>Communication Log</h2>
+    <table style="width: 800px">
+        <thead>
+            <tr>
+                <td style="width: 100px">From</td>
+                <td style="width: 100px">To</td>
+                <td>Data</td>
+            </tr>
+        </thead>
+        <tbody id="commsLog">
+        </tbody>
+    </table>
+
+    <script>
+        var connectionUrl = document.getElementById("connectionUrl");
+        var connectButton = document.getElementById("connectButton");
+        var stateLabel = document.getElementById("stateLabel");
+        var sendMessage = document.getElementById("sendMessage");
+        var sendButton = document.getElementById("sendButton");
+        var commsLog = document.getElementById("commsLog");
+        var closeButton = document.getElementById("closeButton");
+        var socket;
+
+        var scheme = document.location.protocol === "https:" ? "wss" : "ws";
+        var port = document.location.port ? (":" + document.location.port) : "";
+
+        connectionUrl.value = scheme + "://" + document.location.hostname + port + "/ws" ;
+
+        function updateState() {
+            function disable() {
+                sendMessage.disabled = true;
+                sendButton.disabled = true;
+                closeButton.disabled = true;
+            }
+            function enable() {
+                sendMessage.disabled = false;
+                sendButton.disabled = false;
+                closeButton.disabled = false;
+            }
+
+            connectionUrl.disabled = true;
+            connectButton.disabled = true;
+
+            if (!socket) {
+                disable();
+            } else {
+                switch (socket.readyState) {
+                    case WebSocket.CLOSED:
+                        stateLabel.innerHTML = "Closed";
+                        disable();
+                        connectionUrl.disabled = false;
+                        connectButton.disabled = false;
+                        break;
+                    case WebSocket.CLOSING:
+                        stateLabel.innerHTML = "Closing...";
+                        disable();
+                        break;
+                    case WebSocket.CONNECTING:
+                        stateLabel.innerHTML = "Connecting...";
+                        disable();
+                        break;
+                    case WebSocket.OPEN:
+                        stateLabel.innerHTML = "Open";
+                        enable();
+                        break;
+                    default:
+                        stateLabel.innerHTML = "Unknown WebSocket State: " + htmlEscape(socket.readyState);
+                        disable();
+                        break;
+                }
+            }
+        }
+
+        closeButton.onclick = function () {
+            if (!socket || socket.readyState !== WebSocket.OPEN) {
+                alert("socket not connected");
+            }
+            socket.close(1000, "Closing from client");
+        };
+
+        sendButton.onclick = function () {
+            if (!socket || socket.readyState !== WebSocket.OPEN) {
+                alert("socket not connected");
+            }
+            var data = sendMessage.value;
+            socket.send(data);
+            commsLog.innerHTML += '<tr>' +
+                '<td class="commslog-client">Client</td>' +
+                '<td class="commslog-server">Server</td>' +
+                '<td class="commslog-data">' + htmlEscape(data) + '</td></tr>';
+        };
+
+        connectButton.onclick = function() {
+            stateLabel.innerHTML = "Connecting";
+            socket = new WebSocket(connectionUrl.value);
+            socket.onopen = function (event) {
+                updateState();
+                commsLog.innerHTML += '<tr>' +
+                    '<td colspan="3" class="commslog-data">Connection opened</td>' +
+                '</tr>';
+            };
+            socket.onclose = function (event) {
+                updateState();
+                commsLog.innerHTML += '<tr>' +
+                    '<td colspan="3" class="commslog-data">Connection closed. Code: ' + htmlEscape(event.code) + '. Reason: ' + htmlEscape(event.reason) + '</td>' +
+                '</tr>';
+            };
+            socket.onerror = updateState;
+            socket.onmessage = function (event) {
+                commsLog.innerHTML += '<tr>' +
+                    '<td class="commslog-server">Server</td>' +
+                    '<td class="commslog-client">Client</td>' +
+                    '<td class="commslog-data">' + htmlEscape(event.data) + '</td></tr>';
+            };
+        };
+
+        function htmlEscape(str) {
+            return str.toString()
+                .replace(/&/g, '&amp;')
+                .replace(/"/g, '&quot;')
+                .replace(/'/g, '&#39;')
+                .replace(/</g, '&lt;')
+                .replace(/>/g, '&gt;');
+        }
+    </script>
+</body>
+</html>
@@ -0,0 +1,176 @@
+<!DOCTYPE html>
+<html>
+<head>
+    <meta charset="utf-8" />
+    <title></title>
+    <style>
+        table {
+            border: 0
+        }
+
+        .commslog-data {
+            font-family: Consolas, Courier New, Courier, monospace;
+        }
+
+        .commslog-server {
+            background-color: red;
+            color: white
+        }
+
+        .commslog-client {
+            background-color: green;
+            color: white
+        }
+    </style>
+</head>
+<body>
+    <h1>DataStreams via WebSocket</h1>
+    <p id="stateLabel">Ready to connect...</p>
+    <div>
+        <label for="connectionUrl">WebSocket Server URL:</label>
+        <input id="connectionUrl" />
+        <button id="connectButton" type="submit">Connect</button>
+    </div>
+    <p></p>
+    <div>
+        <label for="sendMessage">Query:</label>
+        <input id="sendMessage" disabled />
+        <button id="sendButton" type="submit" disabled>Send</button>
+        <button id="closeButton" disabled>Close Socket</button>
+    </div>
+
+    <h2>Communication Log</h2>
+    <table style="width: 800px">
+        <thead>
+            <tr>
+                <td style="width: 100px">From</td>
+                <td style="width: 100px">To</td>
+                <td>Data</td>
+            </tr>
+        </thead>
+        <tbody id="commsLog">
+        </tbody>
+    </table>
+
+    <script>
+        var connectionUrl = document.getElementById("connectionUrl");
+        var connectButton = document.getElementById("connectButton");
+        var stateLabel = document.getElementById("stateLabel");
+        var sendMessage = document.getElementById("sendMessage");
+        var sendButton = document.getElementById("sendButton");
+        var commsLog = document.getElementById("commsLog");
+        var closeButton = document.getElementById("closeButton");
+        var socket;
+
+        var scheme = document.location.protocol === "https:" ? "wss" : "ws";
+        var port = document.location.port ? (":" + document.location.port) : "";
+
+        connectionUrl.value = scheme + "://" + document.location.hostname + port + "/ws" ;
+
+        function updateState() {
+            function disable() {
+                sendMessage.disabled = true;
+                sendButton.disabled = true;
+                closeButton.disabled = true;
+            }
+            function enable() {
+                sendMessage.disabled = false;
+                sendButton.disabled = false;
+                closeButton.disabled = false;
+            }
+
+            connectionUrl.disabled = true;
+            connectButton.disabled = true;
+
+            if (!socket) {
+                disable();
+            } else {
+                switch (socket.readyState) {
+                    case WebSocket.CLOSED:
+                        stateLabel.innerHTML = "Closed";
+                        disable();
+                        connectionUrl.disabled = false;
+                        connectButton.disabled = false;
+                        break;
+                    case WebSocket.CLOSING:
+                        stateLabel.innerHTML = "Closing...";
+                        disable();
+                        break;
+                    case WebSocket.CONNECTING:
+                        stateLabel.innerHTML = "Connecting...";
+                        disable();
+                        break;
+                    case WebSocket.OPEN:
+                        stateLabel.innerHTML = "Open";
+                        enable();
+                        break;
+                    default:
+                        stateLabel.innerHTML = "Unknown WebSocket State: " + htmlEscape(socket.readyState);
+                        disable();
+                        break;
+                }
+            }
+        }
+
+        closeButton.onclick = function () {
+            if (!socket || socket.readyState !== WebSocket.OPEN) {
+                alert("socket not connected");
+            }
+            socket.close(1000, "Closing from client");
+        };
+
+        sendButton.onclick = function () {
+            if (!socket || socket.readyState !== WebSocket.OPEN) {
+                alert("socket not connected");
+            }
+            const rpcReq = {
+              jsonrpc: "2.0",
+              method: "subscribe",
+              params: {
+                query: sendMessage.value,
+              },
+              id: 1,
+            };
+            const rpcReqStr = JSON.stringify(rpcReq);
+            socket.send(rpcReqStr);
+            commsLog.innerHTML += '<tr>' +
+                '<td class="commslog-client">Client</td>' +
+                '<td class="commslog-server">Server</td>' +
+                '<td class="commslog-data">' + htmlEscape(rpcReqStr) + '</td></tr>';
+        };
+
+        connectButton.onclick = function() {
+            stateLabel.innerHTML = "Connecting";
+            socket = new WebSocket(connectionUrl.value);
+            socket.onopen = function (event) {
+                updateState();
+                commsLog.innerHTML += '<tr>' +
+                    '<td colspan="3" class="commslog-data">Connection opened</td>' +
+                '</tr>';
+            };
+            socket.onclose = function (event) {
+                updateState();
+                commsLog.innerHTML += '<tr>' +
+                    '<td colspan="3" class="commslog-data">Connection closed. Code: ' + htmlEscape(event.code) + '. Reason: ' + htmlEscape(event.reason) + '</td>' +
+                '</tr>';
+            };
+            socket.onerror = updateState;
+            socket.onmessage = function (event) {
+                commsLog.innerHTML += '<tr>' +
+                    '<td class="commslog-server">Server</td>' +
+                    '<td class="commslog-client">Client</td>' +
+                    '<td class="commslog-data">' + htmlEscape(event.data) + '</td></tr>';
+            };
+        };
+
+        function htmlEscape(str) {
+            return str.toString()
+                .replace(/&/g, '&amp;')
+                .replace(/"/g, '&quot;')
+                .replace(/'/g, '&#39;')
+                .replace(/</g, '&lt;')
+                .replace(/>/g, '&gt;');
+        }
+    </script>
+</body>
+</html>
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <TargetFramework>net7.0</TargetFramework>
+    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
  </PropertyGroup>
@@ -0,0 +1,34 @@
+// namespace ColumnMapTests;
+
+// using Bts.ExpressionCore;
+// using Bts.ColumnMapExpressions;
+// using Bts.DataStreamExpressions;
+
+// using DataStream.Console;
+
+// public class TempColumnMapTests {
+//    public static void Run(ExpressionTest test) {
+//       Run(test.Expression, test.ExpectedResults);
+//    }
+
+//    public static void Run(string expression, IDictionary<string, string> expectedResults) {
+//       var interpreter = ColumnMapExpression.Parse(expression);
+//       var env = new Dictionary<string, object?>();
+//       var result = interpreter.Evaluate(env);
+//       foreach (var (key, expected) in expectedResults) {
+//          if (!result.TryGetValue(key, out var actual)) {
+//             throw new Exception($"ExpectedResults did not contain key '{key}'");
+//          }
+//          if (!expected.Equals(actual?.ToString() ?? string.Empty)) {
+//             throw new Exception($"Expected '{key}' to be {expected} (actually {actual})");
+//          }
+//          ConsoleHelper.WriteLine($"PASS: {key}={actual}={expected}", ConsoleColor.Green);
+//       }
+//    }
+
+//    static void assert(bool condition) {
+//       if (!condition) {
+//          throw new Exception();
+//       }
+//    }
+// }
@@ -0,0 +1,29 @@
+namespace DataStream.Console;
+
+using Spark.Core.Disposables;
+
+using System;
+
+static class ConsoleHelper {
+   public static IDisposable PushBg(ConsoleColor value) {
+      var orig = Console.BackgroundColor;
+      Console.BackgroundColor = value;
+      return Disposable.Create(pop);
+      void pop() =>
+         Console.BackgroundColor = orig;
+   }
+
+   public static IDisposable PushFg(ConsoleColor value) {
+      var orig = Console.ForegroundColor;
+      Console.ForegroundColor = value;
+      return Disposable.Create(pop);
+      void pop() =>
+         Console.ForegroundColor = orig;
+   }
+   
+   public static void WriteLine(string text, ConsoleColor? fg = null, ConsoleColor? bg = null) {
+      using IDisposable? fgPush = fg is null ? null : PushFg(fg.Value);
+      using IDisposable? bgPush = bg is null ? null : PushBg(bg.Value);
+      Console.WriteLine(text);
+   }
+}
@@ -0,0 +1,15 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <ItemGroup>
+    <ProjectReference Include="..\ArithmeticExpressions\ArithmeticExpressions.csproj" />
+    <ProjectReference Include="..\DataStreamExpressions\DataStreamExpressions.csproj" />
+  </ItemGroup>
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net8.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+</Project>
@@ -0,0 +1,15 @@
+namespace DataStream.Console;
+
+using Spark.Core;
+
+using System.Text;
+
+public class ExpressionTest {
+   public ExpressionTest(string expression, IDictionary<string, string> expectedResults) {
+      Expression = expression;
+      ExpectedResults = expectedResults;
+   }
+
+   public string Expression { get; }
+   public IDictionary<string, string> ExpectedResults { get; }
+}
@@ -0,0 +1,25 @@
+namespace DataStream.Console;
+
+using Spark.Core;
+
+using System.Text;
+
+public class ExpressionTestParser {
+   public static ExpressionTest Parse(IEnumerable<string> lines) {
+      var expressionLines = lines.SkipWhile(l => l != "BEGIN_TEST").Skip(1);
+      var expressionBd = new StringBuilder();
+      Dictionary<string, string>? expectedResults = null;
+      foreach (var l in expressionLines) {
+         const string END_SENTINEL = "END_TEST";
+         if (l.StartsWith(END_SENTINEL)) {
+            expectedResults = l[END_SENTINEL.Length..].Trim()
+               .Split(" ")
+               .Select(kv => kv.Split(":"))
+               .ToDictionary(kv => kv[0], kv => kv[1]);
+            break;
+         }
+         expressionBd.AppendLine(l);
+      }
+      return new(expressionBd.ToString(), expectedResults ?? new(0));
+   }
+}
@@ -0,0 +1,32 @@
+// See https://aka.ms/new-console-template for more information
+using DataStream.Console;
+
+using ExpressionTests;
+
+using Spark.Core.Collections;
+
+using System.Reactive.Disposables;
+
+
+try {
+   foreach (var arg in args) {
+      if (File.Exists(arg)) {
+         using var fs = File.OpenRead(arg) ??
+            throw new Exception("File.OpenRead returned null. LIES!");
+         var expressionLines = new StringStreamEnumerable(new(fs));
+         var test = ExpressionTestParser.Parse(expressionLines);
+         TempExpressionTests.Run(test);
+      }
+   }
+   if (args.Length == 0) {
+      TempExpressionTests.Run("rv = 3 + 4", new Dictionary<string, string> { { "rv", "7" } });
+   }
+} catch (Exception ex) {
+   using (ConsoleHelper.PushBg(ConsoleColor.Red)) {
+      Console.WriteLine($"Failed! {ex}");
+   }
+   return;
+}
+
+using var _ = ConsoleHelper.PushFg(ConsoleColor.Green);
+Console.WriteLine($"Passed!");
@@ -0,0 +1,32 @@
+namespace ExpressionTests;
+
+using Bts.ArithmeticExpressions;
+
+using DataStream.Console;
+
+public class TempExpressionTests {
+   public static void Run(ExpressionTest test) {
+      Run(test.Expression, test.ExpectedResults);
+   }
+
+   public static void Run(string expression, IDictionary<string, string> expectedResults) {
+      var prg = ArithmeticProgramParser.Parse(expression);
+      var env = new Dictionary<string, object?>();
+      var result = prg.Evaluate(env);
+      foreach (var (key, expected) in expectedResults) {
+         if (!result.TryGetValue(key, out var actual)) {
+            throw new Exception($"ExpectedResults did not contain key '{key}'");
+         }
+         if (!expected.Equals(actual?.ToString() ?? string.Empty)) {
+            throw new Exception($"Expected '{key}' to be {expected} (actually {actual})");
+         }
+         ConsoleHelper.WriteLine($"PASS: {key}={actual}={expected}", ConsoleColor.Green);
+      }
+   }
+
+   static void assert(bool condition) {
+      if (!condition) {
+         throw new Exception();
+      }
+   }
+}
@@ -0,0 +1,6 @@
+BEGIN_TEST
+x=3
+y=4
+xySum= x+y
+constSum = 3 + 4
+END_TEST constSum:7 xySum:7
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <TargetFramework>net7.0</TargetFramework>
+    <TargetFramework>net8.0</TargetFramework>
    <Nullable>enable</Nullable>
  </PropertyGroup>

@@ -30,8 +30,8 @@ namespace BCG
           {
              using (FileStream fs = new FileStream(fileName, FileMode.Open, FileAccess.Read))
              {
-                  BinaryFormatter bf = new BinaryFormatter();
 #pragma warning disable SYSLIB0011
+                  BinaryFormatter bf = new BinaryFormatter();
                  BlueTable bt = (BlueTable)bf.Deserialize(fs);
 #pragma warning restore SYSLIB0011
                  mySetRealSource(new BlueRowCollectionStaticStream(bt));
@@ -1,95 +1,103 @@
-using System;
-
-namespace BCG
-{
-   /// <summary>
-   /// Resprsents information about a column schema, ie its ordinal, the type, the name and constant value or expression. where applicable
-   /// </summary>
-   [Serializable]
-   public class ColumnInfo : IComparable
-   {
-      public static bool isAllowedType (Type t)
-      {
-         return 
-            t.Equals (typeof (System.Int32)) ||
-            t.Equals (typeof (System.Int64)) ||
-            t.Equals (typeof (System.Double)) ||
-            t.Equals (typeof (System.String)) ||
-            t.Equals (typeof (System.DateTime)) ? true :
-            false;
-      }
-
-      //the default column info does not cach values in a lookup table, therefore it is in group 0
-      public ColumnInfo(string colName, int colNo, Type colType)
-         : this(colName, colNo, colType, 0)
-      {
-
-
-      }
-
-      public ColumnInfo(string colName, int colNo, Type colType, short colGroup)
-      {
-         myColName = colName;
-         myColNo = colNo;
-         myColType = colType;
-         myLookupTableGroup = colGroup;
-         myColValue = null;
-      }
-
-
-      public ColumnInfo (string colName, int colNo, object colValue)
-         : this (colName, colNo, colValue.GetType())  {myColValue = colValue == null ? DBNull.Value : colValue;}
-
-      public ColumnInfo (string colName, int colNo, Type colType, object colValue)
-         : this (colName, colNo, colType) 
-      {
-
-         myColValue = colValue == null ? DBNull.Value : colValue;
-      
-      }
-
-
-      public int CompareTo (object obj)
-      {
-         int otherColNo = ((ColumnInfo)(obj)).colNo;
-         if (myColNo < otherColNo)
-            return -1;
-         else if (myColNo > otherColNo)
-            return 1;
-         return 0;
-      }
-
-      //public bool isCalculated{get{return myExpression != null;}}
-
-      public bool isConstant {get{return myColValue != null;}}
-
-      private int myColNo;
-      private string myColName;
-      private Type myColType;
-      //private string myExpression;
-      private object myColValue;
-
-      //no table will ever have more than 65,536 columns
-      //if it does, we are in serious trouble
-      private short myLookupTableGroup;
-
-      public int colNo{get{return myColNo;}}
-      public string colName {get{return myColName;}}
-      public Type colType {get{return myColType;}}
-
-
-      //this value corresponds to how the data in the column is serialized. 
-      //if the group number is 0, the values are not stored in a table
-      //if the vlaue is non-zero, all columns with the the same lookup table group are stored in the same lookup table together
-      //are stored together in a lookup table
-
-      
-      public short lookupTableGroup
-      {
-         get { return myLookupTableGroup; }
-         set { myLookupTableGroup = value; }
-      }
-      //public string expression {get{return myExpression;}}
-      public object colValue {get{return myColValue;}}
-   }
-}
+using System;
+using System.Text.Json.Serialization;
+
+namespace BCG
+{
+   /// <summary>
+   /// Resprsents information about a column schema, ie its ordinal, the type, the name and constant value or expression. where applicable
+   /// </summary>
+   [Serializable]
+   public class ColumnInfo : IComparable
+   {
+      public static bool isAllowedType (Type t)
+      {
+         return 
+            t.Equals (typeof (System.Int32)) ||
+            t.Equals (typeof (System.Int64)) ||
+            t.Equals (typeof (System.Double)) ||
+            t.Equals (typeof (System.String)) ||
+            t.Equals (typeof (System.DateTime)) ? true :
+            false;
+      }
+
+      //the default column info does not cach values in a lookup table, therefore it is in group 0
+      public ColumnInfo(string colName, int colNo, Type colType)
+         : this(colName, colNo, colType, 0)
+      {
+
+
+      }
+
+      public ColumnInfo(string colName, int colNo, Type colType, short colGroup)
+      {
+         myColName = colName;
+         myColNo = colNo;
+         myColType = colType;
+         myLookupTableGroup = colGroup;
+         myColValue = null;
+      }
+
+
+      public ColumnInfo (string colName, int colNo, object colValue)
+         : this (colName, colNo, colValue.GetType())  {myColValue = colValue == null ? DBNull.Value : colValue;}
+
+      public ColumnInfo (string colName, int colNo, Type colType, object colValue)
+         : this (colName, colNo, colType) 
+      {
+
+         myColValue = colValue == null ? DBNull.Value : colValue;
+      
+      }
+
+
+      public int CompareTo (object obj)
+      {
+         int otherColNo = ((ColumnInfo)(obj)).colNo;
+         if (myColNo < otherColNo)
+            return -1;
+         else if (myColNo > otherColNo)
+            return 1;
+         return 0;
+      }
+
+      //public bool isCalculated{get{return myExpression != null;}}
+
+      public bool isConstant {get{return myColValue != null;}}
+
+      private int myColNo;
+      private string myColName;
+      private Type myColType;
+      //private string myExpression;
+      private object myColValue;
+
+      //no table will ever have more than 65,536 columns
+      //if it does, we are in serious trouble
+      private short myLookupTableGroup;
+
+      [JsonPropertyName("index")]
+      public int colNo{get{return myColNo;}}
+
+      [JsonPropertyName("name")]
+      public string colName {get{return myColName;}}
+
+      [JsonPropertyName("type")]
+      public Type colType {get{return myColType;}}
+
+
+      //this value corresponds to how the data in the column is serialized. 
+      //if the group number is 0, the values are not stored in a table
+      //if the vlaue is non-zero, all columns with the the same lookup table group are stored in the same lookup table together
+      //are stored together in a lookup table
+
+      
+      [JsonIgnore]
+      public short lookupTableGroup
+      {
+         get { return myLookupTableGroup; }
+         set { myLookupTableGroup = value; }
+      }
+      //public string expression {get{return myExpression;}}
+      [JsonIgnore]
+      public object colValue {get{return myColValue;}}
+   }
+}
@@ -0,0 +1,63 @@
+// #nullable enable
+// using BCG;
+
+// using Spark.Core.Json;
+
+// using System;
+// using System.Text.Json;
+// using System.Text.Json.Nodes;
+// using System.Text.Json.Serialization;
+
+// namespace Bts.DataStreamCore;
+
+// class BlueRowJsonConverter : JsonConverter<IBlueRow> {
+//    public override IBlueRow Read(
+//       ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) {
+//       var row = JsonSerializer.Deserialize<JsonObject>(ref reader, options)!;
+//       var schemaj = row["schema"]!.GetValue<JsonObject>() ??
+//          throw new InvalidOperationException("Failed to read schema");
+//       // var columns = schemaj["columns"]
+//       var mbr = new MutableBlueRow(schema);
+//       int colNum = 0;
+//       switch (reader.TokenType) {
+//          case JsonTokenType.StartArray:
+//             while (reader.Read() && reader.TokenType != JsonTokenType.EndArray) {
+//                mbr.setValue(colNum++, readCellValue(ref reader));
+//                colNum++;
+//             }
+//             break;
+//          case JsonTokenType.StartObject:
+//             while (reader.Read() && reader.TokenType != JsonTokenType.EndObject) {
+//                if (reader.TokenType != JsonTokenType.PropertyName) {
+//                   throw new InvalidOperationException(
+//                      "Expected 'PropertyName' immediately after 'StartObject' token in row json");
+//                }
+//                var colName = reader.GetString();
+//                if (!reader.Read()) {
+//                   throw new InvalidOperationException(
+//                      "Failed to read token immediately after 'PropertyName' token in row json");
+//                }
+//                mbr.setValue(colName, readCellValue(ref reader));
+//             }
+//             break;
+//       }
+//       if (mbr.Count != schema.columnInfo.Length) {
+//          throw new InvalidOperationException(
+//             $"Row json array length '{mbr.Count}' does not match the schema's " +
+//             $"column length '{schema.columnInfo.Length}'.");
+//       }
+//       return mbr;
+
+//       static object? readCellValue(ref Utf8JsonReader reader) =>
+//          reader.TokenType switch {
+//             JsonTokenType.String => reader.GetString(),
+//             JsonTokenType.Number => reader.GetDouble(),
+//             JsonTokenType.True or JsonTokenType.False => reader.GetBoolean(),
+//             _ => throw new InvalidOperationException("Failed to read cell value"),
+//          };
+//    }
+
+//    public override void Write(Utf8JsonWriter writer, IBlueRow value, JsonSerializerOptions options) {
+//       throw new NotImplementedException();
+//    }
+// }
@@ -0,0 +1,186 @@
+// #nullable enable
+// using BCG;
+
+// using Spark.Core.Json;
+
+// using System;
+// using System.Collections.Generic;
+// using System.Text.Json;
+// using System.Text.Json.Nodes;
+// using System.Text.Json.Serialization;
+
+// namespace Bts.DataStreamCore;
+
+// public class DataStreamUpdateJsonConverter : JsonConverter<DataStreamUpdate> {
+//    /// <summary>
+//    /// Whether or not to include a field for current rows (i.e., snapshot of current state). By
+//    /// default, only incremental changes are included ('added' and 'deleted').
+//    /// </summary>
+//    public bool IncludeCurrent { get; set; }
+
+//    /// <summary>
+//    /// Whether to treat rows as objects (name/value pairs), or arrays (list of values). Arrays
+//    /// result in smaller payloads, but aren't as easy to read as name/value pairs (for mere humans).
+//    /// </summary>
+//    public bool UseObjectRows { get; set; }
+
+//    public override DataStreamUpdate? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) {
+//       var dpj = JsonSerializer.Deserialize<JsonObject>(ref reader);
+//       if (dpj is null) {
+//          return null;
+//       }
+//       var tableName = dpj["tableName"]!.AsString();
+//       var columns = dpj["columns"]!.AsArray()
+//          .Select((c, i) => {
+//             var dataType = c!["type"]!.AsEnum<DataStreamDataType>();
+//             return (info: new ColumnInfo(c["name"]!.AsString(), i, dataType.ToType()), dataType);
+//          })
+//          .ToArray();
+//       var schema = new BlueTableSchema(columns.Cut1().ToArray(), tableName);
+//       return new() {
+//          TableName = tableName,
+//          AddedRows = ToRows(dpj.GetArrayOrEmpty("added")).ToList(),
+//          DeletedRows = ToRows(dpj.GetArrayOrEmpty("deleted")).ToList(),
+//          Columns = [.. schema.columnInfo]
+//       };
+
+//       IEnumerable<IBlueRow> ToRows(JsonArray rows) {
+//          foreach (var row in rows) {
+//             var mbr = new MutableBlueRow(schema);
+//             switch (row!.GetValueKind()) {
+//                case JsonValueKind.Object:
+//                   var rowObj = row.AsObject();
+//                   for (int i = 0; i < columns.Length; i++) {
+//                      SetCellValue(mbr, i, rowObj[columns[i].info.colName]!, columns[i].dataType);
+//                   }
+//                   break;
+//                case JsonValueKind.Array:
+//                   var rowArr = row.AsArray();
+//                   for (int i = 0; i < columns.Length; i++) {
+//                      SetCellValue(mbr, i, rowArr[i]!, columns[i].dataType);
+//                   }
+//                   break;
+//             }
+//             yield return mbr;
+//          }
+//       }
+
+//       static void SetCellValue(MutableBlueRow mbr, int i, JsonNode cell, DataStreamDataType dataType) {
+//          switch (dataType) {
+//             case DataStreamDataType.String:
+//                mbr.setValue(i, cell.GetValue<string>());
+//                break;
+//             case DataStreamDataType.Int32:
+//                mbr.setValue(i, cell.GetValue<int>());
+//                break;
+//             case DataStreamDataType.Int64:
+//                mbr.setValue(i, cell.GetValue<long>());
+//                break;
+//             case DataStreamDataType.Single:
+//                mbr.setValue(i, cell.GetValue<float>());
+//                break;
+//             case DataStreamDataType.Double:
+//                mbr.setValue(i, cell.GetValue<double>());
+//                break;
+//          }
+//       }
+//    }
+
+//    public override void Write(Utf8JsonWriter writer, DataStreamUpdate value, JsonSerializerOptions options) {
+//       using var _ = writer.WriteObject();
+//       writer.WriteString("tableName", value.TableName);
+//       using (writer.WriteArray("columns")) {
+//          foreach (var col in value.Columns.empty_if_null()) {
+//             using (writer.WriteObject()) {
+//                writer.WriteString("name", col.colName);
+//                writer.WriteNumber("index", col.colNo);
+//                // TODO: use language agnostic type names (e.g., 'string', 'integer', 'float', 'date' etc.)
+//                writer.WriteString("type", col.colType.ToString());
+//             }
+//          }
+//       }
+
+//       var cellWriters = value.Columns!.Select(getCellWriter).ToList();
+//       using (writer.WriteArray("added")) {
+//          foreach (IBlueRow row in value.AddedRows!) {
+//             using (writer.WriteArray()) {
+//                for (int i = 0; i < cellWriters.Count; i++) {
+//                   cellWriters[i](row[i]);
+//                }
+//             }
+//          }
+//       }
+//       using (writer.WriteArray("deleted")) {
+//          foreach (IBlueRow row in value.DeletedRows!) {
+//             using (writer.WriteArray()) {
+//                for (int i = 0; i < cellWriters.Count; i++) {
+//                   cellWriters[i](row[i]);
+//                }
+//             }
+//          }
+//       }
+//       if (IncludeCurrent) {
+//          using (writer.WriteArray("current")) {
+//             foreach (IBlueRow row in value.newRows) {
+//                using (writer.WriteArray()) {
+//                   for (int i = 0; i < cellWriters.Count; i++) {
+//                      cellWriters[i](row[i]);
+//                   }
+//                }
+//             }
+//          }
+//       }
+
+//       Action<object?> getCellWriter(ColumnInfo colInfo) {
+//          var writeVal = getCellValueWriter(colInfo);
+//          if (!UseObjectRows) {
+//             return writeVal;
+//          }
+//          return v => {
+//             writer.WritePropertyName(colInfo.colName);
+//             writeVal(v);
+//          };
+//       }
+
+//       Action<object?> getCellValueWriter(ColumnInfo colInfo) {
+//          var t = colInfo.colType;
+//          if (t == typeof(int)) {
+//             return nullCheckWrapper(v => writer.WriteNumberValue((int)v));
+//          }
+//          if (t == typeof(uint)) {
+//             return nullCheckWrapper(v => writer.WriteNumberValue((uint)v));
+//          }
+//          if (t == typeof(long)) {
+//             return nullCheckWrapper(v => writer.WriteNumberValue((long)v));
+//          }
+//          if (t == typeof(ulong)) {
+//             return nullCheckWrapper(v => writer.WriteNumberValue((ulong)v));
+//          }
+//          if (t == typeof(short)) {
+//             return nullCheckWrapper(v => writer.WriteNumberValue((short)v));
+//          }
+//          if (t == typeof(ushort)) {
+//             return nullCheckWrapper(v => writer.WriteNumberValue((ushort)v));
+//          }
+//          if (t == typeof(double)) {
+//             return nullCheckWrapper(v => writer.WriteNumberValue((double)v));
+//          }
+//          if (t == typeof(float)) {
+//             return nullCheckWrapper(v => writer.WriteNumberValue((float)v));
+//          }
+//          if (t == typeof(bool)) {
+//             return nullCheckWrapper(v => writer.WriteBooleanValue((bool)v));
+//          }
+//          return nullCheckWrapper(v => writer.WriteStringValue(v?.ToString()));
+
+//          Action<object?> nullCheckWrapper(Action<object> writeNonNull) =>
+//             v => {
+//                if (v is null) {
+//                   writer.WriteNullValue();
+//                } else {
+//                   writeNonNull(v);
+//                }
+//             };
+//       }
+//    }
+// }
@@ -2,8 +2,9 @@

  <PropertyGroup>
    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
-    <TargetFramework>net7.0</TargetFramework>
+    <TargetFramework>net8.0</TargetFramework>
    <Nullable>disable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
  </PropertyGroup>

  <ItemGroup>
@@ -0,0 +1,11 @@
+namespace Bts.DataStreamCore;
+
+public enum DataStreamDataType {
+   Any,
+   Bytes,
+   String,
+   Int32,
+   Int64,
+   Single,
+   Double
+}
@@ -0,0 +1,42 @@
+#nullable enable
+
+using Bts.DataStreamCore;
+
+public static class DataStreamExtensions {
+   public static Type ToType(this DataStreamDataType dataType) =>
+      dataType switch {
+         DataStreamDataType.Any => typeof(object),
+         DataStreamDataType.String => typeof(string),
+         DataStreamDataType.Int32 => typeof(int),
+         DataStreamDataType.Int64 => typeof(long),
+         DataStreamDataType.Single => typeof(float),
+         DataStreamDataType.Double => typeof(double),
+         _ => throw new ArgumentException($"DataType '{dataType}' not supported.", nameof(dataType)),
+      };
+
+   public static DataStreamDataType FromType(Type type) {
+      if (type == typeof(object)) {
+         return DataStreamDataType.Any;
+      }
+      if (type == typeof(string)) {
+         return DataStreamDataType.String;
+      }
+      if (type == typeof(int)) {
+         return DataStreamDataType.Int32;
+      }
+      if (type == typeof(long)) {
+         return DataStreamDataType.Int64;
+      }
+      if (type == typeof(float)) {
+         return DataStreamDataType.Single;
+      }
+      if (type == typeof(double)) {
+         return DataStreamDataType.Double;
+      }
+      if (type == typeof(byte[])) {
+         return DataStreamDataType.Bytes;
+      }
+      throw new ArgumentException(
+         $"Unable to convert type '{type}' to {typeof(DataStreamDataType)}");
+   }
+}
@@ -0,0 +1,41 @@
+#nullable enable
+using BCG;
+
+using Spark.Core.Json;
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+namespace Bts.DataStreamCore;
+
+public class DataStreamUpdate {
+   [JsonPropertyName("tableName")]
+   public string? TableName { get; set; }
+
+   [JsonPropertyName("columns")]
+   public List<ColumnInfo>? Columns { get; set; }
+
+   [JsonPropertyName("addedRows")]
+   public List<IBlueRow>? AddedRows { get; set; }
+
+   [JsonPropertyName("deletedRows")]
+   public List<IBlueRow>? DeletedRows { get; set; }
+
+   public DataPacket ToDataPacket(BlueTable state, bool applyChanges = true) {
+      var schema = new BlueTableSchema([.. Columns!], TableName);
+      var addedRowsTable = new BlueTable(schema);
+      var deletedRowsTable = new BlueTable(schema);
+      if (applyChanges) {
+         state.deleteRows(deletedRowsTable);
+         state.addRows(addedRowsTable);
+      }
+      return new(state, addedRowsTable, deletedRowsTable);
+   }
+
+   public static string ToJson() {
+      return "";
+   }
+}
@@ -0,0 +1 @@
+using System.Linq;
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
-    <TargetFramework>net7.0</TargetFramework>
+    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
  </PropertyGroup>
Author	SHA1	Message	Date
John Colvin	b8bfc1058d	check point (bc i don't trust googledrive sync yet)	2024-02-09 23:06:26 -06:00
John Colvin	82278a4e94	dev: Add gitpod.yml for cloud workspaces	2024-02-09 23:06:26 -06:00
John Colvin	592ff2a7fe	externals: Change repo url to use read-only access token	2024-02-09 23:06:26 -06:00
John Colvin	d65dc1c8e3	build: Upgrade c# to net8.0 (DataStreamSuite)	2024-02-09 23:06:26 -06:00
John Colvin	464a153582	Add Bts.DataStream.WebsocketClient csproj	2024-02-09 23:06:26 -06:00
John Colvin	0610f9ac04	Add ExpressionTests csproj	2024-02-09 23:06:26 -06:00
John Colvin	91c2f6f033	Add TransportUtilities csproj	2024-02-09 23:06:26 -06:00
John Colvin	486c314f19	Add ExpressionTester csproj	2024-02-09 23:06:26 -06:00
John Colvin	1f0b8bd82d	Add DataStream.Console csproj	2024-02-09 23:06:26 -06:00
John Colvin	6aafb2c5f1	Add json facilities for DataStream websocket api	2024-02-09 23:06:19 -06:00
John Colvin	e4d20be735	refactor: Add handy WebsocketExtensions * IsSocketStateOpen * IsClosedProperly	2024-02-09 23:05:37 -06:00
John Colvin	ca028c58ed	feat: Add Bts.DataStream.Websocket/Server	2024-02-09 22:31:10 -06:00
John Colvin	162969d2f7	Add bones for Bts.DataStreamWebApp.Client	2024-02-09 22:30:30 -06:00
John Colvin	d798594836	Add bones for Bts.DataStreamWebApp.Server	2024-02-09 22:30:07 -06:00
John Colvin	9611462e2c	add util: StringStreamEnumerable	2024-02-09 22:30:07 -06:00
John Colvin	54986aeccd	docker: Clean up, reorganize Dockerfile	2024-02-09 22:30:07 -06:00
John Colvin	6d489ce246	docker: Use explicit release ubuntu:22.04 (not 'latest') We don't want the distro to automatically upgrade one day, when 'latest' suddenly points to a new release. Upgrades like that need to be done deliberately.	2024-02-09 22:30:07 -06:00
John Colvin	6217b190bb	docker: Install gdb (for c++ debugging) VS Code debugger depends on this	2024-02-09 22:30:07 -06:00
John Colvin	6a262719af	cleanup: Remove empty file (clearly added by accident)	2024-02-09 22:30:07 -06:00
John Colvin	cf0282f86d	Add more descriptive logs (via stdout) during initialization The C++ configure/initialize routines were particularly opaque	2024-02-09 22:27:58 -06:00
John Colvin	76be3074cd	fix: Fix improper 'OfType' linq cast Previously, animations were expressed through polymorphism (derived from the base class CellAnimation), but the new impl uses composition. This 'OfType' call was leftover from the polymorphism implementation, an attempt to linq cast elements of base class (CellAnimation) to BackgroundAnimation.	2024-02-09 22:27:58 -06:00
John Colvin	029d89386f	cleanup: Remove static service Locator class It relied on a static singleton 'ServiceContext'. Replace references with a properly scoped (i.e., non-static) ServiceContext	2024-02-09 22:27:58 -06:00
John Colvin	1544a6c3aa	refactor: Reduce code (simplify) in BTS.init Remove the 'args' parameter, since the cmd line can be obtained from Environment.GetCommandLine anyways	2024-02-09 22:27:58 -06:00
John Colvin	158dfdd606	Dispose connections in SimpleTCPServer.Dispose	2024-02-09 22:27:58 -06:00
John Colvin	aaa3ff114b	Polish PubSubServer cli	2024-02-09 22:27:57 -06:00
John Colvin	c98891f5e3	cleanup: Format PubSubServer.csproj	2024-02-09 22:23:08 -06:00
John Colvin	f31d8d011b	improvement: Polish DataStreamPublisher cli Using CommandLineParser nuget pkg	2024-02-09 22:22:50 -06:00
John Colvin	7df97373ef	editorconfig: Use standard c# capitalization rules * Use 'PascalCase' for public members (instead of 'snake_case') * Use 'camelCase' for private members (instead of '_snake_case')	2024-02-09 01:06:35 -06:00
John Colvin	df7f8fc48f	refactor: Change WebsocketServerConfig.port to uint32 It should always be a number between 1-65535	2024-02-09 01:06:35 -06:00
John Colvin	33d9e344ed	refactor: Add WebsocketGatewayObservable An IObservable wrapper for IWebsocketConnectionGateway	2024-02-09 01:06:35 -06:00
John Colvin	368dce37c6	refactor: Remove nullability constraint in GetValueOrDefault	2024-02-09 01:06:34 -06:00
John Colvin	e228b33c5a	refactor: Use events to avoid circular deps for ws sync/gateway	2024-02-09 01:06:34 -06:00
John Colvin	0116a0cfe5	refactor: Let caller specify nullability of T in json parse fn	2024-02-09 01:06:34 -06:00
John Colvin	6b7c115779	refactor: Get rid of redundant UniversalNumericComparer	2024-02-09 01:06:34 -06:00
John Colvin	85593f782d	refactor: Reorganize NumericComparer, support all types	2024-02-09 01:06:34 -06:00
John Colvin	cfa5c5a283	feat: Add JsonRpc types for websocket paradigm	2024-02-09 01:06:34 -06:00
John Colvin	29b901f546	docker: Remove kitware 'rc' (release candidate) apt repo No reason to keep it. Probably shouldn't have been added in the first place	2024-02-09 01:06:34 -06:00
John Colvin	42a1f37dcf	docker: Make clang-17 the default clang	2024-02-09 01:06:34 -06:00
John Colvin	f63e6e2ad7	docker: Install clang 17	2024-02-09 01:06:34 -06:00
John Colvin	d40b34e6d5	docker: Update default compiler versions Via update-alternatives	2024-02-09 01:06:34 -06:00
John Colvin	c1cf6230fa	docker: Upgrade dotnet 7 -> 8	2024-02-09 01:06:34 -06:00
John Colvin	1118e89cf4	cleanup: Remove leftover legacy files	2024-02-09 01:06:33 -06:00
John Colvin	6a536c1208	build: Upgrade c# to net8.0	2024-02-09 01:06:33 -06:00
				`@@ -0,0 +1 @@`
				<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" aria-hidden="true" role="img" class="iconify iconify--logos" width="31.88" height="32" preserveAspectRatio="xMidYMid meet" viewBox="0 0 256 257"><defs><linearGradient id="IconifyId1813088fe1fbc01fb466" x1="-.828%" x2="57.636%" y1="7.652%" y2="78.411%"><stop offset="0%" stop-color="#41D1FF"></stop><stop offset="100%" stop-color="#BD34FE"></stop></linearGradient><linearGradient id="IconifyId1813088fe1fbc01fb467" x1="43.376%" x2="50.316%" y1="2.242%" y2="89.03%"><stop offset="0%" stop-color="#FFEA83"></stop><stop offset="8.333%" stop-color="#FFDD35"></stop><stop offset="100%" stop-color="#FFA800"></stop></linearGradient></defs><path fill="url(#IconifyId1813088fe1fbc01fb466)" d="M255.153 37.938L134.897 252.976c-2.483 4.44-8.862 4.466-11.382.048L.875 37.958c-2.746-4.814 1.371-10.646 6.827-9.67l120.385 21.517a6.537 6.537 0 0 0 2.322-.004l117.867-21.483c5.438-.991 9.574 4.796 6.877 9.62Z"></path><path fill="url(#IconifyId1813088fe1fbc01fb467)" d="M185.432.063L96.44 17.501a3.268 3.268 0 0 0-2.634 3.014l-5.474 92.456a3.268 3.268 0 0 0 3.997 3.378l24.777-5.718c2.318-.535 4.413 1.507 3.936 3.838l-7.361 36.047c-.495 2.426 1.782 4.5 4.151 3.78l15.304-4.649c2.372-.72 4.652 1.36 4.15 3.788l-11.698 56.621c-.732 3.542 3.979 5.473 5.943 2.437l1.313-2.028l72.516-144.72c1.215-2.423-.88-5.186-3.54-4.672l-25.505 4.922c-2.396.462-4.435-1.77-3.759-4.114l16.646-57.705c.677-2.35-1.37-4.583-3.769-4.113Z"></path></svg>