build: Remove dependency on Newtonsoft.Json (in DataStreamCore)

Replace with System.Text.Json (official dotnet)

.devcontainer/devcontainer.json

@@ -1,5 +1,11 @@
 {
-  "dockerFile": "Dockerfile",
+  "build": {
+    "dockerfile": "../docker/images/bluetradingsystems/linux_dotnet8_dev/Dockerfile",
+    "options": [
+      "--secret",
+      "id=bitbucket_access_token,src=${localWorkspaceFolder}/.secrets/bitbucket_access_token.txt"
+    ]
+  },
   "hostRequirements": {
     "memory": "8gb",
     "storage": "10gb"

.devcontainer/emumba/install_emumba-deribit-connector.sh

@@ -1,4 +0,0 @@
-#!/usr/bin/env bash
-
-this_dir="$(realpath "$(dirname "$(readlink -f "${BASH_SOURCE[0]}")")")"
-sudo dpkg -i "$this_dir/emumba-deribit-connector-3.3.1-Linux.LICENSE_DISABLED.deb"

.vscode/launch.json

@@ -1,25 +1,6 @@
 {
   "version": "0.2.0",
   "configurations": [
-  {
-    "name": "(gdb) Attach",
-    "type": "cppdbg",
-    "request": "attach",
-    "program": "${workspaceFolder}/build/sparkd-debug-clang/src/app_sparkd/sparkd",
-    "MIMode": "gdb",
-    "setupCommands": [
-        {
-            "description": "Enable pretty-printing for gdb",
-            "text": "-enable-pretty-printing",
-            "ignoreFailures": true
-        },
-        {
-            "description": "Set Disassembly Flavor to Intel",
-            "text": "-gdb-set disassembly-flavor intel",
-            "ignoreFailures": true
-        }
-    ]
-  },
     {
       "name": "Python Debugger: Current File with Arguments",
       "type": "debugpy",
@@ -161,17 +142,30 @@
       "name": "DataStreamQueryAnalyzer",
       "type": "coreclr",
       "request": "launch",
-      //"preLaunchTask": "Build DataStreamQueryAnalyzer (Debug)",
+      "preLaunchTask": "Build DataStream (Debug)",
       "program": "",
       "windows": {
         "program": "DataStreamQueryAnalyzer.exe",
-        "args": [
-          "--config",
-          "${workspaceFolder}/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json"
-        ],
         "cwd": "build/Debug/bin/net8.0-windows",
         "env": {
-          "PYTHON_EXE": "${workspaceFolder}/.venv/Scripts/python.exe"
+          "PYTHON_EXE": "${workspaceFolder}/.venv/Scripts/python.exe",
+          "DOTNET_ENVIRONMENT": "${input:dataStreamQueryAnalyzerEnvironment}",
+        },
+      },
+      "stopAtEntry": false,
+      "console": "integratedTerminal",
+    },
+    {
+      "name": "DataStreamQueryAnalyzer (Skip Build)",
+      "type": "coreclr",
+      "request": "launch",
+      "program": "",
+      "windows": {
+        "program": "DataStreamQueryAnalyzer.exe",
+        "cwd": "build/Debug/bin/net8.0-windows",
+        "env": {
+          "PYTHON_EXE": "${workspaceFolder}/.venv/Scripts/python.exe",
+          "DOTNET_ENVIRONMENT": "${input:dataStreamQueryAnalyzerEnvironment}",
         },
       },
       "stopAtEntry": false,
@@ -390,6 +384,25 @@
         }
       ]
     },
+    {
+      "name": "Sparkd Attach",
+      "type": "cppdbg",
+      "request": "attach",
+      "program": "${workspaceFolder}/build/sparkd-debug-clang/src/app_sparkd/sparkd",
+      "MIMode": "gdb",
+      "setupCommands": [
+        {
+          "description": "Enable pretty-printing for gdb",
+          "text": "-enable-pretty-printing",
+          "ignoreFailures": true
+        },
+        {
+          "description": "Set Disassembly Flavor to Intel",
+          "text": "-gdb-set disassembly-flavor intel",
+          "ignoreFailures": true
+        }
+      ]
+    },
   ],
   "inputs": [
     {
@@ -415,6 +428,16 @@
       ],
       "default": "MarketDataPublisher",
     },
+    {
+      "id": "dataStreamQueryAnalyzerEnvironment",
+      "type": "pickString",
+      "description": "DataStreamQueryAnalyzer Configuration",
+      "options": [
+        "Development",
+        "Local",
+      ],
+      "default": "Development",
+    },
     {
       "id": "dataStreamPublisherEnvironment",
       "type": "pickString",

.vscode/python.env

@@ -1 +1 @@
-PYTHONPATH=/workspaces/spark/build/sparkd-debug-clang/src/lib_python_interop;/workspaces/spark/src/lib_python_interop/python_interop_plugins
+PYTHONPATH=/workspaces/spark/build/sparkd-debug-clang/src/lib_python_interop;/workspaces/spark/src/lib_python_interop/python_interop_plugins;/workspaces/spark/tools/py

.vscode/tasks.json

@@ -541,12 +541,11 @@
       "command": "",
       "windows": {
         "command": "DataStreamQueryAnalyzer.exe",
-        "args": [
-          "--config",
-          "${workspaceFolder}/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json",
-        ],
         "options": {
-          "cwd": "${workspaceFolder}/build/${input:buildConfiguration}/bin/net8.0-windows"
+          "cwd": "${workspaceFolder}/build/${input:buildConfiguration}/bin/net8.0-windows",
+          "env": {
+            "DOTNET_ENVIRONMENT": "${input:dataStreamQueryAnalyzerEnvironment}",
+          },
         },
       },
       "icon": {
@@ -561,12 +560,11 @@
       "command": "",
       "windows": {
         "command": "DataStreamQueryAnalyzer.exe",
-        "args": [
-          "--config",
-          "${workspaceFolder}/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json",
-        ],
         "options": {
-          "cwd": "${workspaceFolder}/build/Debug/bin/net8.0-windows"
+          "cwd": "${workspaceFolder}/build/Debug/bin/net8.0-windows",
+          "env": {
+            "DOTNET_ENVIRONMENT": "${input:dataStreamQueryAnalyzerEnvironment}",
+          },
         },
       },
       "icon": {
@@ -581,12 +579,11 @@
       "command": "",
       "windows": {
         "command": "DataStreamQueryAnalyzer.exe",
-        "args": [
-          "--config",
-          "${workspaceFolder}/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json",
-        ],
         "options": {
-          "cwd": "${workspaceFolder}/build/Release/bin/net8.0-windows"
+          "cwd": "${workspaceFolder}/build/Release/bin/net8.0-windows",
+          "env": {
+            "DOTNET_ENVIRONMENT": "${input:dataStreamQueryAnalyzerEnvironment}",
+          },
         },
       },
       "icon": {
@@ -1048,6 +1045,16 @@
       ],
       "default": "Debug",
     },
+    {
+      "id": "dataStreamQueryAnalyzerEnvironment",
+      "type": "pickString",
+      "description": "DataStreamQueryAnalyzer Configuration",
+      "options": [
+        "Development",
+        "Local",
+      ],
+      "default": "Development",
+    },
     {
       "id": "sparkdCmakePreset",
       "type": "pickString",

Bts.Spark.Sdk.sln

@@ -11,9 +11,7 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Spark.Core", "src\cs\Bt
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Spark.Api", "src\cs\Bts.Spark.Api\Bts.Spark.Api.csproj", "{8949AF88-F1D3-4BC4-827F-B088B555F7B9}"
 EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "DataStreamSuite", "DataStreamSuite", "{60E0D592-FFB3-0D0A-C4AA-E88919B48F9D}"
-EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Extensions.Hosting", "src\DataStreamSuite\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj", "{37E4E890-D025-4729-90F9-6F95862F47BC}"
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Extensions.Hosting", "src\cs\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj", "{37E4E890-D025-4729-90F9-6F95862F47BC}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Spark.Sdk", "src\cs\Bts.Spark.Sdk\Bts.Spark.Sdk.csproj", "{A5FF3FAF-43CB-496B-AF4B-F61EFD645579}"
 EndProject
@@ -97,8 +95,7 @@ Global
 		{C5A3B5EA-BE9F-76E5-E9B8-8A22B289B295} = {827E0CD3-B72D-47B6-A68D-7590B98EB39B}
 		{E3BF9514-CB1A-4E07-A8DD-73300FD9726C} = {C5A3B5EA-BE9F-76E5-E9B8-8A22B289B295}
 		{8949AF88-F1D3-4BC4-827F-B088B555F7B9} = {C5A3B5EA-BE9F-76E5-E9B8-8A22B289B295}
-		{60E0D592-FFB3-0D0A-C4AA-E88919B48F9D} = {827E0CD3-B72D-47B6-A68D-7590B98EB39B}
-		{37E4E890-D025-4729-90F9-6F95862F47BC} = {60E0D592-FFB3-0D0A-C4AA-E88919B48F9D}
+		{37E4E890-D025-4729-90F9-6F95862F47BC} = {C5A3B5EA-BE9F-76E5-E9B8-8A22B289B295}
 		{A5FF3FAF-43CB-496B-AF4B-F61EFD645579} = {C5A3B5EA-BE9F-76E5-E9B8-8A22B289B295}
 		{FCFDBDBE-74B0-41C4-9011-C651B21056D4} = {C5A3B5EA-BE9F-76E5-E9B8-8A22B289B295}
 	EndGlobalSection

Spark.Api.Server.sln

@@ -43,7 +43,13 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.ColumnExpressions", "sr
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataStream", "src\DataStreamSuite\DataStream\DataStream.csproj", "{39AEA075-6457-46C0-8821-1158BE6EFA57}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Extensions.Hosting", "src\DataStreamSuite\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj", "{F3DC52D7-8C59-4EC0-86D9-0E5732EAFCF7}"
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Extensions.Hosting", "src\cs\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj", "{F3DC52D7-8C59-4EC0-86D9-0E5732EAFCF7}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "cs", "cs", "{385C4948-E080-56A6-E29E-8B0F4913175C}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Net", "src\cs\Bts.Net\Bts.Net.csproj", "{43FB297B-7A4D-4835-A878-E6DF345B580C}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.IO.Compression", "src\cs\Bts.IO.Compression\Bts.IO.Compression.csproj", "{BC277499-8215-4F9F-B9AB-07B6391110BF}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -517,6 +523,54 @@ Global
 		{F3DC52D7-8C59-4EC0-86D9-0E5732EAFCF7}.RelWithDebInfo|Any CPU.Build.0 = Debug|Any CPU
 		{F3DC52D7-8C59-4EC0-86D9-0E5732EAFCF7}.RelWithDebInfo|x86.ActiveCfg = Debug|Any CPU
 		{F3DC52D7-8C59-4EC0-86D9-0E5732EAFCF7}.RelWithDebInfo|x86.Build.0 = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Debug|x64.Build.0 = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Debug|x86.Build.0 = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.MinSizeRel|x64.ActiveCfg = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.MinSizeRel|x64.Build.0 = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.MinSizeRel|Any CPU.ActiveCfg = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.MinSizeRel|Any CPU.Build.0 = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.MinSizeRel|x86.ActiveCfg = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.MinSizeRel|x86.Build.0 = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Release|x64.ActiveCfg = Release|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Release|x64.Build.0 = Release|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Release|Any CPU.Build.0 = Release|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Release|x86.ActiveCfg = Release|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.Release|x86.Build.0 = Release|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.RelWithDebInfo|x64.ActiveCfg = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.RelWithDebInfo|x64.Build.0 = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.RelWithDebInfo|Any CPU.ActiveCfg = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.RelWithDebInfo|Any CPU.Build.0 = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.RelWithDebInfo|x86.ActiveCfg = Debug|Any CPU
+		{43FB297B-7A4D-4835-A878-E6DF345B580C}.RelWithDebInfo|x86.Build.0 = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Debug|x64.Build.0 = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Debug|x86.Build.0 = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.MinSizeRel|x64.ActiveCfg = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.MinSizeRel|x64.Build.0 = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.MinSizeRel|Any CPU.ActiveCfg = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.MinSizeRel|Any CPU.Build.0 = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.MinSizeRel|x86.ActiveCfg = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.MinSizeRel|x86.Build.0 = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Release|x64.ActiveCfg = Release|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Release|x64.Build.0 = Release|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Release|Any CPU.Build.0 = Release|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Release|x86.ActiveCfg = Release|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.Release|x86.Build.0 = Release|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.RelWithDebInfo|x64.ActiveCfg = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.RelWithDebInfo|x64.Build.0 = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.RelWithDebInfo|Any CPU.ActiveCfg = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.RelWithDebInfo|Any CPU.Build.0 = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.RelWithDebInfo|x86.ActiveCfg = Debug|Any CPU
+		{BC277499-8215-4F9F-B9AB-07B6391110BF}.RelWithDebInfo|x86.Build.0 = Debug|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -535,7 +589,9 @@ Global
 		{9F6D7DE8-185B-4C1D-B809-3314D092DF28} = {C09ED521-D2DF-40BC-A76F-EB3CB870B2F6}
 		{6BD3F6D8-B6FC-44E8-B8FE-06352F638B9B} = {C09ED521-D2DF-40BC-A76F-EB3CB870B2F6}
 		{39AEA075-6457-46C0-8821-1158BE6EFA57} = {C09ED521-D2DF-40BC-A76F-EB3CB870B2F6}
-		{F3DC52D7-8C59-4EC0-86D9-0E5732EAFCF7} = {C09ED521-D2DF-40BC-A76F-EB3CB870B2F6}
+		{385C4948-E080-56A6-E29E-8B0F4913175C} = {267F81EF-A8AB-49FD-9F56-D7EC3C1F7C7C}
+		{43FB297B-7A4D-4835-A878-E6DF345B580C} = {385C4948-E080-56A6-E29E-8B0F4913175C}
+		{BC277499-8215-4F9F-B9AB-07B6391110BF} = {385C4948-E080-56A6-E29E-8B0F4913175C}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {422D1620-0166-4C26-9A81-CA8A2EB1AB58}

docker/images/bluetradingsystems/linux_dotnet8_dev/Dockerfile

@@ -55,14 +55,16 @@ RUN \
    clangd-$clang_major
 
 # 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 \
-   && apt-get install -y cmake
+RUN set -eux \
+  && apt-get update \
+  && apt-get install -y --no-install-recommends ca-certificates curl gpg \
+  && curl -fsSL https://apt.kitware.com/keys/kitware-archive-latest.asc \
+    | gpg --dearmor -o /usr/share/keyrings/kitware-archive-keyring.gpg \
+  && echo "deb [signed-by=/usr/share/keyrings/kitware-archive-keyring.gpg] https://apt.kitware.com/ubuntu/ $(lsb_release -cs) main" \
+    > /etc/apt/sources.list.d/kitware.list \
+  && apt-get update \
+  && apt-get install -y --no-install-recommends cmake \
+  && rm -rf /var/lib/apt/lists/*
 
 # Dotnet
 RUN apt-get update && apt-get install -y dotnet8
@@ -80,9 +82,14 @@ RUN mkdir -p /etc/apt/keyrings \
 # Emumba (deps)
 RUN apt-get update && apt-get -y install rapidjson-dev
 # Emumba
-COPY emumba /emumba
-RUN dpkg -i /emumba/emumba-deribit-connector-3.3.1-Linux.LICENSE_DISABLED.deb \
-    && rm -r /emumba
+ARG EMUMBA_DERIBIT_CONNECTOR_DEB_URL="https://api.bitbucket.org/2.0/repositories/bluetradesys/spark/downloads/emumba-deribit-connector-3.3.1-Linux.LICENSE_DISABLED.deb"
+RUN --mount=type=secret,id=bitbucket_access_token \
+  set -eux \
+    && apt-get update \
+    && TOKEN="$(cat /run/secrets/bitbucket_access_token)" \
+    && curl -fL -H "Authorization: Bearer ${TOKEN}" $EMUMBA_DERIBIT_CONNECTOR_DEB_URL -o /tmp/emumba-deribit-connector.deb \
+    && apt-get install -y /tmp/emumba-deribit-connector.deb \
+    && rm -f /tmp/emumba-deribit-connector.deb
 
 # Pcap
 RUN apt-get update && apt-get install -y libpcap0.8-dev
@@ -90,6 +97,18 @@ RUN apt-get update && apt-get install -y libpcap0.8-dev
 # Databento (deps)
 RUN apt-get update && apt-get -y install libzstd-dev
 
+# Rakurai
+ARG RAKURAI_DEB_URL="https://api.bitbucket.org/2.0/repositories/bluetradesys/spark/downloads/rakurai-dex-connector_0.0.5_amd64.deb"
+RUN --mount=type=secret,id=bitbucket_access_token \
+  set -eux \
+    && apt-get update \
+    && TOKEN="$(cat /run/secrets/bitbucket_access_token)" \
+    && curl -fL -H "Authorization: Bearer ${TOKEN}" $RAKURAI_DEB_URL -o /tmp/rakurai.deb \
+    # Need --force-overwrite because rakuirai and emumba-deribit-connector both include the same
+    # spdlog headers.
+    && apt-get install -y -o Dpkg::Options::="--force-overwrite" /tmp/rakurai.deb \
+    && rm -f /tmp/rakuirai.deb
+
 # Docker in Docker
 RUN apt-get update && apt-get install -y \
     docker.io \

requirements.txt

@@ -18,10 +18,10 @@ MarkupSafe==3.0.2
 mccabe==0.6.1
 numpy==2.2.6
 platformdirs==2.3.0
-protobuf==4.21.12
+protobuf==6.33.4
 pybars3==0.9.7
-pybind11==2.13.6
-pybind11-stubgen==2.5.3
+pybind11==3.0.1
+pybind11-stubgen==2.5.5
 pybit==5.11.0
 pycodestyle==2.7.0
 pycparser==2.20

spark_client.sln

@@ -31,9 +31,13 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "src", "src", "{827E0CD3-B72
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "DataStreamSuite", "DataStreamSuite", "{60E0D592-FFB3-0D0A-C4AA-E88919B48F9D}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Net", "src\DataStreamSuite\Bts.Net\Bts.Net.csproj", "{5DE3B533-DCE6-4BF4-8F7E-08C440E177B4}"
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Net", "src\cs\Bts.Net\Bts.Net.csproj", "{5DE3B533-DCE6-4BF4-8F7E-08C440E177B4}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Extensions.Hosting", "src\DataStreamSuite\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj", "{90A48718-F6A7-4747-97EA-89632B451008}"
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Extensions.Hosting", "src\cs\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj", "{90A48718-F6A7-4747-97EA-89632B451008}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "cs", "cs", "{1A5B8C5D-BADC-457E-819F-0050C44FE4D3}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.IO.Compression", "src\cs\Bts.IO.Compression\Bts.IO.Compression.csproj", "{2A0FBDE0-D870-4567-85BD-F7793A5862C6}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -387,14 +391,38 @@ Global
 		{90A48718-F6A7-4747-97EA-89632B451008}.RelWithDebInfo|Any CPU.Build.0 = Debug|Any CPU
 		{90A48718-F6A7-4747-97EA-89632B451008}.RelWithDebInfo|x86.ActiveCfg = Debug|Any CPU
 		{90A48718-F6A7-4747-97EA-89632B451008}.RelWithDebInfo|x86.Build.0 = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Debug|x64.Build.0 = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Debug|x86.Build.0 = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.MinSizeRel|x64.ActiveCfg = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.MinSizeRel|x64.Build.0 = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.MinSizeRel|Any CPU.ActiveCfg = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.MinSizeRel|Any CPU.Build.0 = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.MinSizeRel|x86.ActiveCfg = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.MinSizeRel|x86.Build.0 = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Release|x64.ActiveCfg = Release|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Release|x64.Build.0 = Release|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Release|Any CPU.Build.0 = Release|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Release|x86.ActiveCfg = Release|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.Release|x86.Build.0 = Release|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.RelWithDebInfo|x64.ActiveCfg = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.RelWithDebInfo|x64.Build.0 = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.RelWithDebInfo|Any CPU.ActiveCfg = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.RelWithDebInfo|Any CPU.Build.0 = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.RelWithDebInfo|x86.ActiveCfg = Debug|Any CPU
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6}.RelWithDebInfo|x86.Build.0 = Debug|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
 	EndGlobalSection
 	GlobalSection(NestedProjects) = preSolution
 		{60E0D592-FFB3-0D0A-C4AA-E88919B48F9D} = {827E0CD3-B72D-47B6-A68D-7590B98EB39B}
-		{5DE3B533-DCE6-4BF4-8F7E-08C440E177B4} = {60E0D592-FFB3-0D0A-C4AA-E88919B48F9D}
-		{90A48718-F6A7-4747-97EA-89632B451008} = {60E0D592-FFB3-0D0A-C4AA-E88919B48F9D}
+		{1A5B8C5D-BADC-457E-819F-0050C44FE4D3} = {827E0CD3-B72D-47B6-A68D-7590B98EB39B}
+		{2A0FBDE0-D870-4567-85BD-F7793A5862C6} = {1A5B8C5D-BADC-457E-819F-0050C44FE4D3}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {422D1620-0166-4C26-9A81-CA8A2EB1AB58}

src/DataStreamSuite/DataStream/DataStreamExtensions.cs

@@ -23,16 +23,16 @@ public static class DataStreamExtensions {
    public static ViewStream AsViewStream<T>(this IObservable<IDataPacket<T>> source, string? name = null, ILogger? logger = null) where T : IBlueRow =>
       source as ViewStream ?? new ObservableViewStream(source.Select(dp => new DataPacket(dp.newRows, dp.addedRows, dp.deletedRows)), name, logger);
 
-   public static IObservable<DataPacket> AsObservable(this ViewStream source) {
-      return source as IObservable<DataPacket> ?? Observable.Create<DataPacket>(Subscribe);
+   public static IObservable<IDataPacket> AsObservable(this ViewStream source) {
+      return source as IObservable<IDataPacket> ?? Observable.Create<IDataPacket>(Subscribe);
 
-      IDisposable Subscribe(IObserver<DataPacket> observer) {
+      IDisposable Subscribe(IObserver<IDataPacket> observer) {
          source.onChange += OnChange;
          return Disposable.Create(() => source.onChange -= OnChange);
 
          void OnChange(object? newValue) =>
-            observer.OnNext(newValue as DataPacket ?? throw new ArgumentException(
-               $"Expected DataPacket, got {newValue?.GetType()}", nameof(newValue)));
+            observer.OnNext(newValue as IDataPacket ?? throw new ArgumentException(
+               $"Expected IDataPacket, got {newValue?.GetType()}", nameof(newValue)));
       }
    }
 

src/DataStreamSuite/DataStream/Offloading/RemoteQueryManager.cs

@@ -88,8 +88,7 @@ namespace BCG {
          return Task.CompletedTask;
       }
 
-      private void _on_str(DataPacket dp)
-      {
+      void _on_str(IDataPacket dp) {
          foreach (IBlueRow r in dp.newRows)
          {
             string name = r.get<string>("name");

src/DataStreamSuite/DataStream/Offloading/RemoteQueryStream.cs

@@ -9,7 +9,7 @@ namespace BCG {
    [DataStreamUserRespresentation("remoteQuery")]
    public class RemoteQueryStream : ViewStream {
       private readonly string _name, _stream_exp;
-      private PublisherCore _publisher_core = new PublisherCore();
+      private PublisherCore _publisher_core = PublisherCore.Create();
       private ViewStream _source_str;
 
       public RemoteQueryStream(params string[] lines) {

src/DataStreamSuite/DataStream/Primitives/Misc/PubSubTestStream.cs

@@ -70,7 +70,7 @@ namespace bts
         private readonly int _msg_size;
         private readonly TimeSpan _publish_period;
         private DateTime _last_publish_time = DateTime.MinValue;
-        private PublisherCore _pub_core = new PublisherCore();
+        private PublisherCore _pub_core = PublisherCore.Create();
         private BlueTable _table;
         private int _n_pubs = 0;
         private readonly int _max_pubs;
@@ -224,4 +224,4 @@ namespace bts
         }
 
     }
-}
+}

src/DataStreamSuite/DataStream/Primitives/Misc/PubsubEchoTestStream.cs

@@ -1,6 +1,7 @@
 using BCG;
 using BCG.Utilities.General;
 
+using Bts.PubSub;
 using Bts.PubSub.Routing;
 
 using System;
@@ -62,7 +63,9 @@ namespace bts
 
          public static PayloadMessage deserialize(byte[] ser_bytes)
          {
-            var str = new MemoryStream(ser_bytes);
+            var str = new MemoryStream(ser_bytes) {
+               Position = 0
+            };
             BinaryReader br = new BinaryReader(str);
             string guid = br.ReadString();
             Int64 timestamp_ = br.ReadInt64();
@@ -91,11 +94,17 @@ namespace bts
       };
       IOState _state = IOState.SEND;
 
-      public PubsubEchoTestStream(string server_url, TimeSpan publish_period, int payload_size, string tag_name, ViewStreamDatabase vsd) {
-         _router = new PubSubRouter(new PubSubRouterConfig() {
-            PublisherRoutes = new() { new(server_url, ".*") },
-            SubscriberRoutes = new() { new(server_url, ".*") },
-         });
+      public PubsubEchoTestStream()
+         : this(new PubSubRouter(), TimeSpan.FromSeconds(1), 8, "ECHO_TEST", ViewStreamDatabase.instance) { }
+
+      public PubsubEchoTestStream(string server_url, TimeSpan publish_period, int payload_size, string tag_name)
+         : this(server_url, publish_period, payload_size, tag_name, ViewStreamDatabase.instance) { }
+
+      public PubsubEchoTestStream(string server_url, TimeSpan publish_period, int payload_size, string tag_name, ViewStreamDatabase vsd)
+         : this(createRouter(server_url), publish_period, payload_size, tag_name, vsd) { }
+
+      public PubsubEchoTestStream(IPubSubRouter router, TimeSpan publish_period, int payload_size, string tag_name, ViewStreamDatabase vsd) {
+         _router = router;
          _vsd = vsd;
          _payload_sz = payload_size;
          var schema = BlueDataUtility.form_schema("PubSubTest",
@@ -116,7 +125,7 @@ namespace bts
 
       protected override void myOnFirstSubscriber()
       {
-         _pub_core = new PublisherCore(_router);
+         _pub_core = PublisherCore.Create(_router);
          _sub_core = new SubscriberCore(_router);
          _sub_core.subscribe(_tag_name);
          _vsd.addWorker(this);
@@ -129,6 +138,12 @@ namespace bts
          Log.notnull(_sub_core).unsubscribe(_tag_name);
       }
 
+      static PubSubRouter createRouter(string serverUrl) =>
+         new(new PubSubRouterConfig() {
+            PublisherRoutes = new() { new(serverUrl, ".*") },
+            SubscriberRoutes = new() { new(serverUrl, ".*") },
+         });
+
       private void _fire_dp_after_send()
       {
          Log.notnull(_last_sent_msg);
@@ -136,7 +151,7 @@ namespace bts
          BlueRow r = new BlueRow(_table.schema, _table.Count, _last_sent_msg.msg_id, _last_sent_msg.timestamp, DBNull.Value, DBNull.Value,
             DBNull.Value, DBNull.Value);
          _table.addRow(r);
-         var dp_out = BlueDataUtility.build_dp_inc(_table, myLastValue as DataPacket);
+         var dp_out = BlueDataUtility.build_dp_inc(_table, (myLastValue as IDataPacket)?.newRows);
          myFireEvent(dp_out);
       }
 
@@ -152,8 +167,10 @@ namespace bts
          BlueRow nr = new BlueRow(_table.schema, id, _last_sent_msg.msg_id, _last_sent_msg.timestamp, _last_recvd_msg.msg_id, recv_ts,
                                  String.Compare(_last_recvd_msg.msg_id, _last_sent_msg.msg_id) == 0, recv_ts - _last_recvd_msg.timestamp);
          _table.addRow(nr);
-         var dp_out = BlueDataUtility.build_dp_inc(_table, myLastValue as DataPacket);
-         myFireEvent(dp_out);
+         var dp_out = BlueDataUtility.build_dp_inc(_table, (myLastValue as IDataPacket)?.newRows);
+         if (dp_out.HasChanges()) {
+            myFireEvent(dp_out);
+         }
       }
 
       private string _get_next_id()

src/DataStreamSuite/DataStreamCore/BlueData/BlueDataUtility.cs

@@ -15,7 +15,6 @@ using mbts;
 using BCG.Extensions;
 using System.Text.Json;
 using System.Text.Json.Nodes;
-using Newtonsoft.Json;
 using BCG;
 using System.Runtime.CompilerServices;
 using bts;
@@ -295,7 +294,7 @@ namespace BCG {
       }
 
       public static DataPacket build_dp_inc(IBlueRowCollection new_rows, IDataPacket last_fired_dp, bool deep_row_comapre = false) =>
-         build_dp_inc(new_rows, last_fired_dp.newRows, deep_row_comapre);
+         build_dp_inc(new_rows, last_fired_dp?.newRows ?? new BlueTable(new_rows.schema), deep_row_comapre);
 
       public static DataPacket build_dp_inc(IBlueRowCollection new_rows, IBlueRowCollection lastRows, bool deep_row_comapre = false) {
          if (lastRows is null) {
@@ -360,11 +359,11 @@ namespace BCG {
       }
 
       public static Dictionary<string, string> convert_json_to_dictionary(string json_string) {
-         Dictionary<string, string> dictionary = JsonConvert.DeserializeObject<Dictionary<string, string>>(json_string);
+         Dictionary<string, string> dictionary = JsonSerializer.Deserialize<Dictionary<string, string>>(json_string);
          return dictionary;
       }
       public static IBlueRow convert_json_to_row(string json_string, int id = 0) {
-         Dictionary<string, object> dictionary = JsonConvert.DeserializeObject<Dictionary<string, object>>(json_string);
+         Dictionary<string, object> dictionary = JsonSerializer.Deserialize<Dictionary<string, object>>(json_string);
          if (!dictionary.ContainsKey("id")) {
             dictionary.Add("id", id);
          }

src/DataStreamSuite/DataStreamCore/BlueData/ColumnInfo.cs

@@ -1,4 +1,6 @@
 #nullable enable
+using Bts.DataStreamCore;
+
 using System;
 using System.Text.Json.Serialization;
 
@@ -7,6 +9,7 @@ namespace BCG {
    /// Resprsents information about a column schema, ie its ordinal, the type, the name and constant value or expression. where applicable
    /// </summary>
    [Serializable]
+   [JsonConverter(typeof(ColumnInfoJsonConverter))]
    public class ColumnInfo : IComparable<ColumnInfo>, IComparable, IEquatable<ColumnInfo> {
 
       //the default column info does not cach values in a lookup table, therefore it is in group 0
@@ -67,16 +70,12 @@ namespace BCG {
       private Type myColType;
       private object? myColValue;
 
-      [JsonPropertyName("colNo")]
       public int colNo { get { return myColNo; } }
 
-      [JsonPropertyName("colName")]
       public string colName { get { return myColName; } }
 
-      [JsonPropertyName("colType")]
       public Type colType { get { return myColType; } }
 
-      [JsonIgnore]
       public object? colValue {get{return myColValue;}}
    }
 }

src/DataStreamSuite/DataStreamCore/BlueData/ColumnInfoJsonConverter.cs

@@ -0,0 +1,27 @@
+namespace Bts.DataStreamCore;
+
+using BCG;
+
+using Spark.Core.Json;
+
+using System;
+using System.Text.Json;
+using System.Text.Json.Nodes;
+using System.Text.Json.Serialization;
+
+public class ColumnInfoJsonConverter : JsonConverter<ColumnInfo> {
+   public override ColumnInfo Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) {
+      var jobj = JsonSerializer.Deserialize<JsonObject>(ref reader);
+      return new(
+         jobj[nameof(ColumnInfo.colName)].GetValue<string>(),
+         jobj[nameof(ColumnInfo.colNo)].GetValue<int>(),
+         Type.GetType(jobj[nameof(ColumnInfo.colType)].GetValue<string>()));
+   }
+
+   public override void Write(Utf8JsonWriter writer, ColumnInfo value, JsonSerializerOptions options) {
+      using var _ = writer.WriteObject();
+      writer.WriteString(nameof(ColumnInfo.colName), value.colName);
+      writer.WriteNumber(nameof(ColumnInfo.colNo), value.colNo);
+      writer.WriteString(nameof(ColumnInfo.colType), value.colType.ToString());
+   }
+}

src/DataStreamSuite/DataStreamCore/DataPacket.cs

@@ -41,7 +41,7 @@ namespace BCG {
       }
 
       public IBlueRowCollection newRows, addedRows, deletedRows;
-      public IDictionary<string, object> meta_data = null;
+      public Dictionary<string, object> meta_data = null;
       public DataPacket() { }
 
       public DataPacket(IBlueTableSchema schema)
@@ -53,7 +53,7 @@ namespace BCG {
          this.deletedRows = deletedRows;
       }
       public DataPacket(IBlueRowCollection newRows, IBlueRowCollection addedRows, IBlueRowCollection deletedRows,
-                  IDictionary<string, object> meta_data) {
+                  Dictionary<string, object> meta_data) {
          this.newRows = newRows;
          this.addedRows = addedRows;
          this.deletedRows = deletedRows;
@@ -63,6 +63,7 @@ namespace BCG {
       IBlueRowCollection IDataPacket.newRows => newRows;
       IBlueRowCollection IDataPacket.addedRows => addedRows;
       IBlueRowCollection IDataPacket.deletedRows => deletedRows;
+      IReadOnlyDictionary<string, object> IDataPacket.Metadata => meta_data;
 
       public void get_new_added_deleted(out ICollection new_rows, out ICollection added_rows, out ICollection deleted_rows) {
          new_rows = newRows;

src/DataStreamSuite/DataStreamCore/DataStreamCore.csproj

@@ -8,7 +8,6 @@
 
   <ItemGroup>
     <PackageReference Include="Google.Protobuf" Version="3.24.3" />
-    <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
     <PackageReference Include="OpenAI" Version="2.0.0" />
     <PackageReference Include="SimMetrics.Net" Version="1.0.5" />
   </ItemGroup>

src/DataStreamSuite/DataStreamCore/DynamicLoading/PrimitiveManager.cs

@@ -44,11 +44,6 @@ namespace BCG {
       /// </summary>
       /// <param name="assemblyName"></param>
       public void loadAssembly(string assemblyName) {
-         Func<Type, bool> is_blue_orb = (t) => {
-            //if (!assemblyName.Contains("SharedStreams.dll"))
-            //   return true;
-            return t.Namespace == "bts.BlueOrb";// || t.Namespace == "BCG";
-         };
          bts.Log.Gen.info($"PrimitiveManager.loadAssembly assemblyName: {assemblyName}");
          if (myAssemblyNames.Contains(assemblyName))
             return;
@@ -58,7 +53,7 @@ namespace BCG {
             Assembly.LoadFrom(assemblyName);
 
          foreach (Type type in loadedAssembly.GetTypes())
-            if (!type.IsInterface && is_blue_orb(type)
+            if (!type.IsInterface
                && !type.IsAbstract
                && type.IsSubclassOf(typeof(ViewStream))
                && type.IsPublic) {

src/DataStreamSuite/DataStreamCore/IDataPacket.cs

@@ -13,6 +13,7 @@ namespace BCG {
       IBlueRowCollection newRows { get; }
       IBlueRowCollection addedRows { get; }
       IBlueRowCollection deletedRows { get; }
+      IReadOnlyDictionary<string, object> Metadata { get; }
       //this is an ICollection of IDictionary
       void get_new_added_deleted(out ICollection new_rows, out ICollection added_rows, out ICollection deleted_rows);
    }

src/DataStreamSuite/DataStreamCore/Messaging/Publish/PubTableBroker.cs

@@ -1,15 +1,10 @@
-using System;
-using System.Collections;
-using System.Collections.Generic;
-using System.Linq.Expressions;
-using System.Xml.Serialization;
-
-using BCG.Utilities;
-
 using bts;
 
 using Bts.DataStreamCore;
 
+using System;
+using System.Collections.Generic;
+
 namespace BCG {
    /// <summary>
    /// This class will be used by the clients to publish tabular data to the pubsub
@@ -22,7 +17,7 @@ namespace BCG {
       protected bool myWaitingForFirstOp = false;
 
       public PubTableBroker()
-         : this(new PublisherCore()) { }
+         : this(PublisherCore.Create()) { }
 
       public PubTableBroker(IPublisherCore publisher) {
          _publisher_core = publisher;

src/DataStreamSuite/DataStreamPublisher/DataStreamPublisher.csproj

@@ -11,10 +11,20 @@
     <PackageReference Include="CommandLineParser" Version="2.9.1" />
   </ItemGroup>
 
-  <ItemGroup>
-    <ProjectReference Include="..\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj" />
-    <ProjectReference Include="..\SharedStreams\SharedStreams.csproj" />
-  </ItemGroup>
+  <Choose>
+    <When Condition="'$(UseProjectRefs)' == 'true'">
+      <ItemGroup>
+        <ProjectReference Include="..\..\cs\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj" />
+        <ProjectReference Include="..\SharedStreams\SharedStreams.csproj" />
+      </ItemGroup>
+    </When>
+    <Otherwise>
+      <ItemGroup>
+        <PackageReference Include="Bts.Extensions.Hosting" Version="$(SparkVersion)" />
+        <PackageReference Include="SharedStreams" Version="$(SparkVersion)" />
+      </ItemGroup>
+    </Otherwise>
+  </Choose>
 
   <ItemGroup>
     <ConfigurationProfile Include="appsettings.Development.json" />

src/DataStreamSuite/DataStreamQueryAnalyzer/DataStreamQueryAnalyzer.csproj

@@ -16,6 +16,12 @@
     <ProjectReference Include="..\BaseUtilities\BaseUtilities.csproj" />
     <ProjectReference Include="..\GuiUtilities\GuiUtilities.csproj" />
     <ProjectReference Include="..\SharedStreams\SharedStreams.csproj" />
+    <ProjectReference Include="..\..\cs\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ConfigurationProfile Include="appsettings.json" />
+    <ConfigurationProfile Include="appsettings.*.json" />
   </ItemGroup>
 
 </Project>

src/DataStreamSuite/DataStreamQueryAnalyzer/MainForm.Designer.cs

@@ -2,21 +2,6 @@
 {
    public partial class MainForm
    {
-      /// <summary>
-      /// Clean up any resources being used.
-      /// </summary>
-      protected override void Dispose(bool disposing)
-      {
-         if (disposing)
-         {
-            if (components != null)
-            {
-               components.Dispose();
-            }
-         }
-         base.Dispose(disposing);
-      }
-
       #region Windows Form Designer generated code
       /// <summary>
       /// Required method for Designer support - do not modify

src/DataStreamSuite/DataStreamQueryAnalyzer/MainForm.cs

@@ -1,16 +1,25 @@
 using bts;
 
+using Bts.DataStream;
 using Bts.DataStreamExpressions;
 using Bts.PubSub;
 
+using Spark.Core.Disposables;
+using Spark.Core.Reactive;
+
 using System;
-using System.Collections.Generic;
 using System.Collections;
+using System.Collections.Generic;
 using System.ComponentModel;
 using System.Diagnostics;
 using System.Drawing;
 using System.IO;
+using System.Linq;
+using System.Reactive;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
 using System.Text.RegularExpressions;
+using System.Threading;
 using System.Windows.Forms;
 
 namespace BCG
@@ -35,8 +44,11 @@ namespace BCG
       private string mySite = "TODO";
       private string myLastClassName = "MyView";
       private ViewStream? myAllTagsStream;
+      IBlueRowCollection? tagRows;
       private string myKQ;
       private TextWriter? myConsoleOutput;
+      IDataPacket? pendingData;
+      IDisposable? titleSub;
 
       public MainForm()
       {
@@ -44,6 +56,17 @@ namespace BCG
          InitializeComponent();
          // myKQ = ConfigurationRequester.getApplicationQualifier();         
          myKQ = "TODO";
+         var handleCreatedSubject = new Subject<bool>();
+         HandleCreated += (s, e) => handleCreatedSubject.OnNext(true);
+         titleSub = Disposable.Create(
+            handleCreatedSubject.SelectSeries(_ =>
+               Observable.FromEvent(
+                  cb => Invoke(() => titleInvalidated += cb),
+                  cb => Invoke(() => titleInvalidated -= cb))
+               .Prepend(Unit.Default)
+               .SelectSeries(_ => new PubsubRoutesStream().AsObservable().Take(1)))
+               .Subscribe(dp => BeginInvoke(() => refreshTitle(dp.newRows))),
+            handleCreatedSubject);
          mySetText();
 
          myWindowState = new PersistWindowState();
@@ -58,11 +81,25 @@ namespace BCG
          };
       }
 
+      event Action? titleInvalidated;
+
       public string query_text
       {
          get { return myQueryTextBox.Text; }
          set { myQueryTextBox.Text = value; }
       }
+
+      /// <summary>
+      /// Clean up any resources being used.
+      /// </summary>
+      protected override void Dispose(bool disposing) {
+         if (disposing) {
+            Disposable.Dispose(ref titleSub);
+            components?.Dispose();
+         }
+         base.Dispose(disposing);
+      }
+
       private ViewStream myGetAllTagsStream()
       {
          // TODO: Port PublishedTagsStream from edge (then we can uncomment impl below)
@@ -74,9 +111,10 @@ namespace BCG
 
       private void mySetText()
       {
-         ViewStream routes = new PubsubRoutesStream();
-         routes.onChange += (o) => { };
-         var rows = routes.rows;
+         titleInvalidated?.Invoke();
+      }
+
+      void refreshTitle(IBlueRowCollection rows) {
          string route_txt = String.Empty;
          foreach (IBlueRow r in rows)
          {
@@ -144,8 +182,10 @@ namespace BCG
 
       private void myHandleTableListChange(object o)
       {
-         IBlueRowCollection rows = (o as DataPacket).newRows;
-         myUpdateTableList();
+         BeginInvoke(() => {
+            tagRows = ((IDataPacket)o).newRows;
+            myUpdateTableList();
+         });
       }
 
 
@@ -334,18 +374,24 @@ namespace BCG
       // this is the stream callback
       private void myCallback(object o)
       {
-         _stopwatch.Stop();
-         var timer = Stopwatch.StartNew();
-         DataPacket dp = o as DataPacket;
-         myCrono.Stop();
-         Log.Ui.diag($"Received callback for {dp.schema.tableName}");
-         if (o == null)
-         {
-            System.Windows.Forms.MessageBox.Show(this, "Received a Null Data packet",
-                                                   "Invalid Data", MessageBoxButtons.OK, MessageBoxIcon.Error);
+         var dp = (IDataPacket)o;
+         Log.Ui.diag(dp.GetIncrementalIdString());
+         Volatile.Write(ref pendingData, dp);
+      }
+
+      private Stopwatch myCrono = new();
+
+      private DataStreamExpression? myExp;
+
+      private void myTimer_Tick(object sender, System.EventArgs e)
+      {
+         var dp = Interlocked.Exchange(ref pendingData, null);
+         if (dp is null) {
             return;
          }
-         Log.Ui.diag(dp.toIncrementalIDString());
+         _stopwatch.Stop();
+         var timer = Stopwatch.StartNew();
+         myCrono.Stop();
          myToolStripStatusLabelTime.BackColor = Color.Black;
          myToolStripStatusLabelRows.BackColor = Color.Black;
          myToolStripStatusLabelTime.ForeColor = Color.White;
@@ -362,7 +408,7 @@ namespace BCG
          {
             myChangeStyle = false;
             myDataGrid.Columns.Clear();
-            myDataGrid.Columns.AddRange(Array.ConvertAll(dp.schema.columnInfo, _column_factory));
+            myDataGrid.Columns.AddRange(Array.ConvertAll(dp.Schema.columnInfo, _column_factory));
          }
          myCrono.Restart();
          // myDataGrid.CaptionText = myCrono.time.TotalMilliseconds + " ms ";
@@ -373,35 +419,13 @@ namespace BCG
 
          var varNames = new List<string>(0);
          var code = myExp?.print(9, varNames);
-         myPrintCode(dp.schema, varNames, code);
+         myPrintCode(dp.Schema, varNames, code);
          myConsoleRichTextBox.Clear();
          // this was was very very slow
          //myConsoleRichTextBox.Text = BlueDataUtility.get_printed_rows(dp.newRows, (col) => 30);
          timer.Stop();
          Log.Gen.info(
-            $"DSQA MainForm.myCallback table: {dp.schema.tableName} rows: {dp.newRows.Count} time: {timer.Elapsed.TotalMilliseconds:0.00}ms");
-      }
-
-      private Stopwatch myCrono = new();
-
-      private DataStreamExpression? myExp;
-
-      private void myTimer_Tick(object sender, System.EventArgs e)
-      {
-         // try 
-         // {
-         ViewStreamDatabase.instance.doWork();
-         MasterWorker.getInstance(false).doWork();
-         //}
-         // catch (Exception exc) {
-         //    Log.Ui.warn(exc.ToString());
-
-         //    myDataGrid.DataSource = null;
-         //    //myTabControl.SelectedIndex = 1;
-         //    myTabControl.SelectedTab = myErrorTabPage;
-         //    myErrorTextBox.Text = exc.Message;
-         //    myErrorTextBox.Text += "\n" + exc.StackTrace;
-         // }
+            $"DSQA MainForm.myCallback table: {dp.Schema.tableName} rows: {dp.newRows.Count} time: {timer.Elapsed.TotalMilliseconds:0.00}ms");
       }
 
       class StringWrapper
@@ -684,7 +708,7 @@ namespace BCG
 
       private void myUpdateTableList()
       {
-         IBlueRowCollection rows = myAllTagsStream.rows;
+         var rows = tagRows;
          if (rows == null)
             return;
 
@@ -742,20 +766,25 @@ namespace BCG
 
          ViewStream s = new LoaderStream(@"P:\Dropbox\Data\spread_md_esoa_cme");
          s = new MutableAppendStream(s, s);
-         //s = new MutableAppendStream(s, s);
-         List<IBlueRow> rows = new List<IBlueRow>();
-         foreach (IBlueRow r in s.rows)
-            rows.Add(r);
+         s.onChange += OnChange;
 
-         for (int i = 0; i < 100; i++)
-         {
-            timer.Restart();
-            rows.Sort(delegate (IBlueRow r1, IBlueRow r2)
+         void OnChange(object? obj) {
+            s.onChange -= OnChange;
+            //s = new MutableAppendStream(s, s);
+            var rowCollection = ((IDataPacket)obj!).newRows;
+            var rows = new List<IBlueRow>(rowCollection.Count);
+            rows.AddRange(rowCollection.Cast<IBlueRow>());
+
+            for (int i = 0; i < 100; i++)
             {
-               return _sort_dicts_index(r1, r2, 5, true);
-            });
-            timer.Stop();
-            Log.Ui.diag($"nrows = {rows.Count} sorting time = {timer.Elapsed.TotalMilliseconds:0.00}ms");
+               timer.Restart();
+               rows.Sort(delegate (IBlueRow r1, IBlueRow r2)
+               {
+                  return _sort_dicts_index(r1, r2, 5, true);
+               });
+               timer.Stop();
+               Log.Ui.diag($"nrows = {rows.Count} sorting time = {timer.Elapsed.TotalMilliseconds:0.00}ms");
+            }
          }
       }
 

src/DataStreamSuite/DataStreamQueryAnalyzer/Program.cs

@@ -1,59 +1,60 @@
 using BCG;
 
-using bts;
-using bts.api.core;
+using Bts.DataStream.Services;
+using Bts.DataStreamCore.Services;
+using Bts.Extensions.Hosting;
+using Bts.PubSub.Services;
 
-using Bts.DataStreamQueryAnalyzer;
-using Bts.PubSub;
-
-using CommandLine;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
 
 using Spark.Api.Services;
 using Spark.Core.Services;
 
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using System.Windows.Forms;
 
 static class Program {
-   [STAThread]
-   static async Task Main(string[] args) {
-      await Parser.Default.ParseArguments<DataStreamQueryAnalyzerConfig>(args)
-         .MapResult(RunAsync, errs => Task.FromResult(1));
-
-      async Task<int> RunAsync(DataStreamQueryAnalyzerConfig cfg) {
-         try {
-            cfg.MergeWithConfigFile();
-            cfg.MergeWith(DataStreamQueryAnalyzerConfig.Default());
-            PubSubConfig.SetGlobalConfig(Log.notnull(cfg.PubSub));
-            var configLoader = new SparkConfigLoader(new SparkConfigLoaderOpts() { ConfigServer = cfg.Sparkd });
-            var serviceContext = new ServiceContext(configLoader);
-            var credentials = new SparkApiCredentials() {
-               User = cfg.User,
-               Password = cfg.Password
-            };
-            var loginService = new SparkdLoginService(serviceContext, credentials, configLoader);
-            var permissions = new UserDataPermissions(loginService);
-            var mdClientConfig = new MarketDataClientConfig() { HotPollInterval = TimeSpan.FromSeconds(0.5) };
-            var mdClient = new MarketDataClient(configLoader, loginService, permissions, mdClientConfig);
+   static async Task<int> Main(string[] args) {
+      Console.Write("Starting DataStreamQueryAnalyzer");
+      int exitCode = 0;
+      IHost? host = null;
+      try {
+         host = HostBuilderExtensions.CreateBtsBuilder(args)
+            .ConfigureServices(ConfigureServices)
+            .Build();
+         var mdClient = host.Services.GetService<IMarketDataClient>();
+         if (mdClient is not null) {
             PrimitiveManager.SetMarketDataClient(mdClient);
-            using var vsd = new ViewStreamDatabase(new SubscriberCoreFactory());
-            ViewStreamDatabase.SetGlobalInstance(new ViewStreamDatabase(new SubscriberCoreFactory()));
-            Log.Gen.info("Starting DataStreamQueryAnalyzer");
-            var mainForm = new MainForm();
-            if (cfg.StreamFile is not null) {
-               mainForm.loadFile(cfg.StreamFile);
-            }
-            Application.Run(mainForm);
-         } catch (Exception ex) {
-            // Write the error to console first, in case the logging system is compromised.
-            var error_msg = $"DataStreamPublisher threw exception:\n{ex}";
-            Console.Error.WriteLine(error_msg);
-            Console.Error.Flush();
-            bts.Log.Gen.error(error_msg);
-            return 1;
          }
-         return 0;
+         await host.StartAsync();
+         var winFormsThread = new Thread(RunWinFormsApp);
+         winFormsThread.SetApartmentState(ApartmentState.STA);
+         winFormsThread.Start();
+         winFormsThread.Join();
+      } catch (Exception ex) {
+         Console.Error.Write($"DataStreamQueryAnalyzer crashed: {ex}");
+         exitCode = 1;
       }
+      if (host is not null) {
+         await host.StopAsync();
+      }
+      return exitCode;
    }
+
+   static void RunWinFormsApp() {
+      var win = new MainForm();
+      Application.Run(win);
+   }
+
+   static void ConfigureServices(HostBuilderContext context, IServiceCollection services) =>
+      services.AddSpark(context.Configuration.GetSection("Spark"))
+         .AddSparkApi(context.Configuration.GetSection("SparkApi"))
+         .AddMasterWorker(context.Configuration.GetSection("MasterWorker"))
+         .AddViewStreamDatabase()
+         .AddPubSubClient(context.Configuration.GetSection("PubSub"))
+         .AddPubTableBroker()
+         .AddRemoteQueryManager();
 }

src/DataStreamSuite/DataStreamQueryAnalyzer/appsettings.json

@@ -6,9 +6,9 @@
     "Interval": "00:00:00.1"
   },
   "Spark": {
-    "ConfigServer": "http://127.0.0.1:8000/config"
   },
   "SparkApi": {
+    "IsEnabled": false,
     "Credentials": {
       "User": "bts",
       "Password": ""

src/DataStreamSuite/DataStreamSuite.sln

@@ -37,9 +37,11 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.ColumnExpressions", "Bt
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.ColumnSortExpressions", "Bts.ColumnSortExpressions\Bts.ColumnSortExpressions.csproj", "{B038CB72-4E81-4AF1-976F-ED8655C2C3D6}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Extensions.Hosting", "Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj", "{5C528A65-E43C-48A2-8CA9-ED2D0E356A37}"
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Extensions.Hosting", "..\cs\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj", "{5C528A65-E43C-48A2-8CA9-ED2D0E356A37}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Net", "Bts.Net\Bts.Net.csproj", "{496333E0-E11D-45EE-B0D6-BBAA824A120D}"
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.Net", "..\cs\Bts.Net\Bts.Net.csproj", "{496333E0-E11D-45EE-B0D6-BBAA824A120D}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.IO.Compression", "..\cs\Bts.IO.Compression\Bts.IO.Compression.csproj", "{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -279,6 +281,18 @@ Global
 		{496333E0-E11D-45EE-B0D6-BBAA824A120D}.Release|x64.Build.0 = Release|Any CPU
 		{496333E0-E11D-45EE-B0D6-BBAA824A120D}.Release|x86.ActiveCfg = Release|Any CPU
 		{496333E0-E11D-45EE-B0D6-BBAA824A120D}.Release|x86.Build.0 = Release|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Debug|x64.Build.0 = Debug|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Debug|x86.Build.0 = Debug|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Release|Any CPU.Build.0 = Release|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Release|x64.ActiveCfg = Release|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Release|x64.Build.0 = Release|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Release|x86.ActiveCfg = Release|Any CPU
+		{3BF059B0-51C0-44A8-8297-9B7464DC5BA3}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

src/DataStreamSuite/PubSub/Client/PubSubTagHandlerFactoryConfig.cs

@@ -17,11 +17,6 @@ public class PubSubTagHandlerFactoryConfig {
    public static PubSubTagHandlerFactoryConfig Default() {
       return new() {
          Handlers = new() {
-            new() {
-               TagType = string.Empty,
-               Assembly = "DataStreamCore",
-               Class = "bts.PubsubServerStreamTagHandler",
-            },
             new() {
                TagType = DSTagManager.DataStreamTagType,
                Assembly = "DataStreamCore",
@@ -36,7 +31,12 @@ public class PubSubTagHandlerFactoryConfig {
                TagType = "System_Stats",
                Assembly = "PubSub",
                Class = "BCG.SystemStatsPublisher",
-            }
+            },
+            new() {
+               TagType = ".*",
+               Assembly = "DataStreamCore",
+               Class = "bts.PubsubServerGenericDataHandler",
+            },
          }
       };
    }

src/DataStreamSuite/PubSub/Misc/CompressionUtilities/CompressionUtilities.csproj

@@ -1,273 +0,0 @@
-<?xml version="1.0" ?>
-<Project DefaultTargets="Build" ToolsVersion="15.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
-  <PropertyGroup>
-    <ProjectType>Local</ProjectType>
-    <ProductVersion>9.0.30729</ProductVersion>
-    <SchemaVersion>2.0</SchemaVersion>
-    <ProjectGuid>{41DBB1A3-1A4E-457D-BB2C-BFDE486E328B}</ProjectGuid>
-    <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
-    <Platform Condition=" '$(Platform)' == '' ">x64</Platform>
-    <ApplicationIcon>
-    </ApplicationIcon>
-    <AssemblyKeyContainerName>
-    </AssemblyKeyContainerName>
-    <AssemblyName>CompressionUtilities</AssemblyName>
-    <AssemblyOriginatorKeyFile>..\..\StrongNameKeys\Builder.snk</AssemblyOriginatorKeyFile>
-    <DefaultClientScript>JScript</DefaultClientScript>
-    <DefaultHTMLPageLayout>Grid</DefaultHTMLPageLayout>
-    <DefaultTargetSchema>IE50</DefaultTargetSchema>
-    <DelaySign>false</DelaySign>
-    <OutputType>Library</OutputType>
-    <RootNamespace>CompressionUtilities</RootNamespace>
-    <RunPostBuildEvent>OnBuildSuccess</RunPostBuildEvent>
-    <StartupObject>
-    </StartupObject>
-    <SignAssembly>false</SignAssembly>
-    <TargetFrameworkVersion>v4.0</TargetFrameworkVersion>
-    <ApplicationRevision>0</ApplicationRevision>
-    <ApplicationVersion>1.0.0.%2a</ApplicationVersion>
-  </PropertyGroup>
-  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
-    <AllowUnsafeBlocks>false</AllowUnsafeBlocks>
-    <BaseAddress>285212672</BaseAddress>
-    <CheckForOverflowUnderflow>false</CheckForOverflowUnderflow>
-    <ConfigurationOverrideFile>
-    </ConfigurationOverrideFile>
-    <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <DocumentationFile>
-    </DocumentationFile>
-    <DebugSymbols>true</DebugSymbols>
-    <FileAlignment>4096</FileAlignment>
-    <NoStdLib>false</NoStdLib>
-    <NoWarn>
-    </NoWarn>
-    <Optimize>false</Optimize>
-    <RegisterForComInterop>false</RegisterForComInterop>
-    <RemoveIntegerChecks>false</RemoveIntegerChecks>
-    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
-    <WarningLevel>4</WarningLevel>
-    <DebugType>full</DebugType>
-    <ErrorReport>prompt</ErrorReport>
-    <CodeAnalysisRuleSet>AllRules.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
-  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
-    <AllowUnsafeBlocks>false</AllowUnsafeBlocks>
-    <BaseAddress>285212672</BaseAddress>
-    <CheckForOverflowUnderflow>false</CheckForOverflowUnderflow>
-    <ConfigurationOverrideFile>
-    </ConfigurationOverrideFile>
-    <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <DocumentationFile>
-    </DocumentationFile>
-    <DebugSymbols>true</DebugSymbols>
-    <FileAlignment>4096</FileAlignment>
-    <NoStdLib>false</NoStdLib>
-    <NoWarn>
-    </NoWarn>
-    <Optimize>false</Optimize>
-    <RegisterForComInterop>false</RegisterForComInterop>
-    <RemoveIntegerChecks>false</RemoveIntegerChecks>
-    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
-    <WarningLevel>4</WarningLevel>
-    <DebugType>full</DebugType>
-    <ErrorReport>prompt</ErrorReport>
-    <DefineDebug>true</DefineDebug>
-    <CodeAnalysisRuleSet>AllRules.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
-  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Test|AnyCPU' ">
-    <DebugSymbols>true</DebugSymbols>
-    <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <BaseAddress>285212672</BaseAddress>
-    <DebugType>full</DebugType>
-    <PlatformTarget>x64</PlatformTarget>
-    <ErrorReport>prompt</ErrorReport>
-    <CodeAnalysisRuleSet>AllRules.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
-  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|x64' ">
-    <DebugSymbols>true</DebugSymbols>
-    <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <BaseAddress>285212672</BaseAddress>
-    <FileAlignment>4096</FileAlignment>
-    <DebugType>full</DebugType>
-    <PlatformTarget>x64</PlatformTarget>
-    <ErrorReport>prompt</ErrorReport>
-    <CodeAnalysisRuleSet>AllRules.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
-  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|x64' ">
-    <DebugSymbols>true</DebugSymbols>
-    <DefineDebug>true</DefineDebug>
-    <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <BaseAddress>285212672</BaseAddress>
-    <FileAlignment>4096</FileAlignment>
-    <DebugType>full</DebugType>
-    <PlatformTarget>x64</PlatformTarget>
-    <ErrorReport>prompt</ErrorReport>
-    <CodeAnalysisRuleSet>AllRules.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
-  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Test|x64' ">
-    <DebugSymbols>true</DebugSymbols>
-    <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <BaseAddress>285212672</BaseAddress>
-    <DebugType>full</DebugType>
-    <PlatformTarget>x64</PlatformTarget>
-    <ErrorReport>prompt</ErrorReport>
-    <CodeAnalysisRuleSet>AllRules.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
-  <ItemGroup>
-    <Reference Include="System">
-      <Name>System</Name>
-    </Reference>
-    <Reference Include="System.Core">
-      <RequiredTargetFramework>4.0</RequiredTargetFramework>
-    </Reference>
-    <Reference Include="System.Data">
-      <Name>System.Data</Name>
-    </Reference>
-    <Reference Include="System.Xml">
-      <Name>System.XML</Name>
-    </Reference>
-  </ItemGroup>
-  <ItemGroup>
-    <Compile Include="AssemblyInfo.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="BlueCompressUtilities\IBlueCompressor.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="BlueCompressUtilities\SharpZipCompressor.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\BZip2\BZip2.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\BZip2\BZip2Constants.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\BZip2\BZip2InputStream.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\BZip2\BZip2OutputStream.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Checksums\Adler32.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Checksums\Crc32.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Checksums\IChecksum.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Checksums\StrangeCrc.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\GZip\GZipConstants.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\GZip\GZipInputStream.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\GZip\GZipOutputStream.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Tar\InvalidHeaderException.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Tar\TarArchive.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Tar\TarBuffer.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Tar\TarEntry.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Tar\TarHeader.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Tar\TarInputStream.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Tar\TarOutputStream.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\ZipException.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\Deflater.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\DeflaterConstants.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\DeflaterEngine.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\DeflaterHuffman.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\DeflaterPending.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\Inflater.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\InflaterDynHeader.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\InflaterHuffmanTree.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\PendingBuffer.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\Streams\DeflaterOutputStream.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\Streams\InflaterInputStream.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\Streams\OutputWindow.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\Compression\Streams\StreamManipulator.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\ZipConstants.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\ZipEntry.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\ZipFile.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\ZipInputStream.cs">
-      <SubType>Code</SubType>
-    </Compile>
-    <Compile Include="SharpZipLib\Zip\ZipOutputStream.cs">
-      <SubType>Code</SubType>
-    </Compile>
-  </ItemGroup>
-  <ItemGroup>
-    <BootstrapperPackage Include="Microsoft.Net.Client.3.5">
-      <Visible>False</Visible>
-      <ProductName>.NET Framework 3.5 SP1 Client Profile</ProductName>
-    </BootstrapperPackage>
-    <BootstrapperPackage Include="Microsoft.Net.Framework.3.5.SP1">
-      <Visible>False</Visible>
-      <ProductName>.NET Framework 3.5 SP1</ProductName>
-    </BootstrapperPackage>
-    <BootstrapperPackage Include="Microsoft.Windows.Installer.3.1">
-      <Visible>False</Visible>
-      <ProductName>Windows Installer 3.1</ProductName>
-    </BootstrapperPackage>
-  </ItemGroup>
-  <Import Project="..\..\build\blue2.cs.targets"/>
-  <PropertyGroup>
-    <PreBuildEvent>
-    </PreBuildEvent>
-    <PostBuildEvent>
-    </PostBuildEvent>
-  </PropertyGroup>
-</Project>

src/DataStreamSuite/PubSub/PubSub.csproj

@@ -4,17 +4,20 @@
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <TargetFramework>net8.0</TargetFramework>
     <Nullable>enable</Nullable>
+    <PackageId>Bts.PubSub</PackageId>
   </PropertyGroup>
 
   <Choose>
     <When Condition="'$(UseProjectRefs)' == 'true'">
       <ItemGroup>
+        <ProjectReference Include="..\..\cs\Bts.IO.Compression\Bts.IO.Compression.csproj" />
         <ProjectReference Include="..\..\cs\Bts.Spark.Core\Bts.Spark.Core.csproj" />
-        <ProjectReference Include="..\Bts.Net\Bts.Net.csproj" />
+        <ProjectReference Include="..\..\cs\Bts.Net\Bts.Net.csproj" />
       </ItemGroup>
     </When>
     <Otherwise>
       <ItemGroup>
+        <PackageReference Include="Bts.IO.Compression" Version="$(SparkVersion)" />
         <PackageReference Include="Bts.Spark.Core" Version="$(SparkVersion)" />
         <PackageReference Include="Bts.Net" Version="$(SparkVersion)" />
       </ItemGroup>

src/DataStreamSuite/PubSub/PublisherCore.cs

@@ -43,10 +43,7 @@ namespace BCG {
 
       #region constructors
 
-      public PublisherCore(IPubSubRouter? router = null, IPubSubTagHandlerFactory? tagHandlerFactory = null, IServiceProvider? serviceProvider = null) {
-         router ??= new PubSubRouter();
-         serviceProvider ??= PubSubServices.ServiceProvider;
-         tagHandlerFactory ??= new PubsubTagHandlerFactory(serviceProvider, PubSubTagHandlerFactoryConfig.Default());
+      public PublisherCore(IPubSubRouter router, IPubSubTagHandlerFactory tagHandlerFactory) {
          _router = router;
          _state = new(tagHandlerFactory);
          Thread pump_thread = new Thread(new ThreadStart(_message_pump));
@@ -54,6 +51,14 @@ namespace BCG {
          pump_thread.Start();
       }
 
+      public static PublisherCore Create(IPubSubRouter? router = null, IPubSubTagHandlerFactory? tagHandlerFactory = null, IServiceProvider? serviceProvider = null) {
+         router ??= new PubSubRouter();
+         tagHandlerFactory ??= new PubsubTagHandlerFactory(
+            serviceProvider ?? PubSubServices.ServiceProvider,
+            PubSubConfig.Instance.TagHandlerFactory ?? PubSubTagHandlerFactoryConfig.Default());
+         return new(router, tagHandlerFactory);
+      }
+
       public void Dispose() { }
       #endregion
 

src/DataStreamSuite/PubSub/Services/PubSubServices.cs

@@ -114,6 +114,7 @@ public static class PubSubServices {
       PubSubConfig.SetGlobalConfig(configurationSection.Get<PubSubConfig>() ?? PubSubConfig.Default());
       return services.AddGlobalServiceProvider()
          .AddPubSubRouter(configurationSection.GetSection(nameof(PubSubConfig.Routing)))
+         .AddTagHandlerFactory(configurationSection.GetSection(nameof(PubSubConfig.TagHandlerFactory)))
          .AddPublisherCore()
          .AddSubscriberCore()
          .AddManagedSubscriber();

src/DataStreamSuite/PubSub/Testing/TestPublisher.cs

@@ -1,10 +1,5 @@
-using BCG;
-
-using bts;
-
 using System;
 using System.IO;
-using System.Runtime.Serialization;
 
 namespace BCG {
    public class TestPubItem : IPubItem {
@@ -24,7 +19,7 @@ namespace BCG {
    }
    public class TestPublisher {
       TestPubItem _tbi = new TestPubItem();
-      IPublisherCore pcore = new PublisherCore();
+      IPublisherCore pcore = PublisherCore.Create();
       bool _hit_key_to_send = false;
       public TestPublisher(bool hit_key_to_send) {
          _hit_key_to_send = hit_key_to_send;
@@ -50,4 +45,4 @@ namespace BCG {
          }
       }
    }
-}
+}

src/DataStreamSuite/PubSubServer/PubSubServer.csproj

@@ -6,10 +6,20 @@
     <Nullable>enable</Nullable>
   </PropertyGroup>
 
-  <ItemGroup>
-    <ProjectReference Include="..\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj" />
-    <ProjectReference Include="..\PubSub\PubSub.csproj" />
-  </ItemGroup>
+  <Choose>
+    <When Condition="'$(UseProjectRefs)' == 'true'">
+      <ItemGroup>
+        <ProjectReference Include="..\..\cs\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj" />
+        <ProjectReference Include="..\PubSub\PubSub.csproj" />
+      </ItemGroup>
+    </When>
+    <Otherwise>
+      <ItemGroup>
+        <PackageReference Include="Bts.PubSub" Version="$(SparkVersion)" />
+        <PackageReference Include="Bts.Extensions.Hosting" Version="$(SparkVersion)" />
+      </ItemGroup>
+    </Otherwise>
+  </Choose>
 
   <ItemGroup>
     <ConfigurationProfile Include="appsettings.Development.json" />

src/DataStreamSuite/SharedStreams/BlueOrb/MarketData/Historical/HistoricalBarsStream.cs

@@ -215,8 +215,6 @@ namespace bts.BlueOrb {
 
    using bts.api;
 
-   using Newtonsoft.Json.Serialization;
-
    [DataStreamUserRespresentation("FuturesContinousContract")]
    public class FuturesContinousContractStream : ViewStream {
       // the argument is symbol, like "ES" etc and not the contract itself

src/DataStreamSuite/SharedStreams/Business/MarketData/PriceStatsStream.cs

@@ -80,7 +80,7 @@ namespace BCG {
             int id = Convert.ToInt32(skey);
             PriceStats pstats = kv.Value;
             var today = DateTime.Now.Date;
-            IBlueRow r = new BlueRow(_pstats_scheam, id, skey, pstats.security.ToString(), today, pstats.open, pstats.close, pstats.settlement);
+            IBlueRow r = new BlueRow(_pstats_scheam, id, skey, pstats.Security.ToString(), today, pstats.Open, pstats.Close, pstats.Settlement);
             new_state.addRow(r);
          }
          DataPacket dp;

src/DataStreamSuite/SharedStreams/RowTemplates/RefTypeRow.tt

@@ -76,33 +76,45 @@ public <#= propType #> <#= name #> <#= getter #>
    #>
 
    // Indexer by column name
-   public object? this[string col] => col switch {
-      <# 
-      // Generate case statements for each column name
-      for (int i = 0; i < columns.Count; i++) {
-         var name = columns[i].Item1;
-         var key = columns[i].Item3;
-      #>
-"<#= key #>" => <#= name #>,
-      <# 
-} 
-      #>
-_ => throw new KeyNotFoundException()
-   };
+   public object? this[string col] {
+      get {
+         switch (col) {
+         <#
+         // Generate case statements for each column name
+         for (int i = 0; i < columns.Count; i++) {
+            var name = columns[i].Item1;
+            var key = columns[i].Item3;
+         #>
+   case "<#= key #>":
+               return <#= name #>;
+         <#
+}
+         #>
+   default:
+               throw new KeyNotFoundException();
+         }
+      }
+   }
 
    // Indexer by column index
-   public object? this[int col] => col switch {
-      <# 
-      // Generate case statements for each column index
-      for (int i = 0; i < columns.Count; i++) {
-          var name = columns[i].Item1;
-      #>
-<#= i #> => <#= name #>,
-      <# 
-} 
-      #>
-_ => throw new IndexOutOfRangeException()
-   };
+   public object? this[int col] {
+      get {
+         switch (col) {
+         <#
+         // Generate case statements for each column name
+         for (int i = 0; i < columns.Count; i++) {
+            var name = columns[i].Item1;
+         #>
+   case <#= i #>:
+               return <#= name #>;
+         <#
+}
+         #>
+   default:
+               throw new KeyNotFoundException();
+         }
+      }
+   }
 
    public bool Equals(<#= TypeName #>? other) =>
       other is not null

src/DataStreamSuite/SharedStreams/RowTemplates/ValueTypeRow.tt

@@ -76,33 +76,45 @@ public <#= propType #> <#= name #> <#= getter #>
    #>
 
    // Indexer by column name
-   public readonly object? this[string col] => col switch {
-      <# 
-      // Generate case statements for each column name
-      for (int i = 0; i < columns.Count; i++) {
-         var name = columns[i].Item1;
-         var key = columns[i].Item3;
-      #>
-"<#= key #>" => <#= name #>,
-      <# 
-} 
-      #>
-_ => throw new KeyNotFoundException()
-   };
+   public readonly object? this[string col] {
+      get {
+         switch (col) {
+         <#
+         // Generate case statements for each column name
+         for (int i = 0; i < columns.Count; i++) {
+            var name = columns[i].Item1;
+            var key = columns[i].Item3;
+         #>
+   case "<#= key #>":
+               return <#= name #>;
+         <#
+}
+         #>
+   default:
+               throw new KeyNotFoundException();
+         }
+      }
+   }
 
    // Indexer by column index
-   public readonly object? this[int col] => col switch {
-      <# 
-      // Generate case statements for each column index
-      for (int i = 0; i < columns.Count; i++) {
-          var name = columns[i].Item1;
-      #>
-<#= i #> => <#= name #>,
-      <# 
-} 
-      #>
-_ => throw new IndexOutOfRangeException()
-   };
+   public readonly object? this[int col] {
+      get {
+         switch (col) {
+         <#
+         // Generate case statements for each column name
+         for (int i = 0; i < columns.Count; i++) {
+            var name = columns[i].Item1;
+         #>
+   case <#= i #>:
+               return <#= name #>;
+         <#
+}
+         #>
+   default:
+               throw new KeyNotFoundException();
+         }
+      }
+   }
 
    public bool Equals(<#= TypeName #> other) =>
    <# 

src/DataStreamSuite/SharedStreams/System/PS3/Subscription/PubSubServerStateStream.cs

@@ -12,8 +12,8 @@ namespace bts {
    /// server.
    /// </summary>
    // TODO: Use dependency injection to instantiate this stream, and others like it that depend on
-   // ISubscriberCore(Factory). Until then, hide it from DSQA.
-   [DataStreamUserRespresentation(/*"pss_state"*/ false)]
+   // ISubscriberCore(Factory).
+   [DataStreamUserRespresentation("pss_state")]
    public class PubSubServerStateStream : ViewStream, IWorker {
       private IDictionary<string, TagPubSubInfo> _tag_map = new Dictionary<string, TagPubSubInfo>();
       private readonly string _tag = "System_AllSubscriptions";
@@ -21,6 +21,9 @@ namespace bts {
       readonly ISubscriberCore _sub_core;
       private ViewStreamDatabase _vsd;
 
+      public PubSubServerStateStream()
+         : this(new SubscriberCoreFactory(), ViewStreamDatabase.instance) { }
+
       public PubSubServerStateStream(ISubscriberCoreFactory subscriberCoreFactory, ViewStreamDatabase vsd)
          : this(subscriberCoreFactory.Create(), vsd) { }
 
@@ -77,7 +80,7 @@ namespace bts {
                DeadPublishers = tsi.dead_publishers_as_string,
             });
          }
-         var dp = BlueDataUtility.build_dp_inc(bt, myLastValue as IDataPacket, true);
+         var dp = BlueDataUtility.build_dp_inc(bt, (myLastValue as IDataPacket)?.newRows, true);
          if (dp.addedRows.Count + dp.deletedRows.Count > 0)
             myFireEvent(dp);
 

src/DataStreamSuite/SharedStreams/System/PS3/Subscription/PubsubHeartbeatStream.cs

@@ -1,81 +1,81 @@
-namespace bts {
-   using BCG;
-   using BCG.Utilities.General;
+namespace bts;
 
-   using Bts.PubSub;
-   using Bts.SharedStreams.Tables;
+using BCG;
+using BCG.Utilities.General;
 
-   // TODO: Use dependency injection to instantiate this stream, and others like it that depend on
-   // ISubscriberCore(Factory). Until then, hide it from DSQA.
-   [DataStreamUserRespresentation(/*"pss_hb"*/ false)]
-    public class PubsubHeartbeatStream : ViewStream, IWorker {
-        private ViewStreamDatabase _vsd;
-      readonly ISubscriberCore _subcore;
-        PubSubHeartbeatRow? _row = null;
+using Bts.PubSub;
+using Bts.SharedStreams.Tables;
 
-      public PubsubHeartbeatStream(ISubscriberCoreFactory subscriberCore, ViewStreamDatabase vsd)
-       : this(subscriberCore.Create(), vsd) { }
+// TODO: Use dependency injection to instantiate this stream, and others like it that depend on
+// ISubscriberCore(Factory).
+[DataStreamUserRespresentation("pss_hb")]
+public class PubsubHeartbeatStream : ViewStream, IWorker {
+   private ViewStreamDatabase _vsd;
+   readonly ISubscriberCore _subcore;
+   PubSubHeartbeatRow? _row = null;
 
-      PubsubHeartbeatStream(ISubscriberCore subscriberCore, ViewStreamDatabase vsd) {
-            _subcore = subscriberCore;
-            _vsd = vsd;
-        }
+   public PubsubHeartbeatStream()
+    : this(new SubscriberCoreFactory(), ViewStreamDatabase.instance) { }
 
-        static IBlueTableSchema _schema => PubSubHeartbeatRow.Schema;
+   public PubsubHeartbeatStream(ISubscriberCoreFactory subscriberCore, ViewStreamDatabase vsd)
+    : this(subscriberCore.Create(), vsd) { }
 
-        protected override void myOnFirstSubscriber() {
-            _subcore.subscribe("pss_hb");
-            _vsd.addWorker(this);
+   PubsubHeartbeatStream(ISubscriberCore subscriberCore, ViewStreamDatabase vsd) {
+      _subcore = subscriberCore;
+      _vsd = vsd;
+   }
 
-        }
+   static IBlueTableSchema _schema => PubSubHeartbeatRow.Schema;
 
-        protected override void myOnLastSubscriber() {
-            _vsd.removeWorker(this);
-            _subcore.unsubscribe("pss_hb");
+   protected override void myOnFirstSubscriber() {
+      _subcore.subscribe("pss_hb");
+      _vsd.addWorker(this);
+   }
 
-        }
+   protected override void myOnLastSubscriber() {
+      _vsd.removeWorker(this);
+      _subcore.unsubscribe("pss_hb");
+   }
 
-        public void doWork() {
-            var updates = _subcore.get_updates(TimeSpan.FromMilliseconds(1));
-            if (updates.Count() > 0) {
-                foreach (var m in updates) {
-                    if (m.data != null && m.data.Length > 0) {
-                        BinaryReader br = new BinaryReader(new MemoryStream(m.data));
-                        Int64 hb_sent_time = br.ReadInt64();
-                        Int64 hb_recv_time = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
-                        // _row = _row == null ? new MutableBlueRow(_schema) : _row;
-                        var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
-                        _row = new() {
-                           Id = 0,
-                           SendTime = hb_sent_time,
-                           RecvTime = hb_recv_time,
-                           Latency = hb_recv_time - hb_sent_time,
-                           RecvElapsed = now - hb_recv_time,
-                        };
-                        // TODO, add to table and fire
-                    }
-                }
-            } else {
-                if (_row != null) {
-                    var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
-                    _row = new() {
-                       Id = _row.Id,
-                       SendTime = _row.SendTime,
-                       RecvTime = _row.RecvTime,
-                       Latency = _row.Latency,
-                       RecvElapsed = now - _row.RecvTime,
-                    };
-                    // TODO, add to table and fire
-                }
-                // update the row
+   public void doWork() {
+      var updates = _subcore.get_updates(TimeSpan.FromMilliseconds(1));
+      if (updates.Count() > 0) {
+         foreach (var m in updates) {
+            if (m.data != null && m.data.Length > 0) {
+               BinaryReader br = new BinaryReader(new MemoryStream(m.data));
+               Int64 hb_sent_time = br.ReadInt64();
+               Int64 hb_recv_time = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+               // _row = _row == null ? new MutableBlueRow(_schema) : _row;
+               var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+               _row = new() {
+                  Id = 0,
+                  SendTime = hb_sent_time,
+                  RecvTime = hb_recv_time,
+                  Latency = hb_recv_time - hb_sent_time,
+                  RecvElapsed = now - hb_recv_time,
+               };
+               // TODO, add to table and fire
             }
-            if (_row != null) {
-                var bt = new BlueTable(_schema);
-                bt.addRow(_row);
-                var dp = BlueDataUtility.build_dp_inc(bt, myLastValue as IDataPacket, true);
-                myFireEvent(dp);
-            }
-        }
-
-    }
+         }
+      } else {
+         if (_row != null) {
+            var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+            _row = new() {
+               Id = _row.Id,
+               SendTime = _row.SendTime,
+               RecvTime = _row.RecvTime,
+               Latency = _row.Latency,
+               RecvElapsed = now - _row.RecvTime,
+            };
+            // TODO, add to table and fire
+         }
+         // update the row
+      }
+      if (_row != null) {
+         var bt = new BlueTable(_schema);
+         bt.addRow(_row);
+         var dp = BlueDataUtility.build_dp_inc(bt, myLastValue as IDataPacket, true);
+         myFireEvent(dp);
+      }
+   }
 }

src/app_sparkd/configurations/bybit_dev/.env

@@ -1 +1,2 @@
 SPARKD__CONFIG_DIR=src/app_sparkd/configurations/bybit_dev
+SPARKD__PYTHON_REQUIREMENTS=src/app_sparkd/configurations/bybit_dev/requirements.txt

src/app_sparkd/configurations/bybit_dev/bybit_overrides.json

@@ -0,0 +1,14 @@
+{
+  "currencies": [
+    "USDT"
+  ],
+  "futures": [
+    "BNB",
+    "BTC",
+    "DUCK",
+    "ETH",
+    "PRIME",
+    "XRP",
+    "MILK"
+  ]
+}

src/app_sparkd/configurations/bybit_dev/requirements.txt

@@ -0,0 +1,5 @@
+numpy==2.2.6
+pybit==5.11.0
+requests==2.31.0
+websocket-client==1.8.0
+websockets==15.0.1

src/app_sparkd/configurations/dev/.env

@@ -1,2 +1,3 @@
 SPARKD__CONFIG_DIR=src/app_sparkd/configurations/dev
+SPARKD__PYTHON_REQUIREMENTS=src/app_sparkd/configurations/dev/requirements.txt
 SPARKD__SHM_SIZE=8gb

src/app_sparkd/configurations/dev/requirements.txt

@@ -0,0 +1,5 @@
+numpy==2.2.6
+pybit==5.11.0
+requests==2.31.0
+websocket-client==1.8.0
+websockets==15.0.1

src/app_sparkd/configurations/gateio_dev/.env

@@ -1 +1,2 @@
 SPARKD__CONFIG_DIR=src/app_sparkd/configurations/py_dummy_md
+SPARKD__PYTHON_REQUIREMENTS=src/app_sparkd/configurations/gateio_dev/requirements.txt

src/app_sparkd/configurations/gateio_dev/gateio_overrides.json

@@ -0,0 +1,6 @@
+{
+  "spots": [
+    "BTC",
+    "ETH"
+  ]
+}

src/app_sparkd/configurations/gateio_dev/requirements.txt

@@ -0,0 +1,5 @@
+numpy==2.2.6
+pybit==5.11.0
+requests==2.31.0
+websocket-client==1.8.0
+websockets==15.0.1

src/app_sparkd/configurations/okx_bin_sim/.env

@@ -0,0 +1,3 @@
+SPARKD__CONFIG_DIR=src/app_sparkd/configurations/okx_bin_sim
+SPARKD__SHM_SIZE=1gb
+SPARKD__PYTHON_REQUIREMENTS=src/app_sparkd/configurations/okx_bin_sim/requirements.txt

src/app_sparkd/configurations/okx_bin_sim/requirements.txt

@@ -0,0 +1,5 @@
+numpy==2.2.6
+pybit==5.11.0
+requests==2.31.0
+websocket-client==1.8.0
+websockets==15.0.1

src/app_sparkd/configurations/okx_bin_sim/spark.conf

@@ -0,0 +1,64 @@
+JetFIX {
+  value {
+    connection {
+      name: "binance_usds_futures_1"
+      venue: VENUE_BINANCE_USDS_FUTURES
+      max_order_msgs_per_window: 300
+      window_duration_msecs: 1000
+    }
+    connection {
+      name: "okx_1"
+      venue: VENUE_OKX
+      max_order_msgs_per_window: 300
+      window_duration_msecs: 1000
+    }
+  }
+}
+
+PublisherOptions {
+  value {
+    network_interface: "0.0.0.0"
+    heartbeat_ms: 5000
+    heartbeat_boot_seconds: 60
+    skip_options: true
+    hot_initial: 1024
+    hot_growth: 1024
+    cold_initial: 8192
+    cold_growth: 2048
+    channel_prefix: ""
+    disable_heartbeat: true
+  }
+}
+
+MarketDataConfig {
+  value {
+    binance_usds_futures {
+      num_depth_levels: 5
+      url_rest_base: "https://fapi.binance.com"
+      url_ws_base: "wss://fstream.binance.com"
+      security_criteria {
+        clauses {
+          security_types: SECURITY_TYPE_FUTURE
+          symbols: @SymbolLists.list[BinanceUsdsFutures].symbol
+        }
+      }
+    }
+     okx {
+       num_depth_levels: 5
+       url_rest_base: "https://www.okx.com"
+       url_ws_base: "wss://ws.okx.com:8443"
+       security_criteria {
+         clauses {
+           security_types: SECURITY_TYPE_FUTURE
+           symbols: @SymbolLists.list[Okx].symbol
+         }
+       }
+       is_application_ping_pong_enabled: true
+       application_ping_interval_ms: 20000
+       application_pong_timeout_ms: 10000
+       is_protocol_ping_pong_enabled: true
+       protocol_ping_interval_ms: 60000
+       protocol_pong_timeout_ms: 30000
+     }
+  }
+}

src/app_sparkd/configurations/okx_bin_sim/symbols.conf

@@ -0,0 +1,37 @@
+SymbolLists {
+  value {
+    list {
+      key: "BinanceUsdsFutures"
+      symbol: "SOL"
+      symbol: "BNB"
+      symbol: "BTC"
+      symbol: "DOGE"
+      symbol: "EIGEN"
+      symbol: "ETH"
+      symbol: "XRP"
+    }
+    list {
+      key: "Okx"
+      symbol: "SOL"
+      symbol: "BNB"
+      symbol: "BTC"
+      symbol: "DOGE"
+      symbol: "EIGEN"
+      symbol: "ETH"
+      symbol: "XRP"
+    }
+  }
+}
+
+Symbols {
+  value {
+    venues {
+      name: "VENUE_BINANCE_USDS_FUTURES"
+      future: @SymbolLists.list[BinanceUsdsFutures].symbol
+    }
+    venues {
+      name: "VENUE_OKX"
+      future: @SymbolLists.list[Okx].symbol
+    }
+  }
+}

src/app_sparkd/configurations/sim_dev/.env

@@ -1,2 +1,3 @@
 SPARKD__CONFIG_DIR=src/app_sparkd/configurations/sim_dev
+SPARKD__PYTHON_REQUIREMENTS=src/app_sparkd/configurations/sim_dev/requirements.txt
 SPARKD__SHM_SIZE=8gb

src/app_sparkd/configurations/sim_dev/requirements.txt

@@ -0,0 +1,5 @@
+numpy==2.2.6
+pybit==5.11.0
+requests==2.31.0
+websocket-client==1.8.0
+websockets==15.0.1

src/client/spark/Spark.csproj

@@ -39,7 +39,7 @@
     <When Condition="'$(UseProjectRefs)' == 'true'">
       <ItemGroup>
         <ProjectReference Include="..\..\cs\Bts.Spark.Api\Bts.Spark.Api.csproj" />
-        <ProjectReference Include="..\..\DataStreamSuite\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj" />
+        <ProjectReference Include="..\..\cs\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj" />
         <ProjectReference Include="..\..\DataStreamSuite\SharedStreams\SharedStreams.csproj" />
       </ItemGroup>
     </When>

src/client/spark_api_server/Program.cs

@@ -1,4 +1,6 @@
-using bts;
+using BCG;
+
+using bts;
 
 using Bts.DataStreamCore.Services;
 using Bts.Extensions.Hosting;
@@ -12,6 +14,7 @@ using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 
 using Spark.Api.Server;
+using Spark.Api.MarketData;
 using Spark.Api.Services;
 using Spark.Core;
 using Spark.Core.Services;
@@ -33,8 +36,14 @@ static Task<int> Run(RunOptions opts) {
    HostBuilderExtensions.ConfigureApp(builder.Configuration, builder.Environment.EnvironmentName, appName);
    ConfigureServices(builder.Services, builder.Configuration);
    var app = builder.Build();
+   var mdClient = app.Services.GetService<IMarketDataClient>();
+   if (mdClient is not null) {
+      PrimitiveManager.SetMarketDataClient(mdClient);
+   }
    app.UseWebSockets(app.Configuration.GetValue<WebSocketOptions>("WebSocketOptions") ?? new() { KeepAliveInterval = TimeSpan.FromMinutes(2) });
    var gateway = new Lazy<ISparkApiWebSocketServer>(app.Services.GetRequiredService<ISparkApiWebSocketServer>);
+   app.UseWebSockets();
+
    app.Map("/", HandleRequest);
    var errors = Errors.None;
    Log.Api.info("Starting websocket server...");
@@ -47,8 +56,15 @@ static Task<int> Run(RunOptions opts) {
    }
    return Task.FromResult((int)errors);
 
-   Task HandleRequest(HttpContext context) =>
-      gateway.Value.HandleRequestAsync(context);
+   async Task HandleRequest(HttpContext context){
+      if (!context.WebSockets.IsWebSocketRequest)
+      {
+         context.Response.StatusCode = StatusCodes.Status400BadRequest;
+         await context.Response.WriteAsync("WebSocket endpoint");
+      }
+
+      await gateway.Value.HandleRequestAsync(context);
+   }
 }
 
 static void ConfigureServices(IServiceCollection services, IConfiguration config) {

src/client/spark_api_server/spark_api_server.csproj

@@ -15,9 +15,21 @@
 
   <ItemGroup>
     <ProjectReference Include="..\spark_websocket\spark_websocket.csproj" />
-    <ProjectReference Include="..\..\DataStreamSuite\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj" />
   </ItemGroup>
 
+  <Choose>
+    <When Condition="'$(UseProjectRefs)' == 'true'">
+      <ItemGroup>
+        <ProjectReference Include="..\..\cs\Bts.Extensions.Hosting\Bts.Extensions.Hosting.csproj" />
+      </ItemGroup>
+    </When>
+    <Otherwise>
+      <ItemGroup>
+        <PackageReference Include="Bts.Extensions.Hosting" Version="$(SparkVersion)" />
+      </ItemGroup>
+    </Otherwise>
+  </Choose>
+
   <ItemGroup>
     <ConfigurationProfile Include="appsettings.json" />
     <ConfigurationProfile Include="appsettings.Development.json" />

src/client/spark_websocket/Core/WebsocketDataSync.cs

@@ -7,24 +7,29 @@ namespace bts.websocket_core {
    using bts.api.synthetic;
    using bts.co2;
    using bts.md;
-
-   using Newtonsoft.Json;
+   using bts.spark;
 
    using Spark.Api.Admin;
    using Spark.Api.MarketData;
    using Spark.Api.Orders;
+   using Spark.Core.Disposables;
    using Spark.Core.Reactive;
 
    using System.Collections.Concurrent;
+   using System.Reactive.Linq;
+   using System.Text.Json;
+   using System.Text.Json.Nodes;
 
    /// <summary>
    /// This does all the bookkeeping for connections and data, and passes
    /// it over to the business layer CURRENT VERSION IS JUST TO SHOW HOW TO USE GATEWAY
    /// </summary>
-   public class WebsocketDataSync : IWebsocketDataSync {
+   public class WebsocketDataSync : IWebsocketDataSync, IDisposable {
       readonly AdminService adminService;
       readonly Spark.Api.IServiceProvider _service_ctx;
       readonly ISecurityResolver securityResolver;
+      IReadOnlyDictionary<uint, ExecutionStrategyImmutable> synthStrategies = new Dictionary<uint, ExecutionStrategyImmutable>(0);
+      IDisposable? synthStrategiesSub;
 
       public WebsocketDataSync(Spark.Api.IServiceProvider service_ctx, IMarketDataClient mdClient, ISecurityResolver securityResolver) {
          _service_ctx = service_ctx;
@@ -204,6 +209,10 @@ namespace bts.websocket_core {
          var synth_sec_service = _service_ctx.GetService<SyntheticSecurityService>();
          var admin_client = new AdminClient(admin_service, UserTuple.admin);
          var md_relay = _service_ctx.GetService<MarketDataRelay>();
+         Log.isnull(synthStrategiesSub);
+         synthStrategiesSub  = synth_sec_service.subscribei_metadata()
+            .Select(upd => upd.all.Where(m => m.HasStrategy).ToDictionary(m => m.SecurityKey, m => m.Strategy))
+            .Subscribe(ss => synthStrategies = ss);
          Orders_Client = new OrdersClient(
             order_service, position_service, admin_client, securityResolver, TimeSpan.FromSeconds(0.1));
          var r  = await Orders_Client.GetRegisteredDynamicOrderTypesAsync();
@@ -318,7 +327,7 @@ namespace bts.websocket_core {
       public async void handle_get(WebsocketRequest req) { await HandleRestRequest(req,WebsocketRestHandler.Get);}
       public async void handle_enter_order(WebsocketRequest req) {
          await HandleRestRequest(req, (req, Orders_Client) => WebsocketRestHandler.EnterOrderAsync(
-         req, Orders_Client, _securities, adminService, Dynamic_Descriptors
+         req, Orders_Client, _securities, adminService, Dynamic_Descriptors, synthStrategies
       ));
       } 
       public async void handle_update_order(WebsocketRequest req) { await HandleRestRequest(req,(req, Orders_Client) => WebsocketRestHandler.UpdateOrderAsync(req,Orders_Client,Dynamic_Descriptors));}
@@ -366,7 +375,7 @@ namespace bts.websocket_core {
          send_message(ms, cid);
       }
       public void send_message(object value, WebsocketClientID id) {
-         var requestJson = JsonConvert.SerializeObject(value);
+         var requestJson = System.Text.Json.JsonSerializer.Serialize(value);
          Gateway.send_to_clients(
             new List<WebsocketOutgoingMessage>() {
                 new WebsocketOutgoingMessage(id, requestJson)
@@ -374,5 +383,8 @@ namespace bts.websocket_core {
 
       }
 
+      public void Dispose() {
+         Disposable.Dispose(ref synthStrategiesSub);
+      }
    }
 }

src/client/spark_websocket/Core/WebsocketJsonParsers.cs

@@ -3,11 +3,10 @@ namespace bts.websocket_core {
    using bts.api.core;
    using bts.api.orderman;
 
-   using Newtonsoft.Json;
-   using Newtonsoft.Json.Linq;
-
    using System.Diagnostics.CodeAnalysis;
    using System.Globalization;
+   using System.Text.Json;
+   using System.Text.Json.Nodes;
 
    public class WebsocketJsonParsers {
       public static Object to_json<T, TKey>(TableUpdate<T, TKey> upd, Func<T, Object> conv) where TKey : IEquatable<TKey> {
@@ -78,7 +77,7 @@ namespace bts.websocket_core {
          public static bool TryParse(string input, [NotNullWhen(true)] out PrecomputedTheoPointWrapper? P) {
             P = null;
             try {
-               var temp = JsonConvert.DeserializeObject<PrecomputedTheoPointWrapper>(input);
+               var temp = JsonSerializer.Deserialize<PrecomputedTheoPointWrapper>(input);
                P = temp;
                return true;
             } catch {
@@ -111,7 +110,7 @@ namespace bts.websocket_core {
          public static bool TryParse(string input, [NotNullWhen(true)] out StringListWrapper? l) {
             l = null;
             try {
-               var temp = JsonConvert.DeserializeObject<List<Object>>(input);
+               var temp = JsonSerializer.Deserialize<List<object>>(input);
                if (temp == null) {
                   return false;
                }
@@ -139,7 +138,7 @@ namespace bts.websocket_core {
          }
 
       }
-      public static bool TryParse<T>(JObject filter, string key, out T[]? list, out string error) where T : struct {
+      public static bool TryParse<T>(JsonObject filter, string key, out T[]? list, out string error) where T : struct {
          list = null;
          if (TryParse<StringListWrapper>(filter, key, out StringListWrapper? wrap, out error)) {
             if (wrap != null) {
@@ -157,7 +156,7 @@ namespace bts.websocket_core {
          }
          return false;
       }
-      public static bool TryParse<T>(JObject filter, string key, out T? obj, [NotNullWhen(false)] out string error) where T : struct {
+      public static bool TryParse<T>(JsonObject filter, string key, out T? obj, [NotNullWhen(false)] out string error) where T : struct {
          obj = null;
          var value = filter.GetValueOrDefault(key)?.ToString();
          if (value == null) {

src/client/spark_websocket/Core/WebsocketRequest.cs

@@ -1,6 +1,6 @@
-using Newtonsoft.Json.Linq;
-
 using System.Diagnostics.CodeAnalysis;
+using System.Text.Json;
+using System.Text.Json.Nodes;
 
 namespace bts.websocket_core {
 
@@ -22,20 +22,24 @@ namespace bts.websocket_core {
       }
       public readonly WebsocketRequestID id;
       public readonly WebsocketClientID client_id;
-      public readonly JObject parameters;
+      public readonly JsonObject parameters;
       public readonly REQUEST_TYPES type;
 
-      private static bool TryParseJson(string json, out string _message_id, out REQUEST_TYPES type, [NotNullWhen(true)] out JObject? para, [NotNullWhen(false)] out string message) {
+      static bool TryParseJson(string json, out string _message_id, out REQUEST_TYPES type, [NotNullWhen(true)] out JsonObject? para, [NotNullWhen(false)] out string message) {
          para = null;
          type = REQUEST_TYPES.UNKOWN;
-         JObject request;
+         JsonObject? request;
          _message_id = "missing";
          try {
-            request = JObject.Parse(json);
-         } catch (Newtonsoft.Json.JsonReaderException ex) {
+            request = JsonSerializer.Deserialize<JsonObject>(json);
+         } catch (JsonException ex) {
             message = $"Incorrect Json: {ex.Path}";
             return false;
          }
+         if (request is null) {
+            message = "JsonSerializer.Deserialize returned null";
+            return false;
+         }
          if (!WebsocketJsonParsers.TryParse<WebsocketJsonParsers.StringWrapper>(
             request,
             "id",
@@ -50,11 +54,11 @@ namespace bts.websocket_core {
             return false;
          }
          type = rt ?? REQUEST_TYPES.UNKOWN;
-         para = (JObject?)request["params"] ?? new JObject();
+         para = (JsonObject?)request["params"] ?? new();
          return true;
       }
       public static bool TryGetWebSocketRequest(WebsocketClientID cid, string json, out IUniqueID id, [NotNullWhen(true)] out WebsocketRequest? wr, out string message) {
-         JObject? jobj;
+         JsonObject? jobj;
          REQUEST_TYPES rt;
          if (TryParseJson(json, out string _message_id, out rt, out jobj, out message)) {
             wr = new WebsocketRequest(cid, new WebsocketRequestID(_message_id), jobj, rt);
@@ -68,15 +72,15 @@ namespace bts.websocket_core {
       private WebsocketRequest(
          WebsocketClientID _cid,
          WebsocketRequestID _id,
-         JObject jobj,
+         JsonObject jobj,
          REQUEST_TYPES rt
          ) {
          (client_id, id, parameters, type) = (_cid, _id, jobj, rt);
       }
-      public WebsocketRequest(WebsocketRequest other, JObject para) {
+      public WebsocketRequest(WebsocketRequest other, JsonObject para) {
          client_id = other.client_id;
          id = other.id;
          (parameters, type) = (para, other.type);
       }
    }
-}
+}

src/client/spark_websocket/Core/WebsocketRestHandler.cs

@@ -3,28 +3,21 @@ namespace bts.websocket_core {
    using bts.api.admin;
    using bts.api.core;
    using bts.api.orderman;
-
-   using mbts;
-
-   using Microsoft.VisualBasic;
-
-   using mbts.spark;
-   using Newtonsoft.Json.Linq;
+   using bts.spark;
 
    using Spark.Api.Orders;
 
    using System;
    using System.Diagnostics.CodeAnalysis;
-   using bts.spark;
-   using bts.utils;
+   using System.Text.Json.Nodes;
 
    public class WebsocketRestHandler {
-      public static (bool, DynamicOrderParam?, string) TryGetDynamicParam(JObject custom, string key, DynamicOrderParamDescriptor descriptor, IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors) {
+      public static (bool, DynamicOrderParam?, string) TryGetDynamicParam(JsonObject custom, string key, DynamicOrderParamDescriptor descriptor, IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors) {
          var rv = new DynamicOrderParam();
          string error = "";
          switch (descriptor.ParamType.DescriptorCase) {
             case DynamicOrderTypeDescriptor.DescriptorOneofCase.Array:
-               if (custom[key] is not JArray arr_obj) {
+               if (custom[key] is not JsonArray arr_obj) {
                   return (false, null, key + " not correct");
                }
                rv.AsArray = new DynamicOrderArray();
@@ -35,7 +28,7 @@ namespace bts.websocket_core {
                      ParamType = descriptor.ParamType.Array.ElementType
                   };
 
-                  JObject child = new() {
+                  JsonObject child = new() {
                      [child_key] = p
                   };
 
@@ -69,7 +62,7 @@ namespace bts.websocket_core {
                      rv.AsInt32 = (int)i;
                      return (true, rv, error);
                   case DynamicOrderPrimitiveDescriptor.DynamicOrderPrimitiveOrderDef:
-                     if (custom[key] is not JObject child_custom) {
+                     if (custom[key] is not JsonObject child_custom) {
                         return (false, null, key + " not correct");
                      }
                      (bool suc, var child_custom_def, error) = TryGetCustomDef(child_custom, Dynamic_Descriptors);
@@ -108,13 +101,12 @@ namespace bts.websocket_core {
                }
                return (false, null, key + " not correct");
             case DynamicOrderTypeDescriptor.DescriptorOneofCase.Structure:
-               if (custom[key] is not JObject obj) {
+               if (custom[key] is not JsonObject obj) {
                   return (false, null, key + " not correct");
                }
                rv.AsStruct = new DynamicOrderStruct();
 
-               foreach (var prop in obj.Properties()) {
-                  var child_key = prop.Name;
+               foreach (var (child_key, _) in obj) {
                   var check = descriptor.ParamType.Structure.Field.TryGetFirst(v => v?.ParamName == child_key, out DynamicOrderParamDescriptor? p);
                   if (!check || p == null) {
                      return (false, null, key + " " + child_key + " not correct");
@@ -135,13 +127,13 @@ namespace bts.websocket_core {
          return (false, null, key + " not correct");
       }
 
-      private static (bool, CustomOrderDefinition?, string) TryGetCustomDef(WebsocketRequest req, IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors) {
-         if (req.parameters["options"] is not JObject custom) {
+      private static (bool, CustomOrderDefinition?, string) TryGetCustomDef(WebsocketRequest req, IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors, SecurityImmutable? sec=null,  ExecutionStrategyImmutable? synth_Strats =null) {
+         if (req.parameters["options"] is not JsonObject custom) {
             return (false, null, "options not correct");
          }
-         return TryGetCustomDef(custom, Dynamic_Descriptors);
+         return TryGetCustomDef(custom, Dynamic_Descriptors, sec, synth_Strats);
       }
-      private static (bool, CustomOrderDefinition?, string) TryGetCustomDef(JObject custom, IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors) {
+      private static (bool, CustomOrderDefinition?, string) TryGetCustomDef(JsonObject custom, IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors, SecurityImmutable? sec =null, ExecutionStrategyImmutable? synth_Strats =null) {
          CustomOrderDefinition customdef;
 
          if (!WebsocketJsonParsers.TryParse<CustomOrderDefinition.CustomOneofCase>(custom, "order_type", out CustomOrderDefinition.CustomOneofCase? order_type, out string status) || order_type == null) {
@@ -169,7 +161,7 @@ namespace bts.websocket_core {
                   if (!WebsocketJsonParsers.TryParse<double>(custom, "show_quantity", out double? sq, out status) || sq == null) {
                      return (false, null, status);
                   }
-                  if (custom["child"] is not JObject child) {
+                  if (custom["child"] is not JsonObject child) {
                      return (false, null, "Iceberg child not correct");
                   }
                   (var check, var childdef, status) = TryGetCustomDef(child, Dynamic_Descriptors);
@@ -189,11 +181,17 @@ namespace bts.websocket_core {
                   return (false, null, "timebox not enabled on api");
                }
             case CustomOrderDefinition.CustomOneofCase.Spreader: {
-                  // if (custom == null) {
-                  //    message = "custom not present";
-                  //    return false;
-                  // }
-                  return (false, null, "spreaders not enabled on api");
+                  if(sec is null)
+                     return (false, null, "Could not find Security");
+                  if(synth_Strats is null)
+                     return (false, null, "Could not find Synthetic Strat");
+                   customdef = new CustomOrderDefinition() {
+                     Spreader = new SpreaderOrderDefinition() {
+                         Security = sec.Value.CloneToMutable(),
+                         ExecutionStrategy = synth_Strats.Value.CloneToMutable()
+                     }
+                  };
+                  break;
                }
             case CustomOrderDefinition.CustomOneofCase.Tick: {
                   return (false, null, "Ticks not enabled on api");
@@ -240,14 +238,14 @@ namespace bts.websocket_core {
                      ) {
                      return (false, null, status);
                   }
-                  if (custom["child"] is not JObject child) {
+                  if (custom["child"] is not JsonObject child) {
                      return (false, null, "Cover child not correct");
                   }
                   (var check, var childdef, status) = TryGetCustomDef(child, Dynamic_Descriptors);
                   if (!check) {
                      return (false, null, status);
                   }
-                  if (custom["hedge"] is not JObject hedge) {
+                  if (custom["hedge"] is not JsonObject hedge) {
                      return (false, null, "Cover hedge not correct");
                   }
                   (check, var hedgedef, status) = TryGetCustomDef(hedge, Dynamic_Descriptors);
@@ -270,7 +268,7 @@ namespace bts.websocket_core {
                   if (!WebsocketJsonParsers.TryParse<uint>(custom, "tick_offset", out uint? to, out status) || to == null) {
                      return (false, null, status);
                   }
-                  if (custom["child"] is not JObject child) {
+                  if (custom["child"] is not JsonObject child) {
                      return (false, null, "offset child not correct");
                   }
                   (var check, var childdef, status) = TryGetCustomDef(child, Dynamic_Descriptors);
@@ -371,7 +369,8 @@ namespace bts.websocket_core {
          WebsocketRequest req,
          IReadOnlyDictionary<uint, SecurityImmutable> securities,
          AdminService adminService,
-         IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors
+         IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors,
+         IReadOnlyDictionary<uint, ExecutionStrategyImmutable> Synth_Strats
       ) {
          if (req.parameters == null) {
             return (false, null, "parameters not present");
@@ -381,7 +380,7 @@ namespace bts.websocket_core {
             !WebsocketJsonParsers.TryParse<Venue>(req.parameters, "venue", out Venue? exchange, out status) || exchange == null ||
             !WebsocketJsonParsers.TryParse<WebsocketJsonParsers.StringWrapper>(req.parameters, "account", out WebsocketJsonParsers.StringWrapper? act, out status) || act == null ||
             !WebsocketJsonParsers.TryParse<Side>(req.parameters, "side", out Side? side, out status) || side == null ||
-            !WebsocketJsonParsers.TryParse<WebsocketJsonParsers.StringWrapper>(req.parameters, "exchange_security_name", out WebsocketJsonParsers.StringWrapper? exchange_securty_name, out status) || exchange_securty_name == null
+            !WebsocketJsonParsers.TryParse<WebsocketJsonParsers.StringWrapper>(req.parameters, "exchange_security_name", out WebsocketJsonParsers.StringWrapper? exchange_security_name, out status) || exchange_security_name == null
             ) {
             return (false, null, status);
          }
@@ -411,13 +410,14 @@ namespace bts.websocket_core {
          if (!_securites.Values.TryGetFirst(sec =>
             (
                sec.Exchange == exchange &&
-               sec.ToString() == exchange_securty_name.Value.Val),
+               sec.ToString() == exchange_security_name.Value.Val),
             out var sec
             )) {
             return (false, null, "security not found");
          }
+         Synth_Strats.TryGetValue(sec.Id,out var strat);
 
-         (var check, var customdef, status) = TryGetCustomDef(req, Dynamic_Descriptors);
+         (var check, var customdef, status) = TryGetCustomDef(req, Dynamic_Descriptors, sec, strat);
          if (check == false) {
             return (false, null, status);
          }
@@ -496,17 +496,18 @@ namespace bts.websocket_core {
          status = res.ErrorCode.ToString();
          return (res.ErrorCode != OrderError.Ok, null, status);
       }
+      
       public static async Task<(bool success, uint? orderId, string status)> EnterOrderAsync(
          WebsocketRequest request,
          OrdersClient ordersClient,
          IReadOnlyDictionary<uint, SecurityImmutable> securities,
          AdminService adminService,
-         IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors) {
+         IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors, IReadOnlyDictionary<uint, ExecutionStrategyImmutable> execution_strats) {
          OrderDefinition? orderDefinition = null;
          uint? orderId = null;
          string status = string.Empty;
 
-         (bool success, orderDefinition, status) = await TryGetOrderDefAsync(request, securities, adminService, Dynamic_Descriptors);
+         (bool success, orderDefinition, status) = await TryGetOrderDefAsync(request, securities, adminService, Dynamic_Descriptors, execution_strats);
 
          if (
             !success || orderDefinition == null ||
@@ -515,7 +516,39 @@ namespace bts.websocket_core {
             return (false, null, status);
          }
 
-         var result = await ordersClient.TryEnterOrderAsync(orderDefinition, (uint)price, (double)quantity);
+         var result = await ordersClient.TryEnterOrderAsync(orderDefinition, (double)price, (double)quantity);
+
+         if (result.ErrorCode != OrderError.Ok) {
+            return (false, null, result.ErrorCode.ToString());
+         }
+
+         orderId = result.OrderKey;
+         status = result.ErrorCode.ToString();
+
+         return (true, orderId, status);
+      }
+
+
+      public static async Task<(bool success, uint? orderId, string status)> CreateSynthAsync(
+         WebsocketRequest request,
+         OrdersClient ordersClient,
+         IReadOnlyDictionary<uint, SecurityImmutable> securities,
+         AdminService adminService,
+         IReadOnlyDictionary<string, DynamicOrderDescriptor> Dynamic_Descriptors, IReadOnlyDictionary<uint, ExecutionStrategyImmutable> execution_strats) {
+         OrderDefinition? orderDefinition = null;
+         uint? orderId = null;
+         string status = string.Empty;
+
+         (bool success, orderDefinition, status) = await TryGetOrderDefAsync(request, securities, adminService, Dynamic_Descriptors, execution_strats);
+
+         if (
+            !success || orderDefinition == null ||
+            !WebsocketJsonParsers.TryParse(request.parameters, "quantity", out double? quantity, out status) || quantity == null ||
+            !WebsocketJsonParsers.TryParse(request.parameters, "price", out double? price, out status) || price == null) {
+            return (false, null, status);
+         }
+
+         var result = await ordersClient.TryEnterOrderAsync(orderDefinition, (double)price, (double)quantity);
 
          if (result.ErrorCode != OrderError.Ok) {
             return (false, null, result.ErrorCode.ToString());
@@ -561,13 +594,13 @@ namespace bts.websocket_core {
 
          try {
             // Extract parameters
-            if (!req.parameters.TryGetValue("path", out var pathObj) || pathObj == null) {
+            if (!req.parameters.TryGetPropertyValue("path", out var pathObj) || pathObj == null) {
                return (false, null, "file path missing in parameters");
             }
 
             string filePath = pathObj.ToString()!;
             int lines = 10; // default
-            if (req.parameters.TryGetValue("lines", out var linesObj) && int.TryParse(linesObj?.ToString(), out int parsed)) {
+            if (req.parameters.TryGetPropertyValue("lines", out var linesObj) && int.TryParse(linesObj?.ToString(), out int parsed)) {
                lines = parsed;
             }
 

src/client/spark_websocket/Core/WebsocketSubscriptionHandler.cs

@@ -10,8 +10,6 @@ namespace bts.websocket_core {
 
    using mbts.spark;
 
-   using Newtonsoft.Json.Linq;
-
    using Spark.Api.Data;
    using Spark.Api.MarketData;
    using Spark.Api.Orders;
@@ -19,7 +17,7 @@ namespace bts.websocket_core {
 
    using System;
    using System.Diagnostics.CodeAnalysis;
-
+   using System.Text.Json.Nodes;
    using System.Text.RegularExpressions;
 
 
@@ -118,7 +116,7 @@ namespace bts.websocket_core {
             void OnError(Exception e) {
                bts.Log.Gen.error(e.Message);
             }
-            void OnData(DataPacket dp) {
+            void OnData(IDataPacket dp) {
                lock (syncRoot) {
                   removedRows.Clear();
                   changedRows.Clear();
@@ -140,8 +138,8 @@ namespace bts.websocket_core {
                      changed = changedRows.Values,
                      removed = removedRows.Values,
                      metaData = new {
-                        dsl = dp.meta_data?["dsl"],
-                        dp.schema
+                        dsl = dp.Metadata?["dsl"],
+                        schema = dp.Schema,
                      }
                   });
                }
@@ -159,7 +157,7 @@ namespace bts.websocket_core {
 
       static bool TryGetTradesFilter(WebsocketRequest req, [NotNullWhen(true)] out TradesSubscriptionFilter? trades_filter, out string message) {
          trades_filter = null;
-         JObject? filter = req.parameters?["filter"] as JObject;
+         var filter = req.parameters?["filter"]?.AsObject();
          if (filter == null) {
             message = "filter not present";
             return false;
@@ -172,7 +170,7 @@ namespace bts.websocket_core {
       }
       static bool TryGetFillsFilter(WebsocketRequest req, [NotNullWhen(true)] out FillsSubscriptionFilter? order_filter, out string message) {
          order_filter = null;
-         if (req.parameters?["filter"] is not JObject filter) {
+         if (req.parameters?["filter"] is not JsonObject filter) {
             message = "filter not present";
             return false;
          }
@@ -189,7 +187,7 @@ namespace bts.websocket_core {
       }
       static bool TryGetOrdersFilter(WebsocketRequest req, [NotNullWhen(true)] out OrdersSubscriptionFilter? order_filter, out string message) {
          order_filter = null;
-         if (req.parameters?["filter"] is not JObject filter) {
+         if (req.parameters?["filter"] is not JsonObject filter) {
             message = "filter not present";
             return false;
          }
@@ -234,7 +232,7 @@ namespace bts.websocket_core {
       ) {
          stream = null;
          error = "";
-         var query = (req.parameters?["query"] as JValue)?.Value<string>();
+         var query = req.parameters?["query"]?.GetValue<string>();
          if (query is null) {
             error = "'params.query' field not found (expected a string)";
             return false;
@@ -274,7 +272,7 @@ namespace bts.websocket_core {
 
       static bool TryGetFilter(WebsocketRequest req, [NotNullWhen(true)] out SecurityCriteria? cri, out string message) {
          cri = null;
-         if (req.parameters?["filter"] is not JObject filter) {
+         if (req.parameters?["filter"] is not JsonObject filter) {
             message = "filter not present";
             return false;
          }

src/client/spark_websocket/spark_websocket.csproj

@@ -15,14 +15,24 @@
 
   <ItemGroup>
     <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
-    <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
-    <PackageReference Include="Microsoft.AspNetCore.Cors" Version="2.2.0" />
+    <PackageReference Include="Microsoft.AspNetCore.Cors" Version="2.3.9" />
   </ItemGroup>
 
-  <ItemGroup>
-    <ProjectReference Include="..\..\cs\Bts.Spark.Api\Bts.Spark.Api.csproj" />
-    <ProjectReference Include="..\..\DataStreamSuite\DataStream\DataStream.csproj" />
-    <ProjectReference Include="..\..\DataStreamSuite\SharedStreams\SharedStreams.csproj" />
-  </ItemGroup>
+  <Choose>
+    <When Condition="'$(UseProjectRefs)' == 'true'">
+      <ItemGroup>
+        <ProjectReference Include="..\..\cs\Bts.Spark.Api\Bts.Spark.Api.csproj" />
+        <ProjectReference Include="..\..\DataStreamSuite\DataStream\DataStream.csproj" />
+        <ProjectReference Include="..\..\DataStreamSuite\SharedStreams\SharedStreams.csproj" />
+      </ItemGroup>
+    </When>
+    <Otherwise>
+      <ItemGroup>
+        <PackageReference Include="Bts.Spark.Api" Version="$(SparkVersion)" />
+        <PackageReference Include="DataStream" Version="$(SparkVersion)" />
+        <PackageReference Include="SharedStreams" Version="$(SparkVersion)" />
+      </ItemGroup>
+    </Otherwise>
+  </Choose>
 
 </Project>