Compare commits

..
123 changed files with 1711 additions and 6011 deletions

No files matched your search

+2 -2
View File
@@ -47,8 +47,8 @@ trace_data/
/src/.clang-format
/src/server/sparkd/cmds/spartan/analysis/.ipynb_checkpoints
/src/server/sparkd/cmds/spartan/analysis/data
.tmp/
.venv*/
*.tmp/*
/.venv*/*
.cache/
/src/Spark.Admin.WebApp/tsconfig.*.g.json
/src/Spark.Admin.WebApp/src/bin/
+3 -3
View File
@@ -29,8 +29,8 @@
"CMAKE_TOOLCHAIN_FILE": "${sourceDir}/vcpkg/scripts/buildsystems/vcpkg.cmake",
"EXTERNALS_ROOT": "${sourceDir}/externals/windows",
"CONFIG_DEFAULTS_PATH": "config/profiles/default/*.conf",
"DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net8.0",
"DOTNET_WINDOWS_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net8.0-windows"
"DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net7.0",
"DOTNET_WINDOWS_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net7.0-windows"
}
},
@@ -197,7 +197,7 @@
"CMAKE_CXX_FLAGS_RELEASE": "-fPIC -Wall -pedantic -Wextra -Werror -Wno-missing-braces -Wno-ignored-qualifiers --std=c++20 -fdiagnostics-color -fno-omit-frame-pointer -DNDEBUG -O3",
"CMAKE_INSTALL_PREFIX": "${sourceDir}/install/${presetName}",
"CONFIG_DEFAULTS_PATH": "config/profiles/default/*.conf",
"DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net8.0"
"DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net7.0"
}
},
+8 -12
View File
@@ -37,12 +37,8 @@ RUN apt-get update && apt-get install -y \
&& python3 -m pip install --upgrade pip
# dotnet
RUN \
curl -fsSLO "https://packages.microsoft.com/config/ubuntu/$(lsb_release -rs)/packages-microsoft-prod.deb" \
&& dpkg -i packages-microsoft-prod.deb \
&& rm packages-microsoft-prod.deb \
&& apt-get update \
&& apt-get install -y dotnet-sdk-8.0
RUN apt-get update && apt-get install -y \
dotnet7
# ssh
RUN apt-get update && apt-get install -y \
@@ -106,12 +102,12 @@ RUN apt-get update && apt-get install -y \
# set up default executables (update-alternatives)
RUN \
update-alternatives --install /usr/bin/clang++ clang++ /usr/bin/clang++-15 100; \
update-alternatives --install /usr/bin/clang clang /usr/bin/clang-15 100; \
update-alternatives --install /usr/bin/clang-format clang-format /usr/bin/clang-format-15 100; \
update-alternatives --install /usr/bin/clangd clangd /usr/bin/clangd-15 100; \
update-alternatives --install /usr/bin/g++ g++ /usr/bin/g++-12 100; \
update-alternatives --install /usr/bin/gcc gcc /usr/bin/gcc-12 100
update-alternatives --install /usr/bin/clang++ clang++ /usr/bin/clang++-13 100; \
update-alternatives --install /usr/bin/clang clang /usr/bin/clang-13 100; \
update-alternatives --install /usr/bin/clang-format clang-format /usr/bin/clang-format-13 100; \
update-alternatives --install /usr/bin/clangd clangd /usr/bin/clangd-13 100; \
update-alternatives --install /usr/bin/g++ g++ /usr/bin/g++-11 100; \
update-alternatives --install /usr/bin/gcc gcc /usr/bin/gcc-11 100
# Add compiler symlinks for ccache
RUN \
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0-windows</TargetFramework>
<TargetFramework>net7.0-windows</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<UseWindowsForms>true</UseWindowsForms>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,20 +0,0 @@
module.exports = {
root: true,
env: { browser: true, es2020: true },
extends: [
'eslint:recommended',
'plugin:react/recommended',
'plugin:react/jsx-runtime',
'plugin:react-hooks/recommended',
],
ignorePatterns: ['dist', '.eslintrc.cjs'],
parserOptions: { ecmaVersion: 'latest', sourceType: 'module' },
settings: { react: { version: '18.2' } },
plugins: ['react-refresh'],
rules: {
'react-refresh/only-export-components': [
'warn',
{ allowConstantExport: true },
],
},
}
@@ -1,24 +0,0 @@
# Logs
logs
*.log
npm-debug.log*
yarn-debug.log*
yarn-error.log*
pnpm-debug.log*
lerna-debug.log*
node_modules
dist
dist-ssr
*.local
# Editor directories and files
.vscode/*
!.vscode/extensions.json
.idea
.DS_Store
*.suo
*.ntvs*
*.njsproj
*.sln
*.sw?
@@ -1,11 +0,0 @@
<Project Sdk="Microsoft.VisualStudio.JavaScript.Sdk/0.5.128-alpha">
<PropertyGroup>
<StartupCommand>npm run dev</StartupCommand>
<JavaScriptTestRoot>src\</JavaScriptTestRoot>
<JavaScriptTestFramework>Jest</JavaScriptTestFramework>
<!-- Allows the build (or compile) script located on package.json to run on Build -->
<ShouldRunBuildScript>false</ShouldRunBuildScript>
<!-- Folder where production build objects will be placed -->
<PublishAssetsDirectory>$(MSBuildProjectDirectory)\dist</PublishAssetsDirectory>
</PropertyGroup>
</Project>
@@ -1,8 +0,0 @@
# React + Vite
This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
Currently, two official plugins are available:
- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react/README.md) uses [Babel](https://babeljs.io/) for Fast Refresh
- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react-swc) uses [SWC](https://swc.rs/) for Fast Refresh
@@ -1,13 +0,0 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<link rel="icon" type="image/svg+xml" href="/vite.svg" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Vite + React</title>
</head>
<body>
<div id="root"></div>
<script type="module" src="/src/main.jsx"></script>
</body>
</html>
@@ -1,10 +0,0 @@
<?xml version="1.0" encoding="utf-8"?>
<configuration>
<packageSources>
<clear />
<add key="nuget.org" value="https://api.nuget.org/v3/index.json" />
</packageSources>
<disabledPackageSources>
<clear />
</disabledPackageSources>
</configuration>
File diff suppressed because it is too large. Load diff
@@ -1,26 +0,0 @@
{
"name": "bts.datastreamwebapp.client",
"private": true,
"version": "0.0.0",
"type": "module",
"scripts": {
"dev": "vite",
"build": "vite build",
"lint": "eslint . --ext js,jsx --report-unused-disable-directives --max-warnings 0",
"preview": "vite preview"
},
"dependencies": {
"react": "^18.2.0",
"react-dom": "^18.2.0"
},
"devDependencies": {
"@types/react": "^18.2.43",
"@types/react-dom": "^18.2.17",
"@vitejs/plugin-react": "^4.2.1",
"eslint": "^8.55.0",
"eslint-plugin-react": "^7.33.2",
"eslint-plugin-react-hooks": "^4.6.0",
"eslint-plugin-react-refresh": "^0.4.5",
"vite": "^5.0.8"
}
}
@@ -1 +0,0 @@
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" aria-hidden="true" role="img" class="iconify iconify--logos" width="31.88" height="32" preserveAspectRatio="xMidYMid meet" viewBox="0 0 256 257"><defs><linearGradient id="IconifyId1813088fe1fbc01fb466" x1="-.828%" x2="57.636%" y1="7.652%" y2="78.411%"><stop offset="0%" stop-color="#41D1FF"></stop><stop offset="100%" stop-color="#BD34FE"></stop></linearGradient><linearGradient id="IconifyId1813088fe1fbc01fb467" x1="43.376%" x2="50.316%" y1="2.242%" y2="89.03%"><stop offset="0%" stop-color="#FFEA83"></stop><stop offset="8.333%" stop-color="#FFDD35"></stop><stop offset="100%" stop-color="#FFA800"></stop></linearGradient></defs><path fill="url(#IconifyId1813088fe1fbc01fb466)" d="M255.153 37.938L134.897 252.976c-2.483 4.44-8.862 4.466-11.382.048L.875 37.958c-2.746-4.814 1.371-10.646 6.827-9.67l120.385 21.517a6.537 6.537 0 0 0 2.322-.004l117.867-21.483c5.438-.991 9.574 4.796 6.877 9.62Z"></path><path fill="url(#IconifyId1813088fe1fbc01fb467)" d="M185.432.063L96.44 17.501a3.268 3.268 0 0 0-2.634 3.014l-5.474 92.456a3.268 3.268 0 0 0 3.997 3.378l24.777-5.718c2.318-.535 4.413 1.507 3.936 3.838l-7.361 36.047c-.495 2.426 1.782 4.5 4.151 3.78l15.304-4.649c2.372-.72 4.652 1.36 4.15 3.788l-11.698 56.621c-.732 3.542 3.979 5.473 5.943 2.437l1.313-2.028l72.516-144.72c1.215-2.423-.88-5.186-3.54-4.672l-25.505 4.922c-2.396.462-4.435-1.77-3.759-4.114l16.646-57.705c.677-2.35-1.37-4.583-3.769-4.113Z"></path></svg>

Before

Width:  |  Height:  |  Size: 1.5 KiB

@@ -1,23 +0,0 @@
#root {
max-width: 1280px;
margin: 0 auto;
padding: 2rem;
text-align: center;
--cell-background-color: #293136;
--cell-color: #cbe3ef;
}
tr:nth-child(even) {
background: #F2F2F2;
}
tr:nth-child(odd) {
background: #FFF;
}
th, td {
padding-left: 1rem;
padding-right: 1rem;
background-color: var(--cell-background-color);
color: var(--cell-color);
}
@@ -1,155 +0,0 @@
import { useEffect, useState, Component } from 'react';
import './App.css';
function App() {
const [forecasts, setForecasts] = useState();
useEffect(() => {
populateWeatherData();
}, []);
const contents = forecasts === undefined
? <p><em>Loading... Please refresh once the ASP.NET backend has started. See <a href="https://aka.ms/jspsintegrationreact">https://aka.ms/jspsintegrationreact</a> for more details.</em></p>
: <table className="table table-striped" aria-labelledby="tabelLabel">
<thead>
<tr>
<th>Date</th>
<th>Temp. (C)</th>
<th>Temp. (F)</th>
<th>Summary</th>
</tr>
</thead>
<tbody>
{forecasts.map(forecast =>
<tr key={forecast.date}>
<td>{forecast.date}</td>
<td>{forecast.temperatureC}</td>
<td>{forecast.temperatureF}</td>
<td>{forecast.summary}</td>
</tr>
)}
</tbody>
</table>;
return (
<div>
<h1 id="tabelLabel">Weather forecast</h1>
<p>This component demonstrates fetching data from the server.</p>
{contents}
</div>
);
async function populateWeatherData() {
const response = await fetch('weatherforecast');
const data = await response.json();
setForecasts(data);
}
}
export default App;
export class Home extends Component {
static displayName = Home.name;
// static sparkdHost = '192.168.50.94';
static sparkdHost = 'localhost';
// static websocket_url = 'wss://markets.api.uat.cmegroup.com/marketdatastream/v1';
// static access_token = 'eyJhbGciOiJSUzI1NiIsImtpZCI6IjF0al9MTVFrT2pGVzJWQTZvdkZzOEZHUWViMCIsInBpLmF0bSI6IjEifQ.eyJzY29wZSI6W10sImNsaWVudF9pZCI6ImFwaV90aGlua2FscGhhMzIxIiwiYXVkIjoiTnIiLCJqdGkiOiIweERtb3NnU2NHcE5GUEFaIiwic3ViIjoiYXBpX3RoaW5rYWxwaGEzMjEiLCJleHAiOjE2OTgzNTY5OTJ9.uJJEhWpujGiyJ5_lypiIPB8xR0GkuXqydtCugp5xD0hyTuX6VkQdNoY_oaRXXGRCUPCvfsXpU-HfL7pRkcXCjMLTrVIqf3RjD4_gar_ExbIb180nbYMG87E4rOILi5bIin51KOcMNBdHW7pTKwsS221XGyIAKIj65WxD-K8QSVr1HN6s7KDfYdez5dstWWC6xGg_3IZkrJjx8Df1HNT73K9PwUaAl7gcK4_2RSr8L636QbF3_y4Tv26g5IEv90Mc1mnAFPvUGI-nDR4s5Vso0398DlYl_tw-zbQQmn-QuQ6rS_ZIcZKgyoHKwD-3lgV-J1_--ndHdh8qwJC7Hk8ZKg'
render() {
let socket = null;
function connectToSparkd() {
const wsUrl = `ws://${Home.sparkdHost}:8080`;
// const wsUrl = Home.websocket_url;
if (socket == null) {
socket = new WebSocket(wsUrl)
}
if (socket && socket.readyState === WebSocket.OPEN) {
return;
}
const output = document.getElementById('query-output');
socket.onopen = (event) => {
console.log(`Connected to ${wsUrl}`);
};
socket.onmessage = (event) => {
console.log(`Received message: ${event.data}`);
output.innerHTML = event.data;
};
socket.onerror = (error) => {
console.log(`Error on websocket ${wsUrl}: ${error}`);
};
socket.onclose = (event) => {
if (event.wasClean) {
const msg = `Connection to ${wsUrl} closed cleanly, code=${event.code}, reason=${event.reason}`;
console.log(msg);
} else {
const msg = `Connection to ${wsUrl} abruptly closed`;
console.log(msg);
}
};
}
function executeQuery() {
if (socket == null || !socket) {
connectToSparkd();
}
if (socket == null || !socket) {
console.log('Failed to connect to websocket server');
return;
}
if (socket.readyState === WebSocket.OPEN) {
const queryInput = document.getElementById('query-input');
// const req = {
// header: {
// messageType: "subscribe",
// sentTime: "",
// version: "",
// requestId: "1",
// token: ""
// },
// payload: {
// subscriptionMessageTypes: ["TRD"],
// subscriptions: [{
// exchangeMic: "XCME", productType: "FUT", productCode: "BTC"
// }]
// }
// };
const rpcReq = {
jsonrpc: '2.0',
method: 'subscribe',
params: {
query: queryInput.value
},
id: 1,
};
socket.send(JSON.stringify(rpcReq));
}
}
return (
<div>
<h1>Data Stream Query Analyzer</h1>
<p>Run queries against the BTS Cloud</p>
<div>
<textarea id="query-input" placeholder='Enter Query'></textarea>
</div>
<div>
<button id="query-button" onClick={executeQuery}>Execute</button>
</div>
<table id="query-output">
</table>
<ul>
<li><strong>Client-side navigation</strong>. For example, click <em>Counter</em> then <em>Back</em> to return here.</li>
<li><strong>Development server integration</strong>. In development mode, the development server from <code>create-react-app</code> runs in the background automatically, so your client-side resources are dynamically built on demand and the page refreshes when you modify any file.</li>
<li><strong>Efficient production builds</strong>. In production mode, development-time features are disabled, and your <code>dotnet publish</code> configuration produces minified, efficiently bundled JavaScript files.</li>
</ul>
<p>The <code>ClientApp</code> subdirectory is a standard React application based on the <code>create-react-app</code> template. If you open a command prompt in that directory, you can run <code>npm</code> commands such as <code>npm test</code> or <code>npm install</code>.</p>
</div>
);
}
}
@@ -1 +0,0 @@
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" aria-hidden="true" role="img" class="iconify iconify--logos" width="35.93" height="32" preserveAspectRatio="xMidYMid meet" viewBox="0 0 256 228"><path fill="#00D8FF" d="M210.483 73.824a171.49 171.49 0 0 0-8.24-2.597c.465-1.9.893-3.777 1.273-5.621c6.238-30.281 2.16-54.676-11.769-62.708c-13.355-7.7-35.196.329-57.254 19.526a171.23 171.23 0 0 0-6.375 5.848a155.866 155.866 0 0 0-4.241-3.917C100.759 3.829 77.587-4.822 63.673 3.233C50.33 10.957 46.379 33.89 51.995 62.588a170.974 170.974 0 0 0 1.892 8.48c-3.28.932-6.445 1.924-9.474 2.98C17.309 83.498 0 98.307 0 113.668c0 15.865 18.582 31.778 46.812 41.427a145.52 145.52 0 0 0 6.921 2.165a167.467 167.467 0 0 0-2.01 9.138c-5.354 28.2-1.173 50.591 12.134 58.266c13.744 7.926 36.812-.22 59.273-19.855a145.567 145.567 0 0 0 5.342-4.923a168.064 168.064 0 0 0 6.92 6.314c21.758 18.722 43.246 26.282 56.54 18.586c13.731-7.949 18.194-32.003 12.4-61.268a145.016 145.016 0 0 0-1.535-6.842c1.62-.48 3.21-.974 4.76-1.488c29.348-9.723 48.443-25.443 48.443-41.52c0-15.417-17.868-30.326-45.517-39.844Zm-6.365 70.984c-1.4.463-2.836.91-4.3 1.345c-3.24-10.257-7.612-21.163-12.963-32.432c5.106-11 9.31-21.767 12.459-31.957c2.619.758 5.16 1.557 7.61 2.4c23.69 8.156 38.14 20.213 38.14 29.504c0 9.896-15.606 22.743-40.946 31.14Zm-10.514 20.834c2.562 12.94 2.927 24.64 1.23 33.787c-1.524 8.219-4.59 13.698-8.382 15.893c-8.067 4.67-25.32-1.4-43.927-17.412a156.726 156.726 0 0 1-6.437-5.87c7.214-7.889 14.423-17.06 21.459-27.246c12.376-1.098 24.068-2.894 34.671-5.345a134.17 134.17 0 0 1 1.386 6.193ZM87.276 214.515c-7.882 2.783-14.16 2.863-17.955.675c-8.075-4.657-11.432-22.636-6.853-46.752a156.923 156.923 0 0 1 1.869-8.499c10.486 2.32 22.093 3.988 34.498 4.994c7.084 9.967 14.501 19.128 21.976 27.15a134.668 134.668 0 0 1-4.877 4.492c-9.933 8.682-19.886 14.842-28.658 17.94ZM50.35 144.747c-12.483-4.267-22.792-9.812-29.858-15.863c-6.35-5.437-9.555-10.836-9.555-15.216c0-9.322 13.897-21.212 37.076-29.293c2.813-.98 5.757-1.905 8.812-2.773c3.204 10.42 7.406 21.315 12.477 32.332c-5.137 11.18-9.399 22.249-12.634 32.792a134.718 134.718 0 0 1-6.318-1.979Zm12.378-84.26c-4.811-24.587-1.616-43.134 6.425-47.789c8.564-4.958 27.502 2.111 47.463 19.835a144.318 144.318 0 0 1 3.841 3.545c-7.438 7.987-14.787 17.08-21.808 26.988c-12.04 1.116-23.565 2.908-34.161 5.309a160.342 160.342 0 0 1-1.76-7.887Zm110.427 27.268a347.8 347.8 0 0 0-7.785-12.803c8.168 1.033 15.994 2.404 23.343 4.08c-2.206 7.072-4.956 14.465-8.193 22.045a381.151 381.151 0 0 0-7.365-13.322Zm-45.032-43.861c5.044 5.465 10.096 11.566 15.065 18.186a322.04 322.04 0 0 0-30.257-.006c4.974-6.559 10.069-12.652 15.192-18.18ZM82.802 87.83a323.167 323.167 0 0 0-7.227 13.238c-3.184-7.553-5.909-14.98-8.134-22.152c7.304-1.634 15.093-2.97 23.209-3.984a321.524 321.524 0 0 0-7.848 12.897Zm8.081 65.352c-8.385-.936-16.291-2.203-23.593-3.793c2.26-7.3 5.045-14.885 8.298-22.6a321.187 321.187 0 0 0 7.257 13.246c2.594 4.48 5.28 8.868 8.038 13.147Zm37.542 31.03c-5.184-5.592-10.354-11.779-15.403-18.433c4.902.192 9.899.29 14.978.29c5.218 0 10.376-.117 15.453-.343c-4.985 6.774-10.018 12.97-15.028 18.486Zm52.198-57.817c3.422 7.8 6.306 15.345 8.596 22.52c-7.422 1.694-15.436 3.058-23.88 4.071a382.417 382.417 0 0 0 7.859-13.026a347.403 347.403 0 0 0 7.425-13.565Zm-16.898 8.101a358.557 358.557 0 0 1-12.281 19.815a329.4 329.4 0 0 1-23.444.823c-7.967 0-15.716-.248-23.178-.732a310.202 310.202 0 0 1-12.513-19.846h.001a307.41 307.41 0 0 1-10.923-20.627a310.278 310.278 0 0 1 10.89-20.637l-.001.001a307.318 307.318 0 0 1 12.413-19.761c7.613-.576 15.42-.876 23.31-.876H128c7.926 0 15.743.303 23.354.883a329.357 329.357 0 0 1 12.335 19.695a358.489 358.489 0 0 1 11.036 20.54a329.472 329.472 0 0 1-11 20.722Zm22.56-122.124c8.572 4.944 11.906 24.881 6.52 51.026c-.344 1.668-.73 3.367-1.15 5.09c-10.622-2.452-22.155-4.275-34.23-5.408c-7.034-10.017-14.323-19.124-21.64-27.008a160.789 160.789 0 0 1 5.888-5.4c18.9-16.447 36.564-22.941 44.612-18.3ZM128 90.808c12.625 0 22.86 10.235 22.86 22.86s-10.235 22.86-22.86 22.86s-22.86-10.235-22.86-22.86s10.235-22.86 22.86-22.86Z"></path></svg>

Before

Width:  |  Height:  |  Size: 4.0 KiB

@@ -1,68 +0,0 @@
:root {
font-family: Inter, system-ui, Avenir, Helvetica, Arial, sans-serif;
line-height: 1.5;
font-weight: 400;
color-scheme: light dark;
color: rgba(255, 255, 255, 0.87);
background-color: #242424;
font-synthesis: none;
text-rendering: optimizeLegibility;
-webkit-font-smoothing: antialiased;
-moz-osx-font-smoothing: grayscale;
}
a {
font-weight: 500;
color: #646cff;
text-decoration: inherit;
}
a:hover {
color: #535bf2;
}
body {
margin: 0;
display: flex;
place-items: center;
min-width: 320px;
min-height: 100vh;
}
h1 {
font-size: 3.2em;
line-height: 1.1;
}
button {
border-radius: 8px;
border: 1px solid transparent;
padding: 0.6em 1.2em;
font-size: 1em;
font-weight: 500;
font-family: inherit;
background-color: #1a1a1a;
cursor: pointer;
transition: border-color 0.25s;
}
button:hover {
border-color: #646cff;
}
button:focus,
button:focus-visible {
outline: 4px auto -webkit-focus-ring-color;
}
@media (prefers-color-scheme: light) {
:root {
color: #213547;
background-color: #ffffff;
}
a:hover {
color: #747bff;
}
button {
background-color: #f9f9f9;
}
}
@@ -1,10 +0,0 @@
import React from 'react'
import ReactDOM from 'react-dom/client'
import App from './App.jsx'
import './index.css'
ReactDOM.createRoot(document.getElementById('root')).render(
<React.StrictMode>
<App />
</React.StrictMode>,
)
@@ -1,60 +0,0 @@
import { fileURLToPath, URL } from 'node:url';
import { defineConfig } from 'vite';
import plugin from '@vitejs/plugin-react';
import fs from 'fs';
import path from 'path';
import child_process from 'child_process';
const baseFolder =
process.env.APPDATA !== undefined && process.env.APPDATA !== ''
? `${process.env.APPDATA}/ASP.NET/https`
: `${process.env.HOME}/.aspnet/https`;
const certificateArg = process.argv.map(arg => arg.match(/--name=(?<value>.+)/i)).filter(Boolean)[0];
const certificateName = certificateArg ? certificateArg.groups.value : "bts.datastreamwebapp.client";
if (!certificateName) {
console.error('Invalid certificate name. Run this script in the context of an npm/yarn script or pass --name=<<app>> explicitly.')
process.exit(-1);
}
const certFilePath = path.join(baseFolder, `${certificateName}.pem`);
const keyFilePath = path.join(baseFolder, `${certificateName}.key`);
if (!fs.existsSync(certFilePath) || !fs.existsSync(keyFilePath)) {
if (0 !== child_process.spawnSync('dotnet', [
'dev-certs',
'https',
'--export-path',
certFilePath,
'--format',
'Pem',
'--no-password',
], { stdio: 'inherit', }).status) {
throw new Error("Could not create certificate.");
}
}
// https://vitejs.dev/config/
export default defineConfig({
plugins: [plugin()],
resolve: {
alias: {
'@': fileURLToPath(new URL('./src', import.meta.url))
}
},
server: {
proxy: {
'^/weatherforecast': {
target: 'https://localhost:7128/',
secure: false
}
},
port: 5173,
https: {
key: fs.readFileSync(keyFilePath),
cert: fs.readFileSync(certFilePath),
}
}
})
@@ -1,28 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<InvariantGlobalization>true</InvariantGlobalization>
<SpaRoot>..\Bts.DataStreamWebApp.Client</SpaRoot>
<SpaProxyLaunchCommand>npm run dev</SpaProxyLaunchCommand>
<SpaProxyServerUrl>https://localhost:5173</SpaProxyServerUrl>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.AspNetCore.SpaProxy">
<Version>8.*-*</Version>
</PackageReference>
<PackageReference Include="Swashbuckle.AspNetCore" Version="6.4.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Bts.DataStreamWebApp.Client\Bts.DataStreamWebApp.Client.esproj">
<ReferenceOutputAssembly>false</ReferenceOutputAssembly>
</ProjectReference>
<ProjectReference Include="..\DataStream\DataStream.csproj" />
<ProjectReference Include="..\SharedStreams\SharedStreams.csproj" />
</ItemGroup>
</Project>
@@ -1,6 +0,0 @@
@Bts.DataStreamWebApp.Server_HostAddress = http://localhost:5268
GET {{Bts.DataStreamWebApp.Server_HostAddress}}/weatherforecast/
Accept: application/json
###
@@ -1,25 +0,0 @@
namespace Bts.DataStreamWebApp.Server.Controllers;
using Microsoft.AspNetCore.Mvc;
[ApiController]
[Route("[controller]")]
public class DataStreamSubscriptionController : ControllerBase {
private static readonly string[] Summaries = new[]
{
"Freezing", "Bracing", "Chilly", "Cool", "Mild", "Warm", "Balmy", "Hot", "Sweltering", "Scorching"
};
private readonly ILogger<DataStreamSubscriptionController> _logger;
public DataStreamSubscriptionController(ILogger<DataStreamSubscriptionController> logger) {
_logger = logger;
}
[HttpGet(Name = "GetDataStreamSubscription")]
public IEnumerable<DataStreamSubscription> Get() {
return Enumerable.Range(1, 5).Select(index => new DataStreamSubscription {
Query = "fake queronious"
})
.ToArray();
}
}
@@ -1,31 +0,0 @@
namespace Bts.DataStreamWebApp.Server.Controllers;
using Microsoft.AspNetCore.Mvc;
[ApiController]
[Route("[controller]")]
public class WeatherForecastController : ControllerBase
{
private static readonly string[] Summaries = new[]
{
"Freezing", "Bracing", "Chilly", "Cool", "Mild", "Warm", "Balmy", "Hot", "Sweltering", "Scorching"
};
private readonly ILogger<WeatherForecastController> _logger;
public WeatherForecastController(ILogger<WeatherForecastController> logger)
{
_logger = logger;
}
[HttpGet(Name = "GetWeatherForecast")]
public IEnumerable<WeatherForecast> Get()
{
return Enumerable.Range(1, 5).Select(index => new WeatherForecast
{
Date = DateOnly.FromDateTime(DateTime.Now.AddDays(index)),
TemperatureC = Random.Shared.Next(-20, 55),
Summary = Summaries[Random.Shared.Next(Summaries.Length)]
})
.ToArray();
}
}
@@ -1,5 +0,0 @@
namespace Bts.DataStreamWebApp.Server;
public class DataStreamSubscription {
public string? Query { get; set; }
}
@@ -1,30 +0,0 @@
var builder = WebApplication.CreateBuilder(args);
// Add services to the container.
builder.Services.AddControllers();
// Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();
var app = builder.Build();
app.UseDefaultFiles();
app.UseStaticFiles();
// Configure the HTTP request pipeline.
if (app.Environment.IsDevelopment())
{
app.UseSwagger();
app.UseSwaggerUI();
}
app.UseHttpsRedirection();
app.UseAuthorization();
app.MapControllers();
app.MapFallbackToFile("/index.html");
app.Run();
@@ -1,43 +0,0 @@
{
"profiles": {
"http": {
"commandName": "Project",
"launchBrowser": true,
"launchUrl": "swagger",
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development",
"ASPNETCORE_HOSTINGSTARTUPASSEMBLIES": "Microsoft.AspNetCore.SpaProxy"
},
"dotnetRunMessages": true,
"applicationUrl": "http://localhost:5268"
},
"https": {
"commandName": "Project",
"launchUrl": "swagger",
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development",
"ASPNETCORE_HOSTINGSTARTUPASSEMBLIES": "Microsoft.AspNetCore.SpaProxy"
},
"dotnetRunMessages": true,
"applicationUrl": "https://localhost:7128;http://localhost:5268"
},
"IIS Express": {
"commandName": "IISExpress",
"launchBrowser": true,
"launchUrl": "swagger",
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development",
"ASPNETCORE_HOSTINGSTARTUPASSEMBLIES": "Microsoft.AspNetCore.SpaProxy"
}
}
},
"$schema": "http://json.schemastore.org/launchsettings.json",
"iisSettings": {
"windowsAuthentication": false,
"anonymousAuthentication": true,
"iisExpress": {
"applicationUrl": "http://localhost:2083",
"sslPort": 44383
}
}
}
@@ -1,12 +0,0 @@
namespace Bts.DataStreamWebApp.Server;
public class WeatherForecast
{
public DateOnly Date { get; set; }
public int TemperatureC { get; set; }
public int TemperatureF => 32 + (int)(TemperatureC / 0.5556);
public string? Summary { get; set; }
}
@@ -1,8 +0,0 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
}
}
@@ -1,9 +0,0 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
},
"AllowedHosts": "*"
}
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -15,7 +15,6 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="HtmlAgilityPack" Version="1.11.57" />
<PackageReference Include="MongoDB.Bson" Version="2.23.1" />
<PackageReference Include="MongoDB.Driver" Version="2.23.1" />
</ItemGroup>
@@ -30,8 +30,8 @@ namespace BCG
{
using (FileStream fs = new FileStream(fileName, FileMode.Open, FileAccess.Read))
{
#pragma warning disable SYSLIB0011
BinaryFormatter bf = new BinaryFormatter();
#pragma warning disable SYSLIB0011
BlueTable bt = (BlueTable)bf.Deserialize(fs);
#pragma warning restore SYSLIB0011
mySetRealSource(new BlueRowCollectionStaticStream(bt));
@@ -1,31 +0,0 @@
namespace BCG {
using System.Collections.Generic;
public class KeyValueToRelationalStream : ViewStream {
private readonly ViewStream _kv_str;
private readonly string _doc_id_col, _key_col, _val_col;
public KeyValueToRelationalStream(ViewStream kv_str, string doc_id_col, string key_col, string val_col) {
_kv_str = kv_str;
_doc_id_col = doc_id_col;
_key_col = key_col;
_val_col = val_col;
}
protected override void myOnFirstSubscriber() {
_kv_str.onChange += _on_str_update;
}
protected override void myOnLastSubscriber() {
_kv_str.onChange -= _on_str_update;
}
private void _on_str_update(object dp_obj) {
DataPacket dp = dp_obj as DataPacket;
var rows = dp.newRows;
var rv = BlueDataUtility.keyvalue_to_relational<string>(rows, _doc_id_col, _key_col, _val_col);
var dp_out = BlueDataUtility.build_dp_inc(rv, myLastValue as DataPacket, true);
myFireEvent(dp_out);
}
}
}
@@ -1,121 +0,0 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Security.Cryptography.X509Certificates;
using BCG;
using BCG.Utilities.General;
using HtmlAgilityPack;
namespace BCG {
class Program {
static void Main() {
string url = "http://quotes.toscrape.com";
List<string> quotes = ScrapeQuotes(url);
Console.WriteLine("Quotes from http://quotes.toscrape.com:");
foreach (var quote in quotes) {
Console.WriteLine(quote);
}
}
static List<string> ScrapeQuotes(string url) {
List<string> quotes = new List<string>();
// Create a new HtmlWeb instance
HtmlWeb web = new HtmlWeb();
try {
// Load the HTML document from the specified URL
HtmlDocument document = web.Load(url);
// Extract quotes from the HTML document
var quoteNodes = document.DocumentNode.SelectNodes("//span[@class='text']");
if (quoteNodes != null) {
foreach (var quoteNode in quoteNodes) {
string quote = quoteNode.InnerText.Trim();
quotes.Add(quote);
}
}
} catch (Exception ex) {
Console.WriteLine($"Error: {ex.Message}");
}
return quotes;
}
}
public class WebScapperStream : ViewStream, IWorker {
private readonly string _url;
private readonly string _xpath;
private readonly ViewStreamDatabase _vsd;
private IDictionary<string, int> _line_id_map = new Dictionary<string, int>();
private readonly IBlueTableSchema _schema;
public WebScapperStream()
:
this("http://quotes.toscrape.com", "//span[@class='text']", ViewStreamDatabase.instance) { }
public WebScapperStream(string url, string xpath)
: this(url, xpath, ViewStreamDatabase.instance) { }
public WebScapperStream(string url, string xpath, ViewStreamDatabase vsd) {
bts.Log.Gen.info($"WebScapperStream cons url: {url} xpath: {xpath}");
_schema = BlueDataUtility.form_schema($"{url}_{xpath}",
Tuple.Create("id", TypeCode.Int32),
Tuple.Create("url", TypeCode.UInt32),
Tuple.Create("line", TypeCode.String));
_url = url;
_xpath = xpath;
_vsd = vsd;
}
protected override void myOnFirstSubscriber() {
MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(1));
}
protected override void myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
}
public void doWork() {
var lines = scrape_lines(_url, _xpath);
Func<string, int> get_id = (line) => {
int rv = 0;
if (!_line_id_map.TryGetValue(line, out rv))
_line_id_map.Add(line, _line_id_map.Count);
return rv;
};
BlueTable nrows = new BlueTable(_schema);
foreach (var l in lines) {
bts.Log.Gen.info($"line: {l}");
var id = get_id(l);
BlueRow nr = new BlueRow(_schema, get_id(l), _url, l);
nrows.addRow(nr);
}
var dp = BlueDataUtility.build_dp_inc(nrows, myLastValue as DataPacket, true);
myFireEvent(dp);
}
static List<string> scrape_lines(string url, string xpath) {
List<string> quotes = new List<string>();
// Create a new HtmlWeb instance
HtmlWeb web = new HtmlWeb();
try {
// Load the HTML document from the specified URL
HtmlDocument document = web.Load(url);
// Extract quotes from the HTML document
var quote_nodes = document.DocumentNode.SelectNodes(xpath);
if (quote_nodes != null) {
foreach (var qn in quote_nodes) {
string quote = qn.InnerText.Trim();
quotes.Add(quote);
}
}
} catch (Exception ex) {
Console.WriteLine($"Error: {ex.Message}");
}
return quotes;
}
}
}
@@ -1,212 +1,37 @@
using System;
using System.Collections.Generic;
using MongoDB.Driver;
using MongoDB.Bson;
using BCG;
using BCG.Utilities.General;
using ZstdSharp.Unsafe;
using System.Runtime.CompilerServices;
using mbts.spark;
namespace BCG {
[DataStreamUserRespresentation("mongodb_sql")]
public class MongoDBSQLStream : ViewStream, IWorker {
private ViewStreamDatabase _vsd;
private readonly string _connection_string, _db_name, _collection_name;
private MongoClient _client;
private BlueTable _table;
class Program {
static void Main() {
// Set the connection string and database name
string connectionString = "mongodb://localhost:27017";
string databaseName = "your_database_name";
string collectionName = "your_collection_name";
// Create a MongoClient
MongoClient client = new MongoClient(connectionString);
// Access the database
private IMongoDatabase _database;
private IDictionary<string, int> _id_map = new Dictionary<string, int>();
// mongodb://192.168.1.80:27017
IMongoDatabase database = client.GetDatabase(databaseName);
public MongoDBSQLStream(string connection_string, string db_name, string collection_name)
: this(connection_string, db_name, collection_name, ViewStreamDatabase.instance) {
// Access the collection
IMongoCollection<BsonDocument> collection = database.GetCollection<BsonDocument>(collectionName);
}
public MongoDBSQLStream(string connection_string, string db_name, string collection_name, ViewStreamDatabase vsd) {
bts.Log.Gen.info($"MongoDBStream constructed connection_string: {connection_string} " +
$"db_name: {db_name}");
_connection_string = connection_string;
_db_name = db_name;
_collection_name = collection_name;
_vsd = vsd;
_client = new MongoClient(_connection_string);
_database = _client.GetDatabase(_db_name);
}
// Fetch all documents from the collection
var documents = collection.Find(Builders<BsonDocument>.Filter.Empty).ToCursor();
protected override void myOnFirstSubscriber() {
MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(1));
}
// Iterate through the documents and print values
foreach (var document in documents.ToEnumerable()) {
// Assuming your document structure has fields named "Field1", "Field2", etc.
Console.WriteLine("Field1: " + document["Field1"]);
Console.WriteLine("Field2: " + document["Field2"]);
// Add more fields as needed
protected override void myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
}
public void doWork() {
_read_mogodb();
}
private void _read_mogodb() {
HighPerformanceTimer timer = new HighPerformanceTimer();
timer.start();
// Access the collection
IMongoCollection<BsonDocument> collection = _database.GetCollection<BsonDocument>(name: _collection_name);
// Fetch all documents from the collection
var documents = collection.Find(Builders<BsonDocument>.Filter.Empty).ToCursor();
_table = _table == null ? null : new BlueTable(_table.schema);
// Iterate through the documents and print values
foreach (var document in documents.ToEnumerable()) {
bts.Log.Gen.info($"document: {document} json: {document.ToJson()}");
if (_table == null) {
_init_table(document);
}
_add_row(document);
}
timer.stop();
bts.Log.Gen.info($"MongoDBStream time to read {_collection_name} {timer.lastIntervalTime} msecs");
timer.start();
var dp = BlueDataUtility.build_dp_inc(_table, myLastValue == null ? null : myLastValue as DataPacket, true);
timer.stop();
bts.Log.Gen.info($"MongoDBStream time to compare tables :{timer.lastIntervalTime} msecs");
if (dp.addedRows.Count == 0 && dp.deletedRows.Count == 0)
return;
myFireEvent(dp);
}
private void _init_table(BsonDocument doc_row) {
bts.Log.assert(_table == null);
List<ColumnInfo> cis = new List<ColumnInfo>();
cis.Add((new ColumnInfo("id", 0, typeof(Int32))));
foreach (var n in doc_row.Names) {
cis.Add(new ColumnInfo(n, cis.Count, typeof(string)));
}
IBlueTableSchema schema = new BlueTableSchema(cis.ToArray(), _collection_name);
_table = new BlueTable(schema);
}
private void _add_row(BsonDocument doc) {
var key_val = doc["_id"];
var key = key_val.ToString();
int id;
if (!_id_map.TryGetValue(key, out id)) {
id = _id_map.Count;
_id_map.Add(key, id);
}
MutableBlueRow nr = new MutableBlueRow(_table.schema);
nr.setValue("id", id);
foreach (var n in doc.Names) {
var val = doc[n].ToString();
nr.setValue(n, val);
}
_table.addRow(nr);
// Print a separator between documents
Console.WriteLine(new string('-', 20));
}
}
[DataStreamUserRespresentation("mongodb")]
public class MongoDBStream : ViewStream, IWorker {
private ViewStreamDatabase _vsd;
private readonly string _connection_string, _db_name, _collection_name;
private MongoClient _client;
private BlueTable _table;
// Access the database
private IMongoDatabase _database;
private IDictionary<string, int> _id_map = new Dictionary<string, int>();
// mongodb://192.168.1.80:27017
public MongoDBStream(string connection_string, string db_name, string collection_name)
: this(connection_string, db_name, collection_name, ViewStreamDatabase.instance) {
}
public MongoDBStream(string connection_string, string db_name, string collection_name, ViewStreamDatabase vsd) {
bts.Log.Gen.info($"MongoDBStream constructed connection_string: {connection_string} " +
$"db_name: {db_name}");
_connection_string = connection_string;
_db_name = db_name;
_collection_name = collection_name;
_vsd = vsd;
_client = new MongoClient(_connection_string);
_database = _client.GetDatabase(_db_name);
}
protected override void myOnFirstSubscriber() {
MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(1));
}
protected override void myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
}
public void doWork() {
_read_mogodb();
}
private void _read_mogodb() {
HighPerformanceTimer timer = new HighPerformanceTimer();
timer.start();
// Access the collection
IMongoCollection<BsonDocument> collection = _database.GetCollection<BsonDocument>(name: _collection_name);
// Fetch all documents from the collection
var documents = collection.Find(Builders<BsonDocument>.Filter.Empty).ToCursor();
_table = _table == null ? null : new BlueTable(_table.schema);
// Iterate through the documents and print values
foreach (var document in documents.ToEnumerable()) {
bts.Log.Gen.info($"document: {document} json: {document.ToJson()}");
if (_table == null) {
_init_table(document);
}
_add_doc(document);
}
timer.stop();
bts.Log.Gen.info($"MongoDBStream time to read {_collection_name} {timer.lastIntervalTime} msecs");
timer.start();
var dp = BlueDataUtility.build_dp_inc(_table, myLastValue == null ? null : myLastValue as DataPacket, true);
timer.stop();
bts.Log.Gen.info($"MongoDBStream time to compare tables :{timer.lastIntervalTime} msecs");
if (dp.addedRows.Count == 0 && dp.deletedRows.Count == 0)
return;
myFireEvent(dp);
}
private void _init_table(BsonDocument doc_row) {
bts.Log.assert(_table == null);
var schema = BlueDataUtility.form_schema($"{_database}_{_table}",
Tuple.Create("id", TypeCode.Int32),
Tuple.Create("doc_id", TypeCode.String),
Tuple.Create("key", TypeCode.String),
Tuple.Create("value", TypeCode.String));
_table = new BlueTable(schema);
}
private void _add_doc(BsonDocument doc) {
var doc_id_bson = doc["_id"];
var doc_id = doc_id_bson.ToString();
Func<string, string, int> get_id = (string doc_id, string key) => {
string hash_key = $"{doc_id}_{key}";
int id = 0;
if (!_id_map.TryGetValue(hash_key, out id)) {
id = _id_map.Count;
_id_map.Add(hash_key, id);
}
return id;
};
foreach (var n in doc.Names) {
string key = n;
var id = get_id(doc_id, key);
MutableBlueRow nr = new MutableBlueRow(_table.schema);
var val = doc[n].ToString();
nr.setValue("id", id);
nr.setValue("doc_id", doc_id);
nr.setValue("key", key);
nr.setValue("value", val);
_table.addRow(nr);
}
}
}
}
}
File diff suppressed because it is too large. Load diff
@@ -3,7 +3,8 @@ using System.Collections.Generic;
using System.Linq;
using System.Text;
namespace BCG {
namespace BCG
{
/*
These fields are first going to be used by BlueRay to help it determine what differnet tags mean, but eventually
* the goal is that they will be used for all services in this system.
@@ -11,12 +12,11 @@ namespace BCG {
* not all metadata fields are present in all metadata; it depends upon the class. For example, a bluepages service
* probably won't have a 'symbol' or 'exchange' field populated, but it'll probably have the 'type' of service populated
*/
public enum MetaDataField {
Protocol, // is this a bluepages service, a pubsub tag, a blueray tag, a 29 west thing, or what
Type, // if it's a multicast, is it option, fireprice, whatever. If it's bluepages, is it persistence, pubsub, whatever
Symbol,
Exchange,
MarketDataType, // BAM (bid, ask, mid), last, etc
RealName
}; //does this thing have a better name we could call it
public enum MetaDataField { Protocol, //is this a bluepages service, a pubsub tag, a blueray tag, a 29 west thing, or what
Type, //if it's a multicast, is it option, fireprice, whatever. If it's bluepages, is it persistence, pubsub, whatever
Symbol,
Exchange,
Platform,
MarketDataType, // BAM (bid, ask, mid), last, etc
RealName }; //does this thing have a better name we could call it
}
@@ -2,7 +2,7 @@
<PropertyGroup>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<Nullable>disable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -11,7 +11,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<Nullable>enable</Nullable>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
</PropertyGroup>
@@ -11,7 +11,7 @@ using System.Threading;
bts.Log.Gen.info($"Starting DSPublisher");
Console.WriteLine("Starting DSPublisher...");
bool run_echo_tcp_client_only = false;
bool run_echo_tcp_client_only = true;
static void init(string[] args) {
Func<string, string, Task> login = async (l, p) => {
var login = l;
@@ -31,7 +31,6 @@ else
static ViewStream build_stream(string stream_expression) {
try {
bts.Log.Gen.info($"build_stream stream_expression: {stream_expression}");
var exp = DataStreamExpressionsInterpreter.Evaluate(stream_expression);
var rv = exp.getView(new Dictionary<string, StatementExpression>(0));
return rv;
@@ -59,16 +58,14 @@ void _publish_streams() {
CloudStreamManager csm = new CloudStreamManager();
// get it from config
Dictionary<string, string> streams_to_publish = new Dictionary<string, string>() {
//{"bbo","bbo()"},
//{"instruments","instruments()"},
{"bbo","bbo()"},
{"instruments","instruments()"},
//{"timesales","timesales(10000)"},
//{"books","books()"},
{"seconds_1_10","second(1,10)"},
{"books","books()"},
// {"seconds_1_10","second(1,10)"},
{"literal_420","literal(\"x\", \"420.0\")"},
{"mongodb_testcollection", "mongodb(\"mongodb://192.168.1.80:27017\",\"local\",\"test_collection\")"}
};
// {"bbo_ts","timeSeries(bbo())"}
// {"bbo_ts","timeSeries(bbo())"}
};
foreach (var kv in streams_to_publish) {
var name = kv.Key;
var exp = kv.Value;
@@ -80,7 +77,6 @@ void _publish_streams() {
System.Threading.Thread.Sleep(TimeSpan.FromSeconds(1));
ViewStreamDatabase.instance.doWork();
csm.doWork();
MasterWorker.getInstance(false).doWork();
}
}
@@ -2,7 +2,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net8.0-windows</TargetFramework>
<TargetFramework>net7.0-windows</TargetFramework>
<Nullable>enable</Nullable>
<UseWindowsForms>true</UseWindowsForms>
</PropertyGroup>
-6
View File
@@ -38,8 +38,6 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DataStreamTests", "DataStre
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SharedStreams", "SharedStreams\SharedStreams.csproj", "{C1EBC423-4668-43D4-B7E5-A291086BA823}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.DataStreamWebApp.Server", "Bts.DataStreamWebApp.Server\Bts.DataStreamWebApp.Server.csproj", "{6DE5E4CE-423E-4571-81F7-539D268D6059}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.DataStream.Websocket", "Bts.DataStream.Websocket\Bts.DataStream.Websocket.csproj", "{088655DA-3126-4F8A-9AC7-9061FDA891FF}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bts.DataStream.WebsocketServer", "Bts.DataStream.WebsocketServer\Bts.DataStream.WebsocketServer.csproj", "{35823BE0-C883-476A-8B30-D6097CBA1A8D}"
@@ -133,10 +131,6 @@ Global
{C1EBC423-4668-43D4-B7E5-A291086BA823}.Debug|Any CPU.Build.0 = Debug|Any CPU
{C1EBC423-4668-43D4-B7E5-A291086BA823}.Release|Any CPU.ActiveCfg = Release|Any CPU
{C1EBC423-4668-43D4-B7E5-A291086BA823}.Release|Any CPU.Build.0 = Release|Any CPU
{6DE5E4CE-423E-4571-81F7-539D268D6059}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{6DE5E4CE-423E-4571-81F7-539D268D6059}.Debug|Any CPU.Build.0 = Debug|Any CPU
{6DE5E4CE-423E-4571-81F7-539D268D6059}.Release|Any CPU.ActiveCfg = Release|Any CPU
{6DE5E4CE-423E-4571-81F7-539D268D6059}.Release|Any CPU.Build.0 = Release|Any CPU
{088655DA-3126-4F8A-9AC7-9061FDA891FF}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{088655DA-3126-4F8A-9AC7-9061FDA891FF}.Debug|Any CPU.Build.0 = Debug|Any CPU
{088655DA-3126-4F8A-9AC7-9061FDA891FF}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0-windows</TargetFramework>
<TargetFramework>net7.0-windows</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<UseWindowsForms>true</UseWindowsForms>
@@ -47,30 +47,30 @@ namespace BCG {
/// of Int32.Max would return all the messages it has
/// received between last getUpdates and this call.
/// </summary
IList<byte[]> get_updates(TimeSpan timeout, int maxElements);
IList<byte[]> getUpdates(TimeSpan timeout, int maxElements);
/// <summary>
/// server URL to which the connection is made
/// </summary>
string server_url { get; }
string serverURL { get; }
/// <summary>
/// client URL as seen by the server. if a connection has not
/// been established (as can happen in TCP) an empty string
/// is returned
/// </summary>
string client_url { get; }
string clientURL { get; }
/// <summary>
/// registers the callback to receive
/// connection notifications
/// </summary>
void register_connection_callback(PubSubServerConnectionStatusChange del);
void registerCallback(ConnectionStatusChange del);
/// <summary>
/// registers callbacks to reveive data
/// </summary>
/// <param name="del"></param>
void register_data_callback(PubSubServerDataDelegate del);
void registerDataCallback(DataDelegate del);
void register_send_confirmation_callback(SendConfirmationDelegate del);
void registerSendConfirmationCallback(SendConfirmationDelegate del);
}
}
@@ -8,7 +8,6 @@ using System.IO;
//using BCG.Utilities.General;
using System.Threading;
using mbts;
using bts;
namespace BCG {
/// <summary>
@@ -21,27 +20,26 @@ namespace BCG {
/// ITCPConnectionHandler.onConnection will be called on any arbitrary thread
///
/// </summary>
public class PubSubServerManagedTCPConnection : IPubSubServerConnection,// external facing interface to communicate msgs to Pubsubserver
ITCPMessageHandler, // to get messages from ITCPCOnnection
ITCPConnectionHandler { // to get connections from AsyncTCPConnector_APM
public class PubSubServerManagedTCPConnection : IPubSubServerConnection,
ITCPMessageHandler,
ITCPConnectionHandler {
private ITCPConnection _connection, _new_connection;
private readonly BlutoSubAddress _server_address;
// if a message(with confirm) can not be sent , its id is queued here,
// on reconnection, _on_send_onfirm_callback is fired with a negative on these ids, so the user
//can re-send it. The re-send mechanism is not in implemented in this class, but by the user of this class
private ThreadSafeQueue<string> _negative_unsent_confs = new ThreadSafeQueue<string>();
// in onData, data is queued here, for polling
// in onData, data is queued here
private SyncQueue<byte[]> _in_data_queue = new SyncQueue<byte[]>();
private DateTime _last_hb = DateTime.Now;
private PubSubServerConnectionStatusChange _conn_callback;
private PubSubServerDataDelegate _data_callback;
private SendConfirmationDelegate _on_send_confirm_callback;
private ConnectionStatusChange _conn_callback;
private DataDelegate _data_callback;
private SendConfirmationDelegate _on_send_onfirm_callback;
private readonly TimeSpan _connection_drop_threshold =
new TimeSpan(0, 0, 0, 0,
PubSub3Config.instance.PS3_HeartBeatInterval * PubSub3Config.instance.PS3_MissingHeartBeatsForGC);
new TimeSpan(0, 0, 0, 0, PubSub3Config.instance.PS3_HeartBeatInterval * PubSub3Config.instance.PS3_MissingHeartBeatsForGC);
public PubSubServerManagedTCPConnection(string server_url)
: this(new BlutoSubAddress(server_url)) { }
public PubSubServerManagedTCPConnection(string serverURL)
: this(new BlutoSubAddress(serverURL)) { }
public PubSubServerManagedTCPConnection(BlutoSubAddress server_address) {
_server_address = server_address;
@@ -50,17 +48,17 @@ namespace BCG {
private void _connect_to_ps_server() {
bts.Log.Gen.info($"PubSubServerManagedTCPConnection._connect_to_ps_server server: {_server_address}");
AsyncTCPConnector_APM.connect_to(_server_address, this);
ClientToServerManagedTCPConnector.connect_to(_server_address, this, TimeSpan.FromSeconds(1));
}
#region IPubSubServerConnection Members
public void send_with_confirmation(byte[] data, ref string conf_id) {
bts.Log.Gen.info($"PubSubServerManagedTCPConnection.sendWithConfirmation server: {_server_address} data: {data.Length} _connection : {_connection != null}");
public void sendWithConfirmation(byte[] data, ref string conf_id) {
bts.Log.Gen.info($"PubSubServerManagedTCPConnection.sendWithConfirmation server: {_server_address} data: {data.Length} myConnection : {_connection != null}");
if (_connection != null) {
_connection.sendWithConfirmation(data, ref conf_id, (msg_id, confirmed) => {
if (_on_send_confirm_callback != null)
_on_send_confirm_callback(msg_id, confirmed);
if (_on_send_onfirm_callback != null)
_on_send_onfirm_callback(msg_id, confirmed);
});
} else {
bts.Log.Gen.warn($"PubSubServerManagedTCPConnection.sendWithConfirmation server: {_server_address} no connection, buffering send");
@@ -88,18 +86,18 @@ namespace BCG {
public void send_with_confirmation(PubSubMessage msg, ref string conf_id) {
var bytes = msg.to_bytes();
send_with_confirmation(bytes, ref conf_id);
sendWithConfirmation(bytes, ref conf_id);
}
public IList<byte[]> get_updates(TimeSpan timeout, int maxElements) {
public IList<byte[]> getUpdates(TimeSpan timeout, int maxElements) {
return _in_data_queue.dequeueMultipleElements(Convert.ToInt32(timeout.TotalMilliseconds), maxElements);
}
public string server_url {
public string serverURL {
get { return _server_address.ToString(); }
}
public string client_url {
public string clientURL {
get {
string rv = "";
lock (this) {
@@ -120,16 +118,16 @@ namespace BCG {
}
}
public void register_connection_callback(PubSubServerConnectionStatusChange del) {
public void registerCallback(ConnectionStatusChange del) {
_conn_callback = del;
}
public void register_data_callback(PubSubServerDataDelegate del) {
public void registerDataCallback(DataDelegate del) {
_data_callback = del;
}
public void register_send_confirmation_callback(SendConfirmationDelegate del) {
_on_send_confirm_callback = del;
public void registerSendConfirmationCallback(SendConfirmationDelegate del) {
_on_send_onfirm_callback = del;
}
#endregion
@@ -187,6 +185,18 @@ namespace BCG {
break;
}
// this stupid shit, caused duplicating og messages and 2 days of wild goose chase
// if (message.messageType == RemoteMessageType.Data)
// {
// if (_data_callback != null)
// _data_callback(message.remoteConnectionInfo.ToString(), message.message);
// else
// _in_data_queue.enqueue(message.message);
// }
// else
// {
// _last_hb = DateTime.Now;
// }
}
#endregion
@@ -194,7 +204,8 @@ namespace BCG {
#region ITCPConnectionHandler Members
/// <summary>
/// will be called on ana rbitrary .Net thread
/// This function is called on the main thread (as all other functions
/// except onData)
/// </summary>
/// <param name="connection"></param>
public void onConnection(ITCPConnection connection) {
@@ -235,10 +246,10 @@ namespace BCG {
_connection.register(this);
if (_conn_callback != null)
_conn_callback(_server_address.ToString(), true);
if (_on_send_confirm_callback != null) {
if (_on_send_onfirm_callback != null) {
var msg_ids = _negative_unsent_confs.dequeueAll();
foreach (var id in msg_ids)
_on_send_confirm_callback(id, false);
_on_send_onfirm_callback(id, false);
}
}
}
@@ -253,6 +264,47 @@ namespace BCG {
public void doWork() {
//_dispose_and_reconnect_if_missing_hb();
_check_for_new_connection();
///check for heartbeats and let clients know in
///case of missing hb (connection breaking)
// TimeSpan hb_interval = DateTime.Now - myLastHB;
// bool hb_missed = hb_interval > myConnectionDropThreshold;
// bool conn_disposed = myConnection != null || myConnection.isDisposed;
// if (conn_disposed || hb_missed)
// {
// if (hb_missed)
// {
// // since conn breakage was diagnosed logically, lets physically close the connection
// myConnection.Dispose();
// string dispose_msg = $"PubSubServerManagedTCPConnection.doWork server: {myServerAddress} Disposing connection conn_disposed: {conn_disposed} hb_missed: {hb_missed}";
// Console.WriteLine(dispose_msg);
// }
// myConnection = null; // this is used as a marker for a dead connection
// if (myConnCallback != null)
// myConnCallback(myServerAddress.ToString(), false);
// string reconn_msg = $"PubSubServerManagedTCPConnection.doWork disconnecting client from server: {myServerAddress} conn_disposed: {conn_disposed} hb_missed: {hb_missed} " +
// $"because of missing hbs last_hb: {myLastHB}, now: {DateTime.Now} curr_hb_interval: {hb_interval.TotalSeconds} threshold: {myConnectionDropThreshold.TotalSeconds} conn_disposed: {conn_disposed}";
// ClientToServerManagedTCPConnector.connectTo(myServerAddress, this);
// }
// if (myTempNewConnection != null)
// {
// if (myConnection != null)
// {
// // TODOds
// // log error and close it
// // BlueEventLog.logAssert(myConnection == null, "Connection has to be nulled out and disposed "
// // + " before a new connection is established");
// }
// Console.WriteLine("PubSubServerManagedTCPConnection hooking new connection");
// myConnection = myTempNewConnection;
// myTempNewConnection = null;
// myConnection.register(this);
// myConfs.dequeueAll();
// if (myConnCallback != null)
// myConnCallback(myServerAddress.ToString(), true);
// }
// foreach (Pair<string, bool> p in myConfs.dequeueAll())
// if (myOnSendConfirmCallback != null)
// myOnSendConfirmCallback(p.first, p.second);
}
#endregion
@@ -16,7 +16,6 @@ namespace BCG {
// SubMetaData
// TransactionMetaData
// UnSubMetaData
//
public class MetaData : ICloneable {
private IDictionary _name_value_map = new Dictionary<string, string>();
int mySize = -1;
@@ -1,8 +1,10 @@
using System;
namespace BCG {
//[Serializable]
public class SubMetaData {
namespace BCG
{
[Serializable]
public class SubMetaData
{
/// <summary>
/// connection URL on which the connection will receive data
/// </summary>
@@ -11,18 +13,20 @@ namespace BCG {
/// identification of the client and the connection on which to receive data
/// machine; ip; port; clientURL
/// </summary>
public readonly string subscribing_app = ClientInfo.thisClient.processString;
public string subscribingApp;
/// <summary>
/// tags that are to be subscribed
/// </summary>
public string[] tags;
public SubMetaData() { }
public SubMetaData()
{ }
public SubMetaData(string the_connection, params string[] the_tags) {
connection = the_connection;
tags = the_tags;
//subscribing_app = ClientInfo.thisClient.processString;
public SubMetaData(string theConnection, params string[] theTags)
{
connection = theConnection;
tags = theTags;
subscribingApp = ClientInfo.thisClient.processString;
}
}
}
@@ -1,17 +1,21 @@
using System;
namespace BCG {
//[Serializable]
public class UnSubMetaData {
namespace BCG
{
[Serializable]
public class UnSubMetaData
{
public string connection;
public readonly string[] tags;
public readonly string unsubscribing_app = ClientInfo.thisClient.processString;
public string[] tags;
public string unsubscribingApp = ClientInfo.thisClient.processString;
public UnSubMetaData() { }
public UnSubMetaData()
{}
public UnSubMetaData(string the_connection, params string[] the_tags) {
connection = the_connection;
tags = the_tags;
public UnSubMetaData(string theConnection, params string[] theTags)
{
connection = theConnection;
tags = theTags;
}
}
}
@@ -6,18 +6,12 @@ namespace BCG
public interface IPublisher
{
void publish( IPubItem pi );
#pragma warning disable SYSLIB0011 // Type or member is obsolete
void publish( IPubItem pi, IFormatter f );
#pragma warning restore SYSLIB0011 // Type or member is obsolete
void publish( IPubItem[] pi );
#pragma warning disable SYSLIB0011 // Type or member is obsolete
void publish( IPubItem[] pi, IFormatter f );
#pragma warning restore SYSLIB0011 // Type or member is obsolete
}
#pragma warning disable SYSLIB0011 // Type or member is obsolete
public interface IPublisher<T> where T : IFormatter
#pragma warning restore SYSLIB0011 // Type or member is obsolete
{
void publish (IPubItem pi);
void publish (IPubItem[] pi);
@@ -8,9 +8,7 @@ namespace BCG
public class Publisher : IPublisher
{
private PublisherCore myPubCore = new PublisherCore ();
#pragma warning disable SYSLIB0011 // Type or member is obsolete
private BinaryFormatter myBinFormatter = new BinaryFormatter ();
#pragma warning restore SYSLIB0011 // Type or member is obsolete
private const string myTagType = "DefaultTag";
public Publisher() {}
@@ -23,9 +21,7 @@ namespace BCG
public void publish (IPubItem[] pis)
{ publish( pis, myBinFormatter ); }
#pragma warning disable SYSLIB0011 // Type or member is obsolete
public void publish( IPubItem pi, IFormatter f )
#pragma warning restore SYSLIB0011 // Type or member is obsolete
{
MemoryStream ms = new MemoryStream();
#pragma warning disable
@@ -35,9 +31,7 @@ namespace BCG
myPubCore.publish( myTagType, pi.getTag(), ms );
}
#pragma warning disable SYSLIB0011 // Type or member is obsolete
public void publish( IPubItem[] pis, IFormatter f )
#pragma warning restore SYSLIB0011 // Type or member is obsolete
{
myPubCore.beginTransaction();
foreach ( IPubItem pi in pis )
@@ -48,14 +42,10 @@ namespace BCG
#endregion
}
#pragma warning disable SYSLIB0011 // Type or member is obsolete
public class Publisher<T> : IPublisher<T> where T : IFormatter, new ()
#pragma warning restore SYSLIB0012 // Type or member is obsolete
{
private PublisherCore myPubCore = new PublisherCore ();
#pragma warning disable SYSLIB0011 // Type or member is obsolete
private IFormatter myFormatter = new T ();
#pragma warning restore SYSLIB0011 // Type or member is obsolete
private const string myTagType = "DefaultTag";
public Publisher () { }
@@ -65,9 +55,9 @@ namespace BCG
public void publish (IPubItem pi)
{
MemoryStream ms = new MemoryStream ();
#pragma warning disable
#pragma warning disable
myFormatter.Serialize (ms, pi);
#pragma warning restore SYSLIB0011
#pragma warning restore SYSLIB0011
ms.Position = 0;
myPubCore.publish (myTagType, pi.getTag (), ms);
}
@@ -328,8 +328,7 @@ namespace bts {
return;
}
_send_key_map.Add(message.key);
bts.Log.Gen.info($"AsyncTCPConnection_APM._on_send: pre endSend this: {ToString()} tid: {_tid()} " +
$"res: {_cb_txt(res)} message: {res.AsyncState} _is_alive: {_is_alive}");
bts.Log.Gen.info($"AsyncTCPConnection_APM._on_send: pre endSend this: {ToString()} tid: {_tid()} res: {_cb_txt(res)} message: {res.AsyncState} _is_alive: {_is_alive}");
if (_is_alive) {
try {
SocketError send_err;
@@ -1,82 +1,82 @@
// using System;
// using System.Collections.Generic;
// using System.Text;
// using System.Net;
// using System.Net.Sockets;
// using System.Threading;
// using System.Threading.Tasks;
// //using BCG.Utilities;
// //using BCG.Utilities.General;
using System;
using System.Collections.Generic;
using System.Text;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
//using BCG.Utilities;
//using BCG.Utilities.General;
// namespace BCG {
// internal class SocketConnectionArgs {
// public EndPoint ep;
// public Socket sock;
// public ITCPConnectionHandler handler;
// public string serverURL;
namespace BCG {
internal class SocketConnectionArgs {
public EndPoint ep;
public Socket sock;
public ITCPConnectionHandler handler;
public string serverURL;
// internal SocketConnectionArgs(EndPoint theEp, Socket theSock, ITCPConnectionHandler theHandler,
// string theServerURL) {
// ep = theEp;
// sock = theSock;
// handler = theHandler;
// serverURL = theServerURL;
// }
// }
internal SocketConnectionArgs(EndPoint theEp, Socket theSock, ITCPConnectionHandler theHandler,
string theServerURL) {
ep = theEp;
sock = theSock;
handler = theHandler;
serverURL = theServerURL;
}
}
// public class ClientToServerManagedTCPConnector {
// private static Random myRnd = new Random(DateTime.Now.Millisecond);
// private static TimeSpan myHBTimeSpan = new TimeSpan(0, 0, 0, 0, PubSub3Config.instance.PS3_HeartBeatInterval);
// private static int myWaitTimeBeforeRetryingUpperLimit = 5000,
// myWaitTimeBeforeRetryingLowerLimit = 1000;
public class ClientToServerManagedTCPConnector {
private static Random myRnd = new Random(DateTime.Now.Millisecond);
private static TimeSpan myHBTimeSpan = new TimeSpan(0, 0, 0, 0, PubSub3Config.instance.PS3_HeartBeatInterval);
private static int myWaitTimeBeforeRetryingUpperLimit = 5000,
myWaitTimeBeforeRetryingLowerLimit = 1000;
// public static void connectTo(BlutoSubAddress serverAddress, ITCPConnectionHandler handler) {
// Console.WriteLine($"ClientToServerManagedTCPConnector.connectTo server: {serverAddress}");
// Socket sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
// EndPoint ep = new IPEndPoint(IPAddress.Parse(serverAddress.ip), serverAddress.port);
// sock.BeginConnect(ep, new AsyncCallback(myOnConnect), new SocketConnectionArgs(ep, sock, handler, serverAddress.ToString()));
// }
public static void connectTo(BlutoSubAddress serverAddress, ITCPConnectionHandler handler) {
Console.WriteLine($"ClientToServerManagedTCPConnector.connectTo server: {serverAddress}");
Socket sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
EndPoint ep = new IPEndPoint(IPAddress.Parse(serverAddress.ip), serverAddress.port);
sock.BeginConnect(ep, new AsyncCallback(myOnConnect), new SocketConnectionArgs(ep, sock, handler, serverAddress.ToString()));
}
// public static async Task connect_to_async(BlutoSubAddress serverAddress, ITCPConnectionHandler handler, TimeSpan wait_before_retrying, int max_n_attempts = 1024) {
// bts.Log.Gen.info($"ClientToServerManagedTCPConnector.connectTo server: {serverAddress}");
// await TCPSocketUtils.connect_to_server_async(serverAddress.ip, serverAddress.port, handler, wait_before_retrying, max_n_attempts);
// }
public static async Task connect_to_async(BlutoSubAddress serverAddress, ITCPConnectionHandler handler, TimeSpan wait_before_retrying, int max_n_attempts = 1024) {
bts.Log.Gen.info($"ClientToServerManagedTCPConnector.connectTo server: {serverAddress}");
await TCPSocketUtils.connect_to_server_async(serverAddress.ip, serverAddress.port, handler, wait_before_retrying, max_n_attempts);
}
// public static void connect_to(BlutoSubAddress serverAddress, ITCPConnectionHandler handler, TimeSpan wait_before_retrying, int max_n_attempts = 1024) {
// bts.Log.Gen.info($"ClientToServerManagedTCPConnector.connectTo server: {serverAddress}");
// Task.Run(async () => {
// await TCPSocketUtils.connect_to_server_async(serverAddress.ip, serverAddress.port, handler, wait_before_retrying, max_n_attempts);
// });
// }
public static void connect_to(BlutoSubAddress serverAddress, ITCPConnectionHandler handler, TimeSpan wait_before_retrying, int max_n_attempts = 1024) {
bts.Log.Gen.info($"ClientToServerManagedTCPConnector.connectTo server: {serverAddress}");
Task.Run(async () => {
await TCPSocketUtils.connect_to_server_async(serverAddress.ip, serverAddress.port, handler, wait_before_retrying, max_n_attempts);
});
}
// private static void myOnConnect(IAsyncResult res) {
// Console.WriteLine("ClientToServerManagedTCPConnector.myOnConnect " + res.ToString() + " " + res.AsyncState.ToString());
// SocketConnectionArgs args = res.AsyncState as SocketConnectionArgs;
// try {
// args.sock.EndConnect(res);
// } catch (SocketException sex) {
// string errMsg = String.Format("{0}: Connection to {1} failed.", DateTime.Now.ToString(), args.serverURL.ToString());
// //BlueEventLog.logEvent(sex.Message + ";" + errMsg, PS3LogLevels.disconnection);
// //Console.WriteLine(errMsg);
// if (args.handler.isDisposed) {
// string doneMsg = String.Format("{0}: Connection to {1} was disposed. Not Retrying.", DateTime.Now.ToString(), args.serverURL.ToString());
// //Console.WriteLine(doneMsg);
// // BlueEventLog.logEvent(doneMsg, PS3LogLevels.disconnection);
// return;
// } else {
// string retryMsg = String.Format("{0}: Retrying Connection to {1}.", DateTime.Now.ToString(), args.serverURL.ToString());
// //Console.WriteLine(retryMsg);
// //BlueEventLog.logEvent(retryMsg, PS3LogLevels.disconnection);
private static void myOnConnect(IAsyncResult res) {
Console.WriteLine("ClientToServerManagedTCPConnector.myOnConnect " + res.ToString() + " " + res.AsyncState.ToString());
SocketConnectionArgs args = res.AsyncState as SocketConnectionArgs;
try {
args.sock.EndConnect(res);
} catch (SocketException sex) {
string errMsg = String.Format("{0}: Connection to {1} failed.", DateTime.Now.ToString(), args.serverURL.ToString());
//BlueEventLog.logEvent(sex.Message + ";" + errMsg, PS3LogLevels.disconnection);
//Console.WriteLine(errMsg);
if (args.handler.isDisposed) {
string doneMsg = String.Format("{0}: Connection to {1} was disposed. Not Retrying.", DateTime.Now.ToString(), args.serverURL.ToString());
//Console.WriteLine(doneMsg);
// BlueEventLog.logEvent(doneMsg, PS3LogLevels.disconnection);
return;
} else {
string retryMsg = String.Format("{0}: Retrying Connection to {1}.", DateTime.Now.ToString(), args.serverURL.ToString());
//Console.WriteLine(retryMsg);
//BlueEventLog.logEvent(retryMsg, PS3LogLevels.disconnection);
// int waitTime = myRnd.Next(myWaitTimeBeforeRetryingLowerLimit, myWaitTimeBeforeRetryingUpperLimit);
// Thread.Sleep(waitTime);
// args.sock.BeginConnect(args.ep, new AsyncCallback(myOnConnect), args);
// return;
// }
// }
// //BlueEventLog.logEvent(String.Format("Successful connection made to {0}.", args.serverURL.ToString()), PS3LogLevels.connection);
// args.handler.onConnection(new bts.AsyncTCPConnection_APM(args.sock, myHBTimeSpan, ITCPConnection.ConnectionSide.Client));
// }
int waitTime = myRnd.Next(myWaitTimeBeforeRetryingLowerLimit, myWaitTimeBeforeRetryingUpperLimit);
Thread.Sleep(waitTime);
args.sock.BeginConnect(args.ep, new AsyncCallback(myOnConnect), args);
return;
}
}
//BlueEventLog.logEvent(String.Format("Successful connection made to {0}.", args.serverURL.ToString()), PS3LogLevels.connection);
args.handler.onConnection(new bts.AsyncTCPConnection_APM(args.sock, myHBTimeSpan, ITCPConnection.ConnectionSide.Client));
}
// }
// }
}
}
@@ -5,21 +5,23 @@ using System.Runtime.Serialization.Formatters.Binary;
namespace BCG
{
#pragma warning disable SYSLIB0011 // Type or member is obsolete
public class GenericFormatter<F> : IGenericFormatter where F : IFormatter, new ()
{
IFormatter myFormatter = new F ();
public T Deserialize<T> (Stream serializationStream)
{
#pragma warning disable
return (T) myFormatter.Deserialize (serializationStream);
#pragma warning restore SYSLIB0011
}
public void Serialize<T> (Stream serializationStream, T graph)
{
#pragma warning disable
myFormatter.Serialize (serializationStream, graph);
#pragma warning restore SYSLIB0011
}
}
public class GenericBinaryFormatter : GenericFormatter<BinaryFormatter> { }
#pragma warning restore SYSLIB0011 // Type or member is obsolete
}
+1 -1
View File
@@ -6,7 +6,7 @@
<PropertyGroup>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup>
+5 -8
View File
@@ -4,22 +4,19 @@ using System.IO;
namespace BCG {
/// <summary>
/// this is the data returned from the pubsub server
/// this is the data returned from the
/// pubsub server
/// </summary>
public class PubSubUpdate {
public PubSubUpdate(string tag, string the_tag_type, Stream body) {
this.tag = tag;
this.body = body;
this.tag_type = the_tag_type;
}
public PubSubUpdate(string tag, string tagType, Stream body) { this.tag = tag; this.body = body; this.tagType = tagType; }
public string tag;
public Stream body;
public string tag_type;
public string tagType;
public override string ToString() {
return $"PubSubUpdate tag: {tag} tag_type: {tag_type} len: {body.Position}";
return $"PubSubUpdate tag: {tag} tag_type: {tagType} len: {body.Position}";
}
};
}
@@ -5,16 +5,18 @@ using System.IO;
using System.Collections.Generic;
using mbts;
namespace BCG {
public delegate void PubSubServerConnectionStatusChange(string peerURL, bool connection);
public delegate void PubSubServerDataDelegate(string sender, byte[] data);
namespace BCG
{
public delegate void ConnectionStatusChange(string peerURL, bool connection);
public delegate void DataDelegate(string sender, byte[] data);
/// <summary>
/// Encapsulates PubsubServer's communication with the
/// rest of the world. Has send and get methods, and also
/// an event to tell if a client connects or disconnects.
/// this event is prompted by the doWork.
/// </summary>
public interface IPubsubMessageHub : IWorker {
public interface IPubsubMessageHub : IWorker
{
/// <summary>
/// The port that the message hub is listening on. Return -1 if you're not listening on a port, i.e. for hte case of multicast
@@ -27,14 +29,14 @@ namespace BCG {
/// getUpdates. The delegate will be fired on an arbitrary thread.
/// </summary>
/// <param name="?"></param>
void registerDataCallback(PubSubServerDataDelegate del);
void registerDataCallback(DataDelegate del);
/// <summary>
/// registers a connection callback. this callback
/// will be fired within the doWork call.
/// </summary>
/// <param name="del"></param>
/// <param name="register"></param>
void registerConnectionCallback(PubSubServerConnectionStatusChange del);
void registerConnectionCallback(ConnectionStatusChange del);
/// <summary>
/// this is where this interface starts communication with the
/// rest of the world. Obvioulsy, the resgiters should be
@@ -22,8 +22,8 @@ namespace BCG {
}
private bool myActivated = false;
private PubSubServerConnectionStatusChange myConnCb;
private PubSubServerDataDelegate myDataCb;
private ConnectionStatusChange myConnCb;
private DataDelegate myDataCb;
/// <summary>
/// client to connection
@@ -77,12 +77,12 @@ namespace BCG {
#region IPubsubMessageHub Members
public void registerDataCallback(PubSubServerDataDelegate del) {
public void registerDataCallback(DataDelegate del) {
// BlueEventLog.logAssert(!myActivated, "connection callbacks should be registered before the call to activate");
myDataCb = del;
}
public void registerConnectionCallback(PubSubServerConnectionStatusChange del) {
public void registerConnectionCallback(ConnectionStatusChange del) {
//BlueEventLog.logAssert(!myActivated, "data callbacks should be registered before the call to activate");
myConnCb = del;
}
@@ -4,22 +4,25 @@ using System.Threading;
using System.IO;
using mbts;
namespace BCG {
namespace BCG
{
/// <summary>
/// Wraps around all the IPubsubMessageHub's that pubsub supports.
/// ie MSMQ and TCP to begin with.
/// </summary>
public class ServerMessageHub : IPubsubMessageHub {
public class ServerMessageHub : IPubsubMessageHub
{
private static ServerMessageHub mySelf = null;
private static object myInstanceLock = new object();
private Dictionary<string, IPubsubMessageHub> myHubs = new Dictionary<string, IPubsubMessageHub>();
private SyncQueue<Pair<string, byte[]>> myReceivingQueue = new SyncQueue<Pair<string, byte[]>>();
private PubSubServerConnectionStatusChange myConnectionCb;
private PubSubServerDataDelegate myDataCb;
private ConnectionStatusChange myConnectionCb;
private DataDelegate myDataCb;
private MasterWorker myMasterWorker = MasterWorker.getInstance(false);
protected ServerMessageHub() : this(PubSub3Config.instance.PS3_ServerPort) { }
protected ServerMessageHub(int portNumber) {
protected ServerMessageHub(int portNumber)
{
// TODO: no more MSMQ
// if (!PubSub3Config.instance.PS3_DisableMSMQ)
// myHubs.Add(MSMQSubAddress.prefix, new MSMQMessageHub());
@@ -32,14 +35,19 @@ namespace BCG {
as IPubsubMessageHub);
}
public int port {
public int port
{
get { return myHubs[BlutoSubAddress.prefix].port; }
}
public static ServerMessageHub instance {
get {
if (mySelf == null) {
lock (myInstanceLock) {
public static ServerMessageHub instance
{
get
{
if (mySelf == null)
{
lock (myInstanceLock)
{
if (mySelf == null)
mySelf = new ServerMessageHub();
}
@@ -52,39 +60,47 @@ namespace BCG {
/* Builds the server message hub with a specified port number. Only works if the server message hub has not already been created.
* */
public static ServerMessageHub getInstance(int portNumber) {
if (mySelf == null) {
lock (myInstanceLock) {
public static ServerMessageHub getInstance(int portNumber)
{
if (mySelf == null)
{
lock (myInstanceLock)
{
if (mySelf == null)
mySelf = new ServerMessageHub(portNumber);
}
} else
}
else
throw new ArgumentException("Error: Cannot specify a port number for the ServerMessageHub instance if it has already been created.");
return mySelf;
}
#region IPubsubMessageHub Members
public void registerDataCallback(PubSubServerDataDelegate del) { myDataCb = del; }
public void registerDataCallback(DataDelegate del) { myDataCb = del; }
public void registerConnectionCallback(PubSubServerConnectionStatusChange del) { myConnectionCb = del; }
public void registerConnectionCallback(ConnectionStatusChange del) { myConnectionCb = del; }
public void activate() {
public void activate()
{
string msg = "PubSubServer started .. initializing TCP transport components .. ";
Console.WriteLine(msg);
foreach (IPubsubMessageHub hub in myHubs.Values) {
hub.registerDataCallback(new PubSubServerDataDelegate(myOnData));
hub.registerConnectionCallback(new PubSubServerConnectionStatusChange(myOnConnection));
foreach (IPubsubMessageHub hub in myHubs.Values)
{
hub.registerDataCallback(new DataDelegate(myOnData));
hub.registerConnectionCallback(new ConnectionStatusChange(myOnConnection));
hub.activate();
}
myMasterWorker.addWorker(this, new TimeSpan(0, 0, 0, 0, 250));
}
public List<Pair<string, byte[]>> getUpdates(TimeSpan timeout, int maxElements) {
public List<Pair<string, byte[]>> getUpdates(TimeSpan timeout, int maxElements)
{
return myReceivingQueue.dequeueMultipleElements(timeout, maxElements);
}
public bool send(string receiver, byte[] data) {
public bool send(string receiver, byte[] data)
{
if (receiver.StartsWith(BlutoSubAddress.prefix))
return myHubs[BlutoSubAddress.prefix].send(receiver, data);
else if (receiver.StartsWith(MSMQSubAddress.prefix))
@@ -93,7 +109,8 @@ namespace BCG {
return false;
}
public bool send(string receiver, Stream str) {
public bool send(string receiver, Stream str)
{
if (receiver.StartsWith(BlutoSubAddress.prefix))
return myHubs[BlutoSubAddress.prefix].send(receiver, str);
else if (receiver.StartsWith(MSMQSubAddress.prefix))
@@ -109,7 +126,8 @@ namespace BCG {
/// <summary>
/// connection callback will be called within this doWork
/// </summary>
public void doWork() {
public void doWork()
{
foreach (IPubsubMessageHub hub in myHubs.Values)
hub.doWork();
}
@@ -120,18 +138,21 @@ namespace BCG {
public static string internalPublisherConnection = "InternalPublisher";
public void internalSend(byte[] data) {
public void internalSend(byte[] data)
{
myReceivingQueue.enqueue(new Pair<string, byte[]>(internalPublisherConnection, data));
}
private void myOnData(string sender, byte[] msg) {
private void myOnData(string sender, byte[] msg)
{
if (myDataCb != null)
myOnData(sender, msg);
else
myReceivingQueue.enqueue(new Pair<string, byte[]>(sender, msg));
}
private void myOnConnection(string clientURL, bool connected) {
private void myOnConnection(string clientURL, bool connected)
{
if (myConnectionCb != null)
myConnectionCb(clientURL, connected);
}
@@ -58,7 +58,7 @@ namespace BCG {
myReceivingProcThread = new Thread(new ThreadStart(myReceivingThreadProc));
myMessageHub = ServerMessageHub.instance;
myMessageHub.registerConnectionCallback(new PubSubServerConnectionStatusChange(myHandleConnections));
myMessageHub.registerConnectionCallback(new ConnectionStatusChange(myHandleConnections));
}
@@ -55,7 +55,7 @@ namespace BCG {
myTagToPubsubMap[tag].onSub(sub);
if (onSingleTagChange != null)
onSingleTagChange(new PubSubSingleTagStateEvent(tag, PubSubSingleTagStateEvent.EventType.subscription, sub.connection, sub.subscribing_app));
onSingleTagChange(new PubSubSingleTagStateEvent(tag, PubSubSingleTagStateEvent.EventType.subscription, sub.connection, sub.subscribingApp));
}
if (onTagMapChanged != null)
onTagMapChanged(sub.tags);
@@ -68,10 +68,10 @@ namespace BCG {
///and publishers for the tag then remove it.
myTagToPubsubMap[tag].onUnsub(unsub);
if (onSingleTagChange != null)
onSingleTagChange(new PubSubSingleTagStateEvent(tag, PubSubSingleTagStateEvent.EventType.unsubscription, unsub.connection, unsub.unsubscribing_app));
onSingleTagChange(new PubSubSingleTagStateEvent(tag, PubSubSingleTagStateEvent.EventType.unsubscription, unsub.connection, unsub.unsubscribingApp));
} else {
string warningMsg = String.Format("{0} was never published or subscribed on this pubsub, hence" +
" the unsub call by app {1} is voided", tag, unsub.unsubscribing_app);
" the unsub call by app {1} is voided", tag, unsub.unsubscribingApp);
//BlueEventLog.logEvent(warningMsg, BlueEventLevel.Warning);
}
}
@@ -1,19 +1,19 @@
using System;
using System.Collections.Generic;
using System.Text;
using BCG.Utilities;
using mbts;
namespace BCG {
namespace BCG
{
/// <summary>
/// SingleTagPubSubMap consists of all the state w.r.t to
/// publishing and subscription of a single tag. this information
/// consists of the publisher, and a map of the subscriber
/// connection vs the apps they're coming from
/// </summary>
internal class SingleTagPubSubMap {
internal class SingleTagPubSubMap
{
private string myTag;
/// <summary>
/// map of clientURL to client app (machine;app;pid;clientURL)
@@ -29,7 +29,8 @@ namespace BCG {
/// </summary>
List<Pair<string, string>> myDeadPubConnToApp = new List<Pair<string, string>>();
public SingleTagPubSubMap(string tag) {
public SingleTagPubSubMap(string tag)
{
myTag = tag;
}
@@ -37,7 +38,8 @@ namespace BCG {
/// Returns a list all the active connections that
/// have subscribed to this tag
/// </summary>
public ICollection<string> subscriberConnections {
public ICollection<string> subscriberConnections
{
get { return mySubConnToApp.Keys; }
}
@@ -45,7 +47,8 @@ namespace BCG {
/// Returns a list of all the applications that have
/// subscribed to this tag
/// </summary>
public ICollection<string> subscripingApps {
public ICollection<string> subscripingApps
{
get { return mySubConnToApp.Values; }
}
@@ -54,7 +57,8 @@ namespace BCG {
/// connections as key and the application
/// as value
/// </summary>
public Dictionary<string, string> subscribers {
public Dictionary<string, string> subscribers
{
get { return mySubConnToApp; }
}
@@ -62,7 +66,8 @@ namespace BCG {
/// Returns a dictionary for all active publishing
/// connection as key and the application as value
/// </summary>
public Dictionary<string, string> activePublishers {
public Dictionary<string, string> activePublishers
{
get { return myActivePubConnToApp; }
}
@@ -70,7 +75,8 @@ namespace BCG {
/// Returns a list of pair of dead connections and
/// theur corresponding applications
/// </summary>
public ICollection<Pair<string, string>> deadPublishers {
public ICollection<Pair<string, string>> deadPublishers
{
get { return myDeadPubConnToApp; }
}
@@ -81,14 +87,18 @@ namespace BCG {
/// </summary>
/// <param name="sub"></param>
/// <returns></returns>
public bool onSub(SubMetaData sub) {
if (mySubConnToApp.ContainsKey(sub.connection)) {
public bool onSub(SubMetaData sub)
{
if (mySubConnToApp.ContainsKey(sub.connection))
{
string warningMsg = String.Format("tag {0} has already been subscribed by app {0}, subscriber {1}"
+ "it should unsubscribe before resubscribing", myTag, sub.connection, sub.subscribing_app);
+ "it should unsubscribe before resubscribing", myTag, sub.connection, sub.subscribingApp);
//BlueEventLog.logEvent(warningMsg, BlueEventLevel.Warning);
return false;
} else {
mySubConnToApp.Add(sub.connection, sub.subscribing_app);
}
else
{
mySubConnToApp.Add(sub.connection, sub.subscribingApp);
return true;
}
}
@@ -100,13 +110,17 @@ namespace BCG {
/// </summary>
/// <param name="unsub"></param>
/// <returns></returns>
public bool onUnsub(UnSubMetaData unsub) {
if (mySubConnToApp.ContainsKey(unsub.connection)) {
public bool onUnsub(UnSubMetaData unsub)
{
if (mySubConnToApp.ContainsKey(unsub.connection))
{
mySubConnToApp.Remove(unsub.connection);
return true;
} else {
}
else
{
string warningMsg = String.Format("tag {0} has already was either unsubscribed or never subscribed by app {0}, subscriber {1}",
myTag, unsub.connection, unsub.unsubscribing_app);
myTag, unsub.connection, unsub.unsubscribingApp);
//BlueEventLog.logEvent(warningMsg, BlueEventLevel.Warning);
return false;
}
@@ -119,20 +133,25 @@ namespace BCG {
/// </summary>
/// <param name="subscriber"></param>
/// <returns></returns>
public bool onConnectionRemoveIfExists(string connection, out bool wasPublisher, out bool wasSubscriber) {
public bool onConnectionRemoveIfExists(string connection, out bool wasPublisher, out bool wasSubscriber)
{
bool rv = false;
if (mySubConnToApp.ContainsKey(connection)) {
if (mySubConnToApp.ContainsKey(connection))
{
mySubConnToApp.Remove(connection);
wasSubscriber = true;
rv = true;
} else
}
else
wasSubscriber = false;
if (myActivePubConnToApp.ContainsKey(connection)) {
if (myActivePubConnToApp.ContainsKey(connection))
{
myDeadPubConnToApp.Add(new Pair<string, string>(connection, myActivePubConnToApp[connection]));
myActivePubConnToApp.Remove(connection);
rv = true;
wasPublisher = true;
} else
}
else
wasPublisher = false;
return rv;
}
@@ -143,10 +162,14 @@ namespace BCG {
/// </summary>
/// <param name="pub"></param>
/// <param name="publisherChanged"></param>
public void onPub(PubMetaData pub, out bool newPublisher) {
if (myActivePubConnToApp.ContainsKey(pub.connection)) {
public void onPub(PubMetaData pub, out bool newPublisher)
{
if (myActivePubConnToApp.ContainsKey(pub.connection))
{
newPublisher = false;
} else {
}
else
{
myActivePubConnToApp.Add(pub.connection, pub.publishingApp);
newPublisher = true;
}
@@ -6,79 +6,92 @@ using BCG.Utilities;
using BCG.Utilities.General;
using mbts;
using mbts.spark;
using System.Linq;
namespace BCG {
/// <summary>
/// Deals with the subscriptions (and getting data) from one sole
/// pubsub server. Theres no sharing of conenctions
/// Deals with the subscriptions (and getting data) from one
/// pubsub server. Once again, theres no sharing of conenctions
/// from this point. Every new instance of SingleServerSubscriberCore
/// will open its new dedicated connection to the serverURL that
/// its concerned with (passed as a conctructor arg).
/// </summary>
internal interface ISingleServerSubscriberCore : IDisposable {
IList<PubSubUpdate> get_updates(int timeout);
IList<PubSubUpdate> getUpdates(int timeout);
event PubSubConnectionChange onPubSubConnectionChange;
void subscribe(params string[] tags);
void unsubscribe(params string[] tags);
}
internal class SingleServerSubscriberCore : IDisposable, ISingleServerSubscriberCore {
private readonly IPubSubServerConnection _connection;
private HashSet<string> _tags = new HashSet<string>();
private HighPerformanceTimer myTimer = new HighPerformanceTimer();
private PubSubServerConnectionFactory myConnectionFactory;
private IPubSubServerConnection myConnection;
private List<string> myTags = new List<string>();
public event PubSubConnectionChange onPubSubConnectionChange;
public SingleServerSubscriberCore(string server_url, PubSubServerConnectionFactory connection_factory) {
_connection = connection_factory.getConnection(server_url);
_connection.register_connection_callback(new PubSubServerConnectionStatusChange(_handle_server_reboot));
public SingleServerSubscriberCore(string serverURL, PubSubServerConnectionFactory connectionFactory) {
myConnectionFactory = connectionFactory;
myConnection = myConnectionFactory.getConnection(serverURL);
myConnection.registerCallback(new ConnectionStatusChange(myHandleServerReboot));
}
// gets as many data items that it can get in
// timeout milliseconds
public IList<PubSubUpdate> get_updates(int timeout) {
_connection.doWork();
List<PubSubUpdate> rv = new List<PubSubUpdate>();
IList<byte[]> data_items = _connection.get_updates(new TimeSpan(0, 0, 0, 0, timeout), Int32.MaxValue);
foreach (byte[] data in data_items) {
process_message(data, ref rv);
}
return rv;
public IList<PubSubUpdate> getUpdates() {
return getUpdates(1);
}
/// <summary>
/// gets as many data items that it can get in
/// timeout milliseconds
/// </summary>
/// <param name="timeout"></param>
/// <returns></returns>
public IList<PubSubUpdate> getUpdates(int timeout) {
myConnection.doWork();
List<PubSubUpdate> retVal = new List<PubSubUpdate>();
IList<byte[]> dataItems = myConnection.getUpdates(new TimeSpan(0, 0, 0, 0, timeout), Int32.MaxValue);
foreach (byte[] data in dataItems) {
processMessage(data, retVal);
}
return retVal;
}
public List<string> getSubscribedTags() {
return myTags;
}
public void subscribe(params string[] tags) {
HashSet<string> this_call_tags = new HashSet<string>();
foreach (string tag in tags) {
if (_tags.Contains(tag)) {
bts.Log.Gen.warn($"SingleServerSubscriberCore.subscribe tag: {tag} has already " +
$"been subscribed to server: {_connection.server_url}");
if (myTags.Contains(tag)) {
string errMsg = String.Format("tag {0} was already subscribed from server {1}",
tag, myConnection.serverURL);
// BlueEventLog.logEvent(errMsg, BlueEventLevel.Warning);
} else {
_tags.Add(tag);
this_call_tags.Add(tag);
myTags.Add(tag);
}
}
_subscribe(this_call_tags.ToArray());
mySubscribe(tags);
}
public void unsubscribe(params string[] tags) {
List<string> unsub_tags = new List<string>();
List<string> unsubTags = new List<string>();
foreach (string tag in tags) {
if (_tags.Contains(tag)) {
_tags.Remove(tag);
unsub_tags.Add(tag);
if (myTags.Contains(tag)) {
myTags.Remove(tag);
unsubTags.Add(tag);
} else {
bts.Log.Gen.warn($"tag {tag} is being unsubscribed from server {_connection.server_url}, eventhough " +
"it was never subscribed");
string errMsg = String.Format("tag {0} is being unsubscribed from server {1}, eventhough " +
"it was never subscribed", tag, myConnection.serverURL);
// BlueEventLog.logEvent(errMsg, BlueEventLevel.Warning);
}
}
UnSubMetaData unsub = new UnSubMetaData(_connection.client_url, unsub_tags.ToArray());
UnSubMetaData unsub = new UnSubMetaData(myConnection.clientURL, unsubTags.ToArray());
Stream str = new MemoryStream();
new MetaData(unsub).save(str);
str.Position = 0;
// this is where data is actually sent out
_connection.send(str);
myConnection.send(str);
}
#region IDisposable Members
@@ -86,40 +99,44 @@ namespace BCG {
public void Dispose() {
///remove the subscriber
//myConnection.send(new PubSubMessage(new MetaData(new RemoveSubscriberMetaData(myConnection.clientURL))).toStream());
_connection.Dispose();
myConnection.Dispose();
//clear the list of tags you've subscribed to. This way, if you're ever used again,
//you have to send all the subscription metadata out again.
_tags.Clear();
myTags.Clear();
}
#endregion
// The only place where a tag is subsribed from the Pubsub server. the subscriptions are wrapped into SubMetaData,
// which has a to_stream method
private void _subscribe(params string[] tags) {
SubMetaData sub = new SubMetaData(_connection.client_url, tags);
private void mySubscribe(params string[] tags) {
SubMetaData sub = new SubMetaData(myConnection.clientURL, tags);
Stream str = new MemoryStream();
new MetaData(sub).save(str);
str.Position = 0;
_connection.send(str);
myConnection.send(str);
}
private void _handle_server_reboot(string server_url, bool connection) {
bts.Log.Gen.info($"SingleServerSubscriberCore._handle_server_reboot connection: {connection} server_url: {server_url}");
private void myHandleServerReboot(string serverURL, bool connection) {
if (connection) {
_subscribe(_tags.ToArray());
//ie connection established again, so has to
//resubscribe to all the tags
// BlueEventLog.logEvent("Connection to " + serverURL + " established....resubscribing to all the tags",
// PS3LogLevels.disconnection);
mySubscribe(myTags.ToArray());
} else {
bts.Log.Gen.warn($"SingleServerSubscriberCore._handle_server_reboot connection to {server_url} broke....attempting to reconnect...");
// BlueEventLog.logEvent("Connection to " + serverURL + " broke....attempting to reconnect...",
// PS3LogLevels.disconnection);
}
//let anyone watching you know that your connection has changed.
if (onPubSubConnectionChange != null)
onPubSubConnectionChange(connection, server_url);
onPubSubConnectionChange(connection, serverURL);
}
// processes one message
internal static void process_message(byte[] data, ref List<PubSubUpdate> updates) {
bts.Log.Gen.info($"SingleServerSubscriberCore.process_message data_len: {data.Length}");
internal static void processMessage(byte[] data, List<PubSubUpdate> updates) {
bts.Log.Gen.info($"SingleServerSubscriberCore.processMessage data_len: {data.Length}");
if (data != null && data.Length > 0) {
MemoryStream str = new MemoryStream(data);
MetaData md = new MetaData(str);
@@ -139,8 +156,9 @@ namespace BCG {
if (p.isCompressed) {
long compressedSize = bodyStream.Length - bodyStream.Position;
if (compressedSize == 0) {
string error_msg = "Error: tag " + p.tag + " had compressed size of 0 bytes. This shouldnt' happen. Not decompressing tag or sending out update.";
bts.Log.Gen.error(error_msg);
string errorMsg = "Error: tag " + p.tag + " had compressed size of 0 bytes. This shouldnt' happen. Not decompressing tag or sending out update.";
Console.WriteLine(errorMsg);
//BlueEventLog.logEvent(errorMsg, BlueEventLevel.Error);
return;
}
realBody = StreamDecompressor.decompressStream(bodyStream);
@@ -156,9 +174,8 @@ namespace BCG {
bodyStream.Read(trConstituent, 0, trConstituent.Length);
updates.Add(new PubSubUpdate(pub.tag, pub.type, new MemoryStream(trConstituent)));
}
} else {
bts.Log.assert_d(false, $"Unknown message received: {md} only Pub and TransactionMetaData are legal");
}
} else { }
//BlueEventLog.logAssert(false, "Unknown message received: " + md.ToString() + " , only Pub and TransactionMetaData are legal");
}
}
}
@@ -387,18 +387,18 @@ namespace BCG {
SharpZipCompressor Kompressor = new SharpZipCompressor();
Kompressor.compress(content, contentToPublish);
contentToPublish.Position = 0;
pub_md = new PubMetaData(tagType, tag, (int)contentToPublish.Length, (int)content.Length, serverConn.server_url, ClientInfo.thisClient.processString);
pub_md = new PubMetaData(tagType, tag, (int)contentToPublish.Length, (int)content.Length, serverConn.serverURL, ClientInfo.thisClient.processString);
if (PubSub3Config.instance.shouldReportTagsCompressionRatio(tag)) {
string msg = String.Format("Compressed tag {0} from {1} to {2} bytes, for a ratio of {3}.", tag, content.Length, contentToPublish.Length, (float)content.Length / (float)contentToPublish.Length);
// BlueEventLog.logEvent(msg, PS3LogLevels.compressionRatio);
}
} else
pub_md = new PubMetaData(tagType, tag, (int)content.Length, serverConn.server_url, ClientInfo.thisClient.processString);
pub_md = new PubMetaData(tagType, tag, (int)content.Length, serverConn.serverURL, ClientInfo.thisClient.processString);
content.Position = initPos;
if (CrushedTagManager.instance.is_tag_crushable(tag)) {
string refId = String.Empty;
Console.WriteLine("StaticPublisherCore.mySend serverConn.sendWithConfirmation tag: " + tag + " server: " + serverConn.server_url);
Console.WriteLine("StaticPublisherCore.mySend serverConn.sendWithConfirmation tag: " + tag + " server: " + serverConn.serverURL);
// this is where the data is actually sent out (there are multiple sends)
PubSubMessage ps_msg = new PubSubMessage(new MetaData(pub_md), contentToPublish);
//Console.WriteLine($"serverConn.send_with_confirmation metadata: {ps_msg.meta_data}");
@@ -428,8 +428,8 @@ namespace BCG {
myTagToSender.Add(tag, sender);
if (!mySenders.Contains(sender)) {
mySenders.Add(sender);
sender.register_connection_callback(new PubSubServerConnectionStatusChange(myHandleServerReboot));
sender.register_send_confirmation_callback(new SendConfirmationDelegate(myOnSendConfirm));
sender.registerCallback(new ConnectionStatusChange(myHandleServerReboot));
sender.registerSendConfirmationCallback(new SendConfirmationDelegate(myOnSendConfirm));
}
}
@@ -443,7 +443,7 @@ namespace BCG {
if (connection) {
string thisPublisherInfo = ClientInfo.thisClient.processString;
foreach (KeyValuePair<string, IPubSubServerConnection> p in myTagToSender) {
if (p.Value.server_url == serverURL) {
if (p.Value.serverURL == serverURL) {
string tagType;
string tag = p.Key;
myCrushableTagsWaitingOnAck.Clear();
@@ -80,7 +80,7 @@ namespace BCG {
int timeShare = timeout / myServerURLToSubCore.Count;
List<PubSubUpdate> retVal = new List<PubSubUpdate>();
foreach (ISingleServerSubscriberCore subCore in myServerURLToSubCore.Values)
retVal.AddRange(subCore.get_updates(timeShare));
retVal.AddRange(subCore.getUpdates(timeShare));
_log_n_updates(retVal);
myTimer.stop();
//if (myTimer.lastIntervalTime > 10)
+1 -1
View File
@@ -15,7 +15,7 @@ namespace BCG {
/// </summary>
[STAThread]
public static void Main(string[] args) {
bool run_echo_tcp_server_only = false;
bool run_echo_tcp_server_only = true;
bts.utils.BTSNativeMethods.provide_command_line(new string[] { }, 0);
mbts.config.Configuration.initialize();
if (run_echo_tcp_server_only) {
@@ -8,7 +8,7 @@
</ItemGroup>
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,20 +1,26 @@
namespace BCG {
namespace BCG
{
using bts.api.orderman;
enum TSGroupByOperation {
enum TSGroupByOperation
{
OHLC = 0,
Statistics
}
[DataStreamUserRespresentation("timeseriesanalysis")]
public class TimeSeriesGroupByStream : ViewStream {
public TimeSeriesGroupByStream(ViewStream md_str, string timebucket_col, string output_col, string operation) {
public class TimeSeriesGroupByStream : ViewStream
{
public TimeSeriesGroupByStream(ViewStream md_str, string timebucket_col, string output_col, string operation)
{
ViewStream rs = new NonIncGroupByStream(md_str, new OHLCNonIncGroupByEvaluator(timebucket_col, output_col));
mySetRealSource(rs);
}
}
public class OHLC {
public OHLC(double open_args, double high_args, double low_args, double close_args) {
public class OHLC
{
public OHLC(double open_args, double high_args, double low_args, double close_args)
{
open = open_args;
high = high_args;
low = low_args;
@@ -22,36 +28,47 @@ namespace BCG {
}
public readonly double open, high, low, close;
}
public class OHLCNonIncGroupByEvaluator : INonIncGroupByEvaluator {
public class OHLCNonIncGroupByEvaluator : INonIncGroupByEvaluator
{
private string _timestamp_col;
private string _output_col;
private IBlueTableSchema _out_schema;
public OHLCNonIncGroupByEvaluator(string timestamp_col, string output_col) {
public OHLCNonIncGroupByEvaluator(string timestamp_col, string output_col)
{
_timestamp_col = timestamp_col;
_output_col = output_col;
}
string open_col {
get {
string open_col
{
get
{
return $"{_output_col}_open"; ;
}
}
string high_col {
get {
string high_col
{
get
{
return $"{_output_col}_high"; ;
}
}
string low_col {
get {
string low_col
{
get
{
return $"{_output_col}_low"; ;
}
}
string close_col {
get {
string close_col
{
get
{
return $"{_output_col}_close"; ;
}
}
private void _build_schema(IBlueTableSchema input_schema) {
private void _build_schema(IBlueTableSchema input_schema)
{
List<Tuple<Type, string>> new_cols = new List<Tuple<Type, string>>();
ColumnInfo ci = input_schema[_output_col];
new_cols.Add(Tuple.Create(ci.colType, open_col));
@@ -62,25 +79,32 @@ namespace BCG {
_out_schema = BlueDataUtility.add_columns(input_schema, new_cols.ToArray());
}
public IBlueRowCollection evaluate(IDictionary<int, List<IBlueRow>> grouped_rows) {
if (_out_schema == null) {
if (grouped_rows.Count > 0) {
foreach (var kv in grouped_rows) {
public IBlueRowCollection evaluate(IDictionary<int, List<IBlueRow>> grouped_rows)
{
if (_out_schema == null)
{
if (grouped_rows.Count > 0)
{
foreach (var kv in grouped_rows)
{
var rows = kv.Value;
if (rows != null && rows.Count > 0) {
if (rows != null && rows.Count > 0)
{
_build_schema(rows.ElementAt(0).schema);
}
}
}
}
BlueTable rv = new BlueTable(_out_schema);
foreach (var kv in grouped_rows) {
foreach (var kv in grouped_rows)
{
int id = kv.Key;
IEnumerable<IBlueRow> rows = kv.Value;
// sort rows on time to compute o,c correctly??
if (rows != null && rows.Count() > 0) {
if (rows != null && rows.Count() > 0)
{
IBlueRow reprow = rows.ElementAt(0);
IEnumerable<double> vals = BlueDataUtility.get_colvalues<double>(rows, _output_col);
IEnumerable<double> vals = BlueDataUtility.get_covalues<double>(rows, _output_col);
OHLC ohlc = compute_ohlc(vals);
MutableBlueRow nr = new MutableBlueRow(_out_schema, reprow);
nr.setValue("id", kv.Key);
@@ -94,11 +118,13 @@ namespace BCG {
return rv;
}
public string[] get_group_by_cols() {
public string[] get_group_by_cols()
{
return new string[] { _timestamp_col };
}
static OHLC compute_ohlc(IEnumerable<double> prices) {
static OHLC compute_ohlc(IEnumerable<double> prices)
{
double o = prices.ElementAt(0);
double c = prices.ElementAt(prices.Count() - 1);
double l = prices.Min();
@@ -107,22 +133,28 @@ namespace BCG {
}
}
[DataStreamUserRespresentation("RSI")]
public class RSIStream : ViewStream {
public class RSIStream : ViewStream
{
// for ms_str expects a timeseries with a timestamp col and an output col (price col) to calculate the ohlc for
public RSIStream(ViewStream md_str, string timebucket_column, string output_col) {
public RSIStream(ViewStream md_str, string timebucket_column, string output_col)
{
}
}
[DataStreamUserRespresentation("Bollinger")]
public class BollingerBandStream : ViewStream {
public class BollingerBandStream : ViewStream
{
// for ms_str expects a timeseries with a timestamp col and an output col (price col) to calculate the ohlc for
public BollingerBandStream(ViewStream md_str, string timebucket_column, string output_col) {
public BollingerBandStream(ViewStream md_str, string timebucket_column, string output_col)
{
}
}
[DataStreamUserRespresentation("OHLC")]
public class OHLCStream : ViewStream {
public class OHLCStream : ViewStream
{
// for ms_str expects a timeseries with a timestamp col and an output col (price col) to calculate the ohlc for
public OHLCStream(ViewStream md_str, string timebucket_column, string output_col) {
public OHLCStream(ViewStream md_str, string timebucket_column, string output_col)
{
ViewStream rs = new TimeSeriesGroupByStream(md_str, timebucket_column, output_col, TSGroupByOperation.OHLC.ToString());
mySetRealSource(rs);
}
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<Version>1.0.0</Version>
<AssemblyVersion>1.0.0</AssemblyVersion>
</PropertyGroup>
@@ -2,7 +2,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFrameworks>net8.0</TargetFrameworks>
<TargetFrameworks>net7.0</TargetFrameworks>
<Nullable>enable</Nullable>
<RuntimeIdentifiers>win10-x64;ubuntu.16.04-x64</RuntimeIdentifiers>
<Authors>Blue Trading Systems</Authors>
+1 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0-windows</TargetFramework>
<TargetFramework>net7.0-windows</TargetFramework>
<UseWPF>true</UseWPF>
<Authors>Blue Trading Systems</Authors>
<Company>Blue Trading Systems</Company>
@@ -320,6 +320,7 @@
internal uint security_count;
internal bool has_depth;
internal Platform platform;
internal ulong trade_count; // atomic
internal ulong coconut_status_count; // atomic
internal ulong exchange_status_count; // atomic
+1 -1
View File
@@ -1,6 +1,6 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<Authors>Blue Trading Systems</Authors>
<Company>Blue Trading Systems</Company>
<Description />
@@ -2,7 +2,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<Nullable>enable</Nullable>
<Platforms>x64</Platforms>
<PlatformTarget>x64</PlatformTarget>
@@ -215,6 +215,9 @@ namespace System.Collections.Generic {
);
}
public static Dictionary<T1, T2> ToDictionary<T1, T2>(this IEnumerable<(T1, T2)> source) =>
source.ToDictionary(tup => tup.Item1, tup => tup.Item2);
/// <summary>
/// Gets the value associated with the specified key, then removes the key if the lookup was successful.
/// </summary>
@@ -652,12 +652,12 @@ namespace System.Collections.Generic {
public static IReadOnlyList<T> ToReadOnlyList<T>(this IEnumerable<T> enumerable) =>
enumerable.ToList();
// public static Dictionary<TKey, T> ToDictionary<T, TKey>(this IEnumerable<KeyValuePair<TKey, T>> key_values)
// where TKey : notnull =>
// key_values.ToDictionary(kv => kv.Key, kv => kv.Value);
public static Dictionary<TKey, T> ToDictionary<T, TKey>(this IEnumerable<KeyValuePair<TKey, T>> key_values)
where TKey : notnull =>
key_values.ToDictionary(kv => kv.Key, kv => kv.Value);
// public static Dictionary<TKey, T> ToDictionary<T, TKey>(this IEnumerable<KeyValuePair<TKey, T>> key_values, IEqualityComparer<TKey> comparer) where TKey : notnull =>
// key_values.ToDictionary(kv => kv.Key, kv => kv.Value, comparer);
public static Dictionary<TKey, T> ToDictionary<T, TKey>(this IEnumerable<KeyValuePair<TKey, T>> key_values, IEqualityComparer<TKey> comparer) where TKey : notnull =>
key_values.ToDictionary(kv => kv.Key, kv => kv.Value, comparer);
public static Dictionary<TKey, T> ToDictionaryCrushed<T, TKey>(this IEnumerable<T> source, Func<T, TKey> key_selector) where TKey : notnull {
var rv = new Dictionary<TKey, T>();
+1 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<Platforms>x64</Platforms>
<PlatformTarget>x64</PlatformTarget>
+7 -7
View File
@@ -1,9 +1,9 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<OutputType>Exe</OutputType>
<GeneratePackageOnBuild Condition="$(Configuration) == 'Release' or $(Configuration) == 'Release'">true</GeneratePackageOnBuild>
<PackageOutputPath>$(OutputPath)\net8.0</PackageOutputPath>
<PackageOutputPath>$(OutputPath)\net7.0</PackageOutputPath>
<IncludeSource>true</IncludeSource>
<IncludeSymbols>true</IncludeSymbols>
<TargetsForTfmSpecificBuildOutput>$(TargetsForTfmSpecificBuildOutput);CopyProjectReferencesToPackage</TargetsForTfmSpecificBuildOutput>
@@ -43,19 +43,19 @@
<Target Name="CopyProjectReferencesToPackage" DependsOnTargets="ResolveReferences" Condition=" '$(SparkRoot)' != '' ">
<ItemGroup>
<BuildOutputInPackage Include="@(ReferenceCopyLocalPaths-&gt;WithMetadataValue('ReferenceSourceTarget', 'ProjectReference'))" />
<BuildOutputInPackage Include="$(OutputPath)..\net8.0\libspark.dll" />
<BuildOutputInPackage Include="$(OutputPath)..\net7.0\libspark.dll" />
</ItemGroup>
</Target>
<!-- deployed sdk build net8.0 -->
<ItemGroup Condition=" '$(SparkRoot)' == '' and '$(TargetFramework)' == 'net8.0'">
<!-- deployed sdk build net7.0 -->
<ItemGroup Condition=" '$(SparkRoot)' == '' and '$(TargetFramework)' == 'net7.0'">
<Reference Include="spark_api">
<HintPath>..\..\lib\net8.0\spark_api.dll</HintPath>
<HintPath>..\..\lib\net7.0\spark_api.dll</HintPath>
</Reference>
</ItemGroup>
<Target Name="copy_lib_deployed" AfterTargets="PostBuildEvent" Condition=" '$(SparkRoot)' == '' ">
<Copy SourceFiles="..\..\lib\net8.0\libspark.dll" DestinationFolder="$(OutputPath)\" SkipUnchangedFiles="true" />
<Copy SourceFiles="..\..\lib\net7.0\libspark.dll" DestinationFolder="$(OutputPath)\" SkipUnchangedFiles="true" />
</Target>
</Project>
+1
View File
@@ -155,6 +155,7 @@ namespace mbts.co2.tests
Assert.IsTrue(header.heartbeat.time > 0);
Assert.IsTrue(header.security_count == 77000);
Assert.IsTrue(header.has_depth == true);
Assert.IsTrue(header.platform == Platform.Sim);
Assert.IsTrue(header.trade_count == 100);
Assert.IsTrue(header.coconut_status_count == 101);
Assert.IsTrue(header.exchange_status_count == 102);
+1 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<Platforms>x64</Platforms>
<PlatformTarget>x64</PlatformTarget>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.0</TargetFramework>
<Nullable>enable</Nullable>
<Platforms>x64</Platforms>
<PlatformTarget>x64</PlatformTarget>
@@ -7,6 +7,13 @@ HistoricMarketDataConfig {
}
}
PlatformMap {
value {
glbx_platform: "PLATFORM_MDP3ONIXS"
ice_platform: "PLATFORM_ICEONIXS"
}
}
PublisherOptions {
value {
network_interface: "0.0.0.0"
@@ -202,6 +209,14 @@ HistoricMarketDataConfig {
}
}
PlatformMap {
rule: "DeploymentType.type == 'prod'"
value {
glbx_platform: "PLATFORM_MDP3ONIXS"
ice_platform: "PLATFORM_ICEONIXS"
}
}
MarketDataConfig {
value {
emumba_deribit {
@@ -222,4 +237,4 @@ MarketDataConfig {
value{
onixs_license: "{default}"
}
}
}
+7
View File
@@ -12,6 +12,13 @@ message HistoricMarketDataConfig {
bool use_sim = 4;
}
message PlatformMap {
// PLATFORM_MDP3ONIXS
string glbx_platform = 1;
// PLATFORM_ICEONIXS
string ice_platform = 2;
}
// LOCAL TIME
message MDBootDelayWindow {
uint32 delay_seconds = 1;
+26 -45
View File
@@ -150,53 +150,34 @@ bts::api::SecurityType DomainSecurityUtils::get_type(
}
std::string DomainSecurityUtils::get_exchange_id(const bts::api::Security &s) {
std::string rv;
bts_assert(get_security_desc(s, rv));
return rv;
}
bool DomainSecurityUtils::get_security_desc(
const bts::api::Security &security, std::string &desc) {
switch (security.ExchangeData_case()) {
case bts::api::Security::ExchangeDataCase::EXCHANGEDATA_NOT_SET:
switch (s.ExchangeData_case()) {
case (bts::api::Security::ExchangeDataCase::EXCHANGEDATA_NOT_SET):
bts_assert(false);
case bts::api::Security::ExchangeDataCase::kCboe:
desc = security.cboe().security_desc();
break;
case bts::api::Security::ExchangeDataCase::kCfe:
desc = security.cfe().security_desc();
break;
case bts::api::Security::ExchangeDataCase::kDeribit:
desc = security.deribit().security_desc();
break;
case bts::api::Security::ExchangeDataCase::kGlbx:
desc = security.glbx().security_desc();
break;
case bts::api::Security::ExchangeDataCase::kIce:
desc = security.ice().security_desc();
break;
case bts::api::Security::ExchangeDataCase::kBinance:
desc = security.binance().security_desc();
break;
case bts::api::Security::ExchangeDataCase::kBinanceUs:
desc = security.binance_us().security_desc();
break;
case bts::api::Security::ExchangeDataCase::kCoinbase:
desc = security.coinbase().security_desc();
break;
case bts::api::Security::ExchangeDataCase::kBinanceUsdsFutures:
desc = security.binance_usds_futures().security_desc();
break;
case bts::api::Security::ExchangeDataCase::kOkx:
desc = security.okx().security_desc();
break;
case bts::api::Security::ExchangeDataCase::kFtx:
desc = security.ftx().security_desc();
break;
default:
return false;
case (bts::api::Security::ExchangeDataCase::kCboe):
return s.cboe().security_desc();
case (bts::api::Security::ExchangeDataCase::kCfe):
return s.cfe().security_desc();
case (bts::api::Security::ExchangeDataCase::kDeribit):
return s.deribit().security_desc();
case (bts::api::Security::ExchangeDataCase::kGlbx):
return s.glbx().security_desc();
case (bts::api::Security::ExchangeDataCase::kIce):
return s.ice().security_desc();
case (bts::api::Security::ExchangeDataCase::kBinance):
return s.binance().security_desc();
case (bts::api::Security::ExchangeDataCase::kBinanceUs):
return s.binance_us().security_desc();
case (bts::api::Security::ExchangeDataCase::kCoinbase):
return s.coinbase().security_desc();
case (bts::api::Security::ExchangeDataCase::kBinanceUsdsFutures):
return s.binance_usds_futures().security_desc();
case (bts::api::Security::ExchangeDataCase::kOkx):
return s.okx().security_desc();
case (bts::api::Security::ExchangeDataCase::kFtx):
return s.ftx().security_desc();
}
return true;
bts_assert(false);
return "";
}
} // namespace bts::domain
-2
View File
@@ -27,8 +27,6 @@ class DomainSecurityUtils {
const bts::api::SpreadContract &spr);
static bts::api::SecurityType get_type(const bts::api::Security &s);
static std::string get_exchange_id(const bts::api::Security &s);
static bool get_security_desc(
const bts::api::Security &security, std::string &desc);
};
} // namespace bts::domain
+40 -18
View File
@@ -78,20 +78,33 @@ bool is_heart_beat_ok(const ColdHeader &);
class MDConsumerInst {
public:
MDConsumerInst(
const string &symbol, SecurityType type, Venue exchange, bool shared)
: _symbol(symbol), _type(type), _exchange(exchange), _security_count(0),
_found_keys(), _hot(), _hot_depth(),
_cold(ShmSegmentName::cold_name(symbol, type, exchange), shared),
const string &symbol,
SecurityType type,
Venue exchange,
bool shared,
co2::Platform platform)
: _symbol(symbol), _type(type), _exchange(exchange),
_platform(
platform == co2::PLATFORM_UNSET ? platform_for_exchange(exchange)
: platform),
_security_count(0), _found_keys(), _hot(), _hot_depth(),
_cold(
ShmSegmentName::cold_name(symbol, type, exchange, _platform),
shared),
_queues() {
LOG(md, Diag,
"MDConsumerInst::MDConsumerInst symbol: {} venue: {} type: {}",
symbol, static_cast<int>(exchange), static_cast<int>(type));
"MDConsumerInst::MDConsumerInst symbol: {} venue: {} type: {} "
"platform: {}",
symbol, static_cast<int>(exchange), static_cast<int>(type),
static_cast<int>(platform));
if (enable_depth(symbol, type)) {
_hot_depth.reset(new utils::SharedListReader<HotHeader, HotDepthEntry>(
ShmSegmentName::hot_name(symbol, type, exchange), shared));
ShmSegmentName::hot_name(symbol, type, exchange, _platform),
shared));
} else {
_hot.reset(new utils::SharedListReader<HotHeader, HotEntry>(
ShmSegmentName::hot_name(symbol, type, exchange), shared));
ShmSegmentName::hot_name(symbol, type, exchange, _platform),
shared));
}
}
~MDConsumerInst() {
@@ -252,6 +265,7 @@ class MDConsumerInst {
const string &symbol() const { return _symbol; }
SecurityType type() const { return _type; }
Venue exchange() const { return _exchange; }
co2::Platform platform() const { return _platform; }
//
// Queued data
@@ -327,6 +341,7 @@ class MDConsumerInst {
string _symbol;
SecurityType _type;
Venue _exchange;
co2::Platform _platform;
size_t _security_count;
set<security_key> _found_keys;
unique_ptr<utils::SharedListReader<HotHeader, HotEntry>> _hot;
@@ -577,6 +592,7 @@ class MDConsumerImpl {
map<security_key, shared_ptr<MDConsumerHandle>> _deferred_handles;
vector<SecurityCallback> _callbacks;
mutable SecurityCache _cache;
// co2::Platform _platform;
public:
MDConsumerImpl(
@@ -584,7 +600,8 @@ class MDConsumerImpl {
SecurityCache &cache,
api::ISyntheticSecurityService *ss,
MBOReader *mbo,
bool shared)
bool shared,
co2::Platform platform)
: _workmutex(), _instances(), _sci(), _mbo(mbo), _deferred_handles(),
_callbacks(), _cache(cache) {
std::ostringstream msg;
@@ -595,14 +612,15 @@ class MDConsumerImpl {
msg << symbol.symbol() << "-" << SecurityType_Name(symbol.type())
<< " " << Venue_Name(symbol.exchange());
}
LOG(md, Info, "Created MDConsumerImpl for securities: {}", msg.str());
LOG(md, Info, "Created MDConsumerImpl for securities: {} on {}",
msg.str(), co2::Platform_Name(platform));
for (auto &s : symbols) {
if (s.type() == SECURITYTYPE_FUTURE ||
s.type() == SECURITYTYPE_OPTION ||
s.type() == api::SECURITYTYPE_SPOT) {
_instances.push_back(make_shared<MDConsumerInst>(
s.symbol(), s.type(), s.exchange(), shared));
s.symbol(), s.type(), s.exchange(), shared, platform));
}
// TODO, add options here
}
@@ -614,12 +632,13 @@ class MDConsumerImpl {
// ISyntheticSecurityService?
if (has_synthetic_instance == false) {
_instances.push_back(make_shared<MDConsumerInst>(
SYNTHETIC_SYMBOL, s.type(), api::VENUE_BTS, shared));
SYNTHETIC_SYMBOL, s.type(), api::VENUE_BTS, shared,
platform));
}
has_synthetic_instance = true;
} else {
_instances.push_back(make_shared<MDConsumerInst>(
s.symbol(), s.type(), s.exchange(), shared));
s.symbol(), s.type(), s.exchange(), shared, platform));
}
}
}
@@ -1003,7 +1022,7 @@ class MDConsumerImpl {
// MDConsumer public interface, forwards to pimple
///////////////////////////////////////////////////////////////////////////////////////////////////
void log_cons(const std::set<UserSymbol> *symbols) {
void log_cons(const std::set<UserSymbol> *symbols, co2::Platform platform) {
std::string log_str = "";
if (symbols != nullptr) {
for (const auto &s : *symbols) {
@@ -1011,7 +1030,8 @@ void log_cons(const std::set<UserSymbol> *symbols) {
Venue_Name(s.exchange()) + ",";
}
}
LOG(md, Info, "MDConsumer::MDConsumer symbols: {}", log_str);
LOG(md, Info, "MDConsumer::MDConsumer symbols: {} platform: {} ", log_str,
static_cast<int>(platform));
}
MDConsumer::MDConsumer(
@@ -1019,16 +1039,17 @@ MDConsumer::MDConsumer(
api::ISyntheticSecurityService *ss,
MBOReader *mbo,
const std::set<UserSymbol> *symbols,
bool shared)
bool shared,
co2::Platform platform)
: _pimpl() {
log_cons(symbols);
log_cons(symbols, platform);
set<UserSymbol> all_symbols;
if (symbols == nullptr || symbols->empty()) {
all_symbols = SymbolLoader::load();
symbols = &all_symbols;
}
_pimpl.reset(new MDConsumerImpl(*symbols, cache, ss, mbo, shared));
_pimpl.reset(new MDConsumerImpl(*symbols, cache, ss, mbo, shared, platform));
}
MDConsumer::~MDConsumer() {}
@@ -1093,6 +1114,7 @@ const std::string &MDConsumerChannel::symbol() const {
}
SecurityType MDConsumerChannel::type() const { return _pimpl->type(); }
Venue MDConsumerChannel::exchange() const { return _pimpl->exchange(); }
Platform MDConsumerChannel::platform() const { return _pimpl->platform(); }
bool MDConsumerChannel::is_heart_beating() const {
return _pimpl->is_heart_beating();
+3 -1
View File
@@ -36,7 +36,8 @@ class MDConsumer {
api::ISyntheticSecurityService *ss,
MBOReader *mbo = nullptr,
const std::set<UserSymbol> *sym = nullptr,
bool shared = true);
bool shared = true,
co2::Platform platform = co2::PLATFORM_UNSET);
~MDConsumer();
bool all_heartbeats_ok() const;
@@ -88,6 +89,7 @@ class BTS_EXPORT MDConsumerChannel {
const std::string &symbol() const;
SecurityType type() const;
Venue exchange() const;
Platform platform() const;
bool is_initialized() const;
utils::TimeStamp create_time() const;
+15 -10
View File
@@ -24,15 +24,16 @@ class MDProducerImpl {
const string &symbol,
SecurityType type,
Venue exchange,
Platform platform,
MDProducerFlags flags)
: _symbol(symbol), _type(type), _exchange(exchange),
_hot(
ShmSegmentName::hot_name(symbol, type, exchange),
ShmSegmentName::hot_name(symbol, type, exchange, platform),
config::get<PublisherOptions>().hot_initial(),
config::get<PublisherOptions>().hot_growth(),
flags & MDProducerFlags::Shared),
_cold(
ShmSegmentName::cold_name(symbol, type, exchange),
ShmSegmentName::cold_name(symbol, type, exchange, platform),
config::get<PublisherOptions>().cold_initial(),
config::get<PublisherOptions>().cold_growth(),
flags & MDProducerFlags::Shared),
@@ -48,6 +49,7 @@ class MDProducerImpl {
// We should monitor and adjust based on production use for optimal
// performance
_cold.header().has_depth = enable_depth(symbol, type);
_cold.header().platform = platform;
// static_assert(sizeof(HotEntry) <= 32,"HotEntry should be less than 32
// bytes");
@@ -128,19 +130,22 @@ class MDProducerImpl {
// MDProducer
//
MDProducer::MDProducer(MDProducerFlags flags)
: _mutex(new std::mutex()), _producers(), _id_recorder(), _flags(flags),
MDProducer::MDProducer(Platform platform, MDProducerFlags flags)
: _mutex(new std::mutex()), _producers(), _id_recorder(),
_platform(platform), _flags(flags),
_smapper(flags & MDProducerFlags::SecurityMapper) {}
#ifndef _WIN32
MDProducer::MDProducer(SecurityIDRecorder recorder, MDProducerFlags flags)
MDProducer::MDProducer(
Platform platform, SecurityIDRecorder recorder, MDProducerFlags flags)
: _mutex(new std::mutex()), _producers(),
_id_recorder(std::make_unique<SecurityIDRecorder>(recorder)),
_flags(flags), _smapper(flags & MDProducerFlags::SecurityMapper) {}
_platform(platform), _flags(flags),
_smapper(flags & MDProducerFlags::SecurityMapper) {}
#endif
MDProducer::MDProducer(MDProducer &&other)
: _mutex(std::move(other._mutex)), _producers(std::move(other._producers)),
_flags(other._flags), _smapper(_flags & MDProducerFlags::SecurityMapper) {
}
_platform(other._platform), _flags(other._flags),
_smapper(_flags & MDProducerFlags::SecurityMapper) {}
MDProducer::~MDProducer() {}
@@ -164,9 +169,9 @@ MDProducerImpl &MDProducer::get_producer(
// for each one so we put them all in one
newp.reset(new MDProducerImpl(
SYNTHETIC_SYMBOL, api::SECURITYTYPE_SYNTHETIC, Venue::VENUE_BTS,
_flags));
_platform, _flags));
} else {
newp.reset(new MDProducerImpl(symbol, type, exchange, _flags));
newp.reset(new MDProducerImpl(symbol, type, exchange, _platform, _flags));
}
_producers.push_back(std::move(newp));
return *_producers.back();
+4 -1
View File
@@ -42,8 +42,10 @@ constexpr MDProducerFlags DefaultMDProducerFlags =
// the same for he other venue))
class BTS_EXPORT MDProducer {
public:
MDProducer(MDProducerFlags flags = DefaultMDProducerFlags);
MDProducer(
Platform platform, MDProducerFlags flags = DefaultMDProducerFlags);
MDProducer(
Platform platform,
SecurityIDRecorder recorder,
MDProducerFlags flags = DefaultMDProducerFlags);
MDProducer(MDProducer &&other);
@@ -70,6 +72,7 @@ class BTS_EXPORT MDProducer {
std::unique_ptr<std::mutex> _mutex;
std::vector<std::unique_ptr<MDProducerImpl>> _producers;
std::unique_ptr<SecurityIDRecorder> _id_recorder;
Platform _platform;
MDProducerFlags _flags;
SecurityMapper _smapper;
};
+15 -10
View File
@@ -1,4 +1,5 @@
#if !defined(_WIN32) && defined(SPARK_ENABLE_CCAPI)
#include "server/sparkd/md/platforms/mdp3_onixs/MDP3Main.h"
#include "MarketDataGateway.h"
#include "lib/api/md/EmumbaDeribitMarketDataConfig.pb.h"
#include "lib/api/md/MarketDataConfig.pb.h"
@@ -39,7 +40,6 @@
#include "lib/utils/bts_fatal.h"
#include "lib/utils/thread_manager.h"
#include "server/sparkd/md/platforms/emumba/EmumbaMain.h"
#include "server/sparkd/md/platforms/mdp3_onixs/MDP3Main.h"
#include <bits/stdint-uintn.h>
#include <boost/filesystem.hpp>
@@ -97,6 +97,7 @@ struct MDReplayData {
MDReplayData(const bts::api::md::MarketDataReplayConfig &replay_conf)
: replay_conf(replay_conf), venue(replay_conf.venue()), sec_index(),
md_producer(bts::co2::PLATFORM_SIM),
md_replayer(replay_conf.venue(), sec_index, md_producer) {}
};
@@ -106,7 +107,8 @@ struct CCMDSourceData {
bts::md::GeneralMarketDataListener listener;
CCMDSourceData(bts::api::Venue venue)
: sec_index(), listener(sec_index, md_producer, venue) {}
: sec_index(), md_producer(bts::co2::PLATFORM_SIM),
listener(sec_index, md_producer, venue) {}
};
struct MarketDataGateway::PImpl {
@@ -140,7 +142,7 @@ void MarketDataGateway::run() {
bts_assert(me_->md_config.emumba_deribit().size() == 1);
_start_emumba_deribit(me_->md_config.emumba_deribit().at(0));
}
if (!me_->md_config.glbx().empty()) {
if(!me_->md_config.glbx().empty()){
bts_assert(me_->md_config.glbx().size() == 1);
_start_onix_glbx(me_->md_config.glbx().at(0));
}
@@ -242,17 +244,18 @@ void MarketDataGateway::unsubscribe(MarketDataListenerID id) {
/// glbx //////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////
static void run_onixs() {
LOG(md, Info, "MarketDataGateway::run_onixs t_id:{} t_name",
os::get_thread_id(), os::get_thread_name());
LOG(md, Info, "MarketDataGateway::run_onixs t_id:{} t_name", os::get_thread_id(),
os::get_thread_name());
bts::md::MBOWriter *mbo = nullptr;
bts::md::TradeSummarySignalDispatcher *tssd = nullptr;
std::vector<std::shared_ptr<bts::md::IMDPlatform>> platforms;
// get all glbx symbols
auto cme_symbols = SymbolLoader::load(
bts::domain::SecurityCriteria::create({api::domain::VENUE_GLBX}));
auto cme_symbols =
SymbolLoader::load(bts::domain::SecurityCriteria::create({api::domain::VENUE_GLBX}));
if (!cme_symbols.empty()) {
auto *main = new bts::md::MDP3Main();
main->start({cme_symbols.begin(), cme_symbols.end()}, mbo, tssd);
main->start(
{cme_symbols.begin(), cme_symbols.end()}, mbo, tssd);
platforms.emplace_back(main);
}
// deribit
@@ -268,8 +271,9 @@ static void run_onixs() {
disp.run();
}
void MarketDataGateway::_start_onix_glbx(
bts::api::md::GlbxMarketDataConfig cfg) {
void MarketDataGateway::_start_onix_glbx( bts::api::md::GlbxMarketDataConfig cfg){
LOG(md, Info,
"MarketDataGateway::_start_onix_glbx starting t_id:{} t_name:{}. "
"cfg:\n{}",
@@ -277,6 +281,7 @@ void MarketDataGateway::_start_onix_glbx(
bts::md::run_onixs();
}
// deribit ////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////
+1 -1
View File
@@ -65,7 +65,7 @@ class SimpleSecurityIndex {
auto result =
_map.insert(std::make_pair(key, std::unique_ptr<SecurityData>()));
if (!result.second) {
LOG(md, Error, "Exchange key '{}' has already been added.", key);
LOG(md, Error, "Found duplicate entry for platform id '{} - ", key);
bts_assert(result.second);
}
result.first->second = std::move(entry);
+4 -2
View File
@@ -83,10 +83,12 @@ void apply_work(SettlementRecorderImpl &me, Work work) {
}
}
SettlementRecorder::SettlementRecorder() : _self(new SettlementRecorderImpl()) {
SettlementRecorder::SettlementRecorder(Platform platform)
: _self(new SettlementRecorderImpl()) {
string db_dir =
config::get<config::SystemConfig>().data_path() + "databases/md/";
string db_path = db_dir + "/settlements.sqlite";
string db_path =
db_dir + "/settlements_" + Platform_Name(platform) + ".sqlite";
if (!boost::filesystem::exists(db_dir)) {
boost::filesystem::create_directories(db_dir);
}
+1 -1
View File
@@ -11,7 +11,7 @@ using namespace md;
struct SettlementRecorderImpl;
class SettlementRecorder {
public:
SettlementRecorder();
SettlementRecorder(Platform platform);
~SettlementRecorder();
// Put settlement update in a queue
+8 -6
View File
@@ -93,8 +93,8 @@ class HotDepthEntry {
*bid = implied[depth].bid();
*ask = implied[depth].ask();
}
bool write_direct(
uint32_t depth, const PriceSize &bid, const PriceSize &ask) {
bool
write_direct(uint32_t depth, const PriceSize &bid, const PriceSize &ask) {
bts_assert_d(depth < direct.size());
auto temp = PriceLevel(bid, ask);
if (direct[depth] != temp) {
@@ -103,8 +103,8 @@ class HotDepthEntry {
}
return false;
}
bool write_implied(
uint32_t depth, const PriceSize &bid, const PriceSize &ask) {
bool
write_implied(uint32_t depth, const PriceSize &bid, const PriceSize &ask) {
bts_assert_d(depth < implied.size());
auto temp = PriceLevel(bid, ask);
if (implied[depth] != temp) {
@@ -137,8 +137,9 @@ struct ALIGNED_(CACHE_LINE) ColdHeader {
heartbeat_ms(config::get<md::PublisherOptions>().heartbeat_ms()),
heartbeat_boot_seconds(
config::get<md::PublisherOptions>().heartbeat_boot_seconds()),
security_count(0), has_depth(false), trade_count(0),
coconut_status_count(0), exchange_status_count(0), rfq_count(0) {}
security_count(0), has_depth(false), platform(co2::PLATFORM_UNSET),
trade_count(0), coconut_status_count(0), exchange_status_count(0),
rfq_count(0) {}
uint32_t version;
size_t count;
@@ -151,6 +152,7 @@ struct ALIGNED_(CACHE_LINE) ColdHeader {
uint32_t security_count; // incremented when a security is made public
bool has_depth;
Platform platform;
std::atomic<size_t> trade_count;
std::atomic<size_t> coconut_status_count;
Loaded 100 of 123 files, more files were not shown because too many files have changed in this diff. Show more