Compare commits

..
Author SHA1 Message Date
John Colvin 1e4fdcc404 check pt: Before going to Chi (11/07/23, random scripts, etc.) 2024-01-23 10:39:29 -06:00
John Colvin ddfd34da80 check pt: Before going to Chi (11/07/23) 2024-01-23 10:39:28 -06:00
John Colvin ea322ddb5c add util: StringStreamEnumerable 2024-01-23 10:39:28 -06:00
John Colvin 477b05d06f add websocket server config, README 2024-01-23 10:39:28 -06:00
John Colvin 7413790bd4 refactor: Change WebsocketServerConfig.port to uint32
It should always be a number between 1-65535
2024-01-23 10:39:28 -06:00
John Colvin 7a8643b186 feat: Add Bts.DataStream.Websocket/Server 2024-01-23 10:39:25 -06:00
John Colvin e4359c43ac feat: Add JsonRpc types for websocket paradigm 2024-01-23 10:35:47 -06:00
John Colvin 71eab08be1 refactor: Add WebsocketGatewayObservable
An IObservable wrapper for IWebsocketConnectionGateway
2024-01-23 10:35:47 -06:00
John Colvin 09a57631c3 refactor: Remove nullability constraint in GetValueOrDefault 2024-01-23 10:35:47 -06:00
John Colvin 7956754e73 refactor: Use events to avoid circular deps for ws sync/gateway 2024-01-23 10:35:47 -06:00
John Colvin d6abd90bfc refactor: Let caller specify nullability of T in json parse fn 2024-01-23 10:35:47 -06:00
John Colvin ce7ec0c68d refactor: Get rid of redundant UniversalNumericComparer 2024-01-23 10:35:47 -06:00
John Colvin 07fb055bf2 refactor: Reorganize NumericComparer, support all types 2024-01-23 10:35:47 -06:00
John Colvin cae9268f94 docker: Update default compiler versions
Via update-alternatives
2024-01-19 16:00:57 -05:00
John Colvin 2d8f942e62 cleanup: Remove leftover legacy files 2024-01-19 15:25:12 -05:00
John Colvin 23c5a4c46b build: Upgrade c# to net8.0 (spark_websocket, Spark.Admin.WebApp) 2024-01-19 15:24:16 -05:00
John Colvin adc043a1ec docker: Upgrade dotnet 7 -> 8 2024-01-19 15:22:08 -05:00
John Colvin 8bcae22020 Check pt before trip to NC 2024-01-18 2024-01-18 10:26:04 -05:00
John Colvin 5882332129 build: Upgrade c# to net8.0 2024-01-18 10:25:50 -05:00
John Colvin 22f1e61f82 Add bones for Bts.DataStreamWebApp.Client 2024-01-18 00:33:23 -05:00
John Colvin 8bb7262d81 Add bones for Bts.DataStreamWebApp.Server 2024-01-18 00:32:30 -05:00
John Colvin c39a270115 fix: Support all exchs in QueryServer::get_names 2024-01-17 11:15:01 -05:00
Taha Afzal 6e3f9f99c0 Merge branch 'development' of bitbucket.org:bluetradesys/spark into development 2024-01-17 10:08:17 -05:00
John Colvin b53d9d7d4c refactor: Remove 'Platform' (md source) config
Good riddance. Improper 'PlatformMap' config was often the reason market
data mysteriously wouldn't show up in the client. Especially with new
exchanges. We should keep all md source/connection configuration in one
place (one proto message), to avoid any confusion.
2024-01-17 01:42:57 -05:00
Taha Afzal cc3e289345 KeyValueToRelationalStream added 2024-01-16 19:08:40 -05:00
John Colvin 60a8b4f8c7 Gitignore all .tmp dirs (at any level) 2024-01-15 13:34:35 -05:00
John Colvin c5e0a93869 Gitignore all .venv* dirs (at any level) 2024-01-15 13:33:05 -05:00
John Colvin 3a3216421b Remove unnecessary file temp.cpp 2024-01-15 13:33:05 -05:00
Taha Afzal 8cbbbc6da3 Merging 2024-01-13 23:45:51 -05:00
123 changed files with 6010 additions and 1710 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/net7.0",
"DOTNET_WINDOWS_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net7.0-windows"
"DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net8.0",
"DOTNET_WINDOWS_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net8.0-windows"
}
},
@@ -197,7 +197,7 @@
"CMAKE_CXX_FLAGS_RELEASE": "-fPIC -Wall -pedantic -Wextra -Werror -Wno-missing-braces -Wno-ignored-qualifiers --std=c++20 -fdiagnostics-color -fno-omit-frame-pointer -DNDEBUG -O3",
"CMAKE_INSTALL_PREFIX": "${sourceDir}/install/${presetName}",
"CONFIG_DEFAULTS_PATH": "config/profiles/default/*.conf",
"DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net7.0"
"DOTNET_BINARY_DIR": "${sourceDir}/build/$<CONFIG>/bin/net8.0"
}
},
+12 -8
View File
@@ -37,8 +37,12 @@ RUN apt-get update && apt-get install -y \
&& python3 -m pip install --upgrade pip
# dotnet
RUN apt-get update && apt-get install -y \
dotnet7
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
# ssh
RUN apt-get update && apt-get install -y \
@@ -102,12 +106,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++-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
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
# Add compiler symlinks for ccache
RUN \
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0-windows</TargetFramework>
<TargetFramework>net8.0-windows</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<UseWindowsForms>true</UseWindowsForms>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -0,0 +1,20 @@
module.exports = {
root: true,
env: { browser: true, es2020: true },
extends: [
'eslint:recommended',
'plugin:react/recommended',
'plugin:react/jsx-runtime',
'plugin:react-hooks/recommended',
],
ignorePatterns: ['dist', '.eslintrc.cjs'],
parserOptions: { ecmaVersion: 'latest', sourceType: 'module' },
settings: { react: { version: '18.2' } },
plugins: ['react-refresh'],
rules: {
'react-refresh/only-export-components': [
'warn',
{ allowConstantExport: true },
],
},
}
@@ -0,0 +1,24 @@
# Logs
logs
*.log
npm-debug.log*
yarn-debug.log*
yarn-error.log*
pnpm-debug.log*
lerna-debug.log*
node_modules
dist
dist-ssr
*.local
# Editor directories and files
.vscode/*
!.vscode/extensions.json
.idea
.DS_Store
*.suo
*.ntvs*
*.njsproj
*.sln
*.sw?
@@ -0,0 +1,11 @@
<Project Sdk="Microsoft.VisualStudio.JavaScript.Sdk/0.5.128-alpha">
<PropertyGroup>
<StartupCommand>npm run dev</StartupCommand>
<JavaScriptTestRoot>src\</JavaScriptTestRoot>
<JavaScriptTestFramework>Jest</JavaScriptTestFramework>
<!-- Allows the build (or compile) script located on package.json to run on Build -->
<ShouldRunBuildScript>false</ShouldRunBuildScript>
<!-- Folder where production build objects will be placed -->
<PublishAssetsDirectory>$(MSBuildProjectDirectory)\dist</PublishAssetsDirectory>
</PropertyGroup>
</Project>
@@ -0,0 +1,8 @@
# React + Vite
This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
Currently, two official plugins are available:
- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react/README.md) uses [Babel](https://babeljs.io/) for Fast Refresh
- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react-swc) uses [SWC](https://swc.rs/) for Fast Refresh
@@ -0,0 +1,13 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<link rel="icon" type="image/svg+xml" href="/vite.svg" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Vite + React</title>
</head>
<body>
<div id="root"></div>
<script type="module" src="/src/main.jsx"></script>
</body>
</html>
@@ -0,0 +1,10 @@
<?xml version="1.0" encoding="utf-8"?>
<configuration>
<packageSources>
<clear />
<add key="nuget.org" value="https://api.nuget.org/v3/index.json" />
</packageSources>
<disabledPackageSources>
<clear />
</disabledPackageSources>
</configuration>
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,26 @@
{
"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"
}
}
@@ -0,0 +1 @@
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" aria-hidden="true" role="img" class="iconify iconify--logos" width="31.88" height="32" preserveAspectRatio="xMidYMid meet" viewBox="0 0 256 257"><defs><linearGradient id="IconifyId1813088fe1fbc01fb466" x1="-.828%" x2="57.636%" y1="7.652%" y2="78.411%"><stop offset="0%" stop-color="#41D1FF"></stop><stop offset="100%" stop-color="#BD34FE"></stop></linearGradient><linearGradient id="IconifyId1813088fe1fbc01fb467" x1="43.376%" x2="50.316%" y1="2.242%" y2="89.03%"><stop offset="0%" stop-color="#FFEA83"></stop><stop offset="8.333%" stop-color="#FFDD35"></stop><stop offset="100%" stop-color="#FFA800"></stop></linearGradient></defs><path fill="url(#IconifyId1813088fe1fbc01fb466)" d="M255.153 37.938L134.897 252.976c-2.483 4.44-8.862 4.466-11.382.048L.875 37.958c-2.746-4.814 1.371-10.646 6.827-9.67l120.385 21.517a6.537 6.537 0 0 0 2.322-.004l117.867-21.483c5.438-.991 9.574 4.796 6.877 9.62Z"></path><path fill="url(#IconifyId1813088fe1fbc01fb467)" d="M185.432.063L96.44 17.501a3.268 3.268 0 0 0-2.634 3.014l-5.474 92.456a3.268 3.268 0 0 0 3.997 3.378l24.777-5.718c2.318-.535 4.413 1.507 3.936 3.838l-7.361 36.047c-.495 2.426 1.782 4.5 4.151 3.78l15.304-4.649c2.372-.72 4.652 1.36 4.15 3.788l-11.698 56.621c-.732 3.542 3.979 5.473 5.943 2.437l1.313-2.028l72.516-144.72c1.215-2.423-.88-5.186-3.54-4.672l-25.505 4.922c-2.396.462-4.435-1.77-3.759-4.114l16.646-57.705c.677-2.35-1.37-4.583-3.769-4.113Z"></path></svg>

After

Width:  |  Height:  |  Size: 1.5 KiB

@@ -0,0 +1,23 @@
#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);
}
@@ -0,0 +1,155 @@
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>
);
}
}
@@ -0,0 +1 @@
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" aria-hidden="true" role="img" class="iconify iconify--logos" width="35.93" height="32" preserveAspectRatio="xMidYMid meet" viewBox="0 0 256 228"><path fill="#00D8FF" d="M210.483 73.824a171.49 171.49 0 0 0-8.24-2.597c.465-1.9.893-3.777 1.273-5.621c6.238-30.281 2.16-54.676-11.769-62.708c-13.355-7.7-35.196.329-57.254 19.526a171.23 171.23 0 0 0-6.375 5.848a155.866 155.866 0 0 0-4.241-3.917C100.759 3.829 77.587-4.822 63.673 3.233C50.33 10.957 46.379 33.89 51.995 62.588a170.974 170.974 0 0 0 1.892 8.48c-3.28.932-6.445 1.924-9.474 2.98C17.309 83.498 0 98.307 0 113.668c0 15.865 18.582 31.778 46.812 41.427a145.52 145.52 0 0 0 6.921 2.165a167.467 167.467 0 0 0-2.01 9.138c-5.354 28.2-1.173 50.591 12.134 58.266c13.744 7.926 36.812-.22 59.273-19.855a145.567 145.567 0 0 0 5.342-4.923a168.064 168.064 0 0 0 6.92 6.314c21.758 18.722 43.246 26.282 56.54 18.586c13.731-7.949 18.194-32.003 12.4-61.268a145.016 145.016 0 0 0-1.535-6.842c1.62-.48 3.21-.974 4.76-1.488c29.348-9.723 48.443-25.443 48.443-41.52c0-15.417-17.868-30.326-45.517-39.844Zm-6.365 70.984c-1.4.463-2.836.91-4.3 1.345c-3.24-10.257-7.612-21.163-12.963-32.432c5.106-11 9.31-21.767 12.459-31.957c2.619.758 5.16 1.557 7.61 2.4c23.69 8.156 38.14 20.213 38.14 29.504c0 9.896-15.606 22.743-40.946 31.14Zm-10.514 20.834c2.562 12.94 2.927 24.64 1.23 33.787c-1.524 8.219-4.59 13.698-8.382 15.893c-8.067 4.67-25.32-1.4-43.927-17.412a156.726 156.726 0 0 1-6.437-5.87c7.214-7.889 14.423-17.06 21.459-27.246c12.376-1.098 24.068-2.894 34.671-5.345a134.17 134.17 0 0 1 1.386 6.193ZM87.276 214.515c-7.882 2.783-14.16 2.863-17.955.675c-8.075-4.657-11.432-22.636-6.853-46.752a156.923 156.923 0 0 1 1.869-8.499c10.486 2.32 22.093 3.988 34.498 4.994c7.084 9.967 14.501 19.128 21.976 27.15a134.668 134.668 0 0 1-4.877 4.492c-9.933 8.682-19.886 14.842-28.658 17.94ZM50.35 144.747c-12.483-4.267-22.792-9.812-29.858-15.863c-6.35-5.437-9.555-10.836-9.555-15.216c0-9.322 13.897-21.212 37.076-29.293c2.813-.98 5.757-1.905 8.812-2.773c3.204 10.42 7.406 21.315 12.477 32.332c-5.137 11.18-9.399 22.249-12.634 32.792a134.718 134.718 0 0 1-6.318-1.979Zm12.378-84.26c-4.811-24.587-1.616-43.134 6.425-47.789c8.564-4.958 27.502 2.111 47.463 19.835a144.318 144.318 0 0 1 3.841 3.545c-7.438 7.987-14.787 17.08-21.808 26.988c-12.04 1.116-23.565 2.908-34.161 5.309a160.342 160.342 0 0 1-1.76-7.887Zm110.427 27.268a347.8 347.8 0 0 0-7.785-12.803c8.168 1.033 15.994 2.404 23.343 4.08c-2.206 7.072-4.956 14.465-8.193 22.045a381.151 381.151 0 0 0-7.365-13.322Zm-45.032-43.861c5.044 5.465 10.096 11.566 15.065 18.186a322.04 322.04 0 0 0-30.257-.006c4.974-6.559 10.069-12.652 15.192-18.18ZM82.802 87.83a323.167 323.167 0 0 0-7.227 13.238c-3.184-7.553-5.909-14.98-8.134-22.152c7.304-1.634 15.093-2.97 23.209-3.984a321.524 321.524 0 0 0-7.848 12.897Zm8.081 65.352c-8.385-.936-16.291-2.203-23.593-3.793c2.26-7.3 5.045-14.885 8.298-22.6a321.187 321.187 0 0 0 7.257 13.246c2.594 4.48 5.28 8.868 8.038 13.147Zm37.542 31.03c-5.184-5.592-10.354-11.779-15.403-18.433c4.902.192 9.899.29 14.978.29c5.218 0 10.376-.117 15.453-.343c-4.985 6.774-10.018 12.97-15.028 18.486Zm52.198-57.817c3.422 7.8 6.306 15.345 8.596 22.52c-7.422 1.694-15.436 3.058-23.88 4.071a382.417 382.417 0 0 0 7.859-13.026a347.403 347.403 0 0 0 7.425-13.565Zm-16.898 8.101a358.557 358.557 0 0 1-12.281 19.815a329.4 329.4 0 0 1-23.444.823c-7.967 0-15.716-.248-23.178-.732a310.202 310.202 0 0 1-12.513-19.846h.001a307.41 307.41 0 0 1-10.923-20.627a310.278 310.278 0 0 1 10.89-20.637l-.001.001a307.318 307.318 0 0 1 12.413-19.761c7.613-.576 15.42-.876 23.31-.876H128c7.926 0 15.743.303 23.354.883a329.357 329.357 0 0 1 12.335 19.695a358.489 358.489 0 0 1 11.036 20.54a329.472 329.472 0 0 1-11 20.722Zm22.56-122.124c8.572 4.944 11.906 24.881 6.52 51.026c-.344 1.668-.73 3.367-1.15 5.09c-10.622-2.452-22.155-4.275-34.23-5.408c-7.034-10.017-14.323-19.124-21.64-27.008a160.789 160.789 0 0 1 5.888-5.4c18.9-16.447 36.564-22.941 44.612-18.3ZM128 90.808c12.625 0 22.86 10.235 22.86 22.86s-10.235 22.86-22.86 22.86s-22.86-10.235-22.86-22.86s10.235-22.86 22.86-22.86Z"></path></svg>

After

Width:  |  Height:  |  Size: 4.0 KiB

@@ -0,0 +1,68 @@
:root {
font-family: Inter, system-ui, Avenir, Helvetica, Arial, sans-serif;
line-height: 1.5;
font-weight: 400;
color-scheme: light dark;
color: rgba(255, 255, 255, 0.87);
background-color: #242424;
font-synthesis: none;
text-rendering: optimizeLegibility;
-webkit-font-smoothing: antialiased;
-moz-osx-font-smoothing: grayscale;
}
a {
font-weight: 500;
color: #646cff;
text-decoration: inherit;
}
a:hover {
color: #535bf2;
}
body {
margin: 0;
display: flex;
place-items: center;
min-width: 320px;
min-height: 100vh;
}
h1 {
font-size: 3.2em;
line-height: 1.1;
}
button {
border-radius: 8px;
border: 1px solid transparent;
padding: 0.6em 1.2em;
font-size: 1em;
font-weight: 500;
font-family: inherit;
background-color: #1a1a1a;
cursor: pointer;
transition: border-color 0.25s;
}
button:hover {
border-color: #646cff;
}
button:focus,
button:focus-visible {
outline: 4px auto -webkit-focus-ring-color;
}
@media (prefers-color-scheme: light) {
:root {
color: #213547;
background-color: #ffffff;
}
a:hover {
color: #747bff;
}
button {
background-color: #f9f9f9;
}
}
@@ -0,0 +1,10 @@
import React from 'react'
import ReactDOM from 'react-dom/client'
import App from './App.jsx'
import './index.css'
ReactDOM.createRoot(document.getElementById('root')).render(
<React.StrictMode>
<App />
</React.StrictMode>,
)
@@ -0,0 +1,60 @@
import { fileURLToPath, URL } from 'node:url';
import { defineConfig } from 'vite';
import plugin from '@vitejs/plugin-react';
import fs from 'fs';
import path from 'path';
import child_process from 'child_process';
const baseFolder =
process.env.APPDATA !== undefined && process.env.APPDATA !== ''
? `${process.env.APPDATA}/ASP.NET/https`
: `${process.env.HOME}/.aspnet/https`;
const certificateArg = process.argv.map(arg => arg.match(/--name=(?<value>.+)/i)).filter(Boolean)[0];
const certificateName = certificateArg ? certificateArg.groups.value : "bts.datastreamwebapp.client";
if (!certificateName) {
console.error('Invalid certificate name. Run this script in the context of an npm/yarn script or pass --name=<<app>> explicitly.')
process.exit(-1);
}
const certFilePath = path.join(baseFolder, `${certificateName}.pem`);
const keyFilePath = path.join(baseFolder, `${certificateName}.key`);
if (!fs.existsSync(certFilePath) || !fs.existsSync(keyFilePath)) {
if (0 !== child_process.spawnSync('dotnet', [
'dev-certs',
'https',
'--export-path',
certFilePath,
'--format',
'Pem',
'--no-password',
], { stdio: 'inherit', }).status) {
throw new Error("Could not create certificate.");
}
}
// https://vitejs.dev/config/
export default defineConfig({
plugins: [plugin()],
resolve: {
alias: {
'@': fileURLToPath(new URL('./src', import.meta.url))
}
},
server: {
proxy: {
'^/weatherforecast': {
target: 'https://localhost:7128/',
secure: false
}
},
port: 5173,
https: {
key: fs.readFileSync(keyFilePath),
cert: fs.readFileSync(certFilePath),
}
}
})
@@ -0,0 +1,28 @@
<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>
@@ -0,0 +1,6 @@
@Bts.DataStreamWebApp.Server_HostAddress = http://localhost:5268
GET {{Bts.DataStreamWebApp.Server_HostAddress}}/weatherforecast/
Accept: application/json
###
@@ -0,0 +1,25 @@
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();
}
}
@@ -0,0 +1,31 @@
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();
}
}
@@ -0,0 +1,5 @@
namespace Bts.DataStreamWebApp.Server;
public class DataStreamSubscription {
public string? Query { get; set; }
}
@@ -0,0 +1,30 @@
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();
@@ -0,0 +1,43 @@
{
"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
}
}
}
@@ -0,0 +1,12 @@
namespace Bts.DataStreamWebApp.Server;
public class WeatherForecast
{
public DateOnly Date { get; set; }
public int TemperatureC { get; set; }
public int TemperatureF => 32 + (int)(TemperatureC / 0.5556);
public string? Summary { get; set; }
}
@@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
}
}
@@ -0,0 +1,9 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
},
"AllowedHosts": "*"
}
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -15,6 +15,7 @@
</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))
{
BinaryFormatter bf = new BinaryFormatter();
#pragma warning disable SYSLIB0011
BinaryFormatter bf = new BinaryFormatter();
BlueTable bt = (BlueTable)bf.Deserialize(fs);
#pragma warning restore SYSLIB0011
mySetRealSource(new BlueRowCollectionStaticStream(bt));
@@ -0,0 +1,31 @@
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);
}
}
}
@@ -0,0 +1,121 @@
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,37 +1,212 @@
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;
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);
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;
// Access the database
IMongoDatabase database = client.GetDatabase(databaseName);
private IMongoDatabase _database;
private IDictionary<string, int> _id_map = new Dictionary<string, int>();
// mongodb://192.168.1.80:27017
// Access the collection
IMongoCollection<BsonDocument> collection = database.GetCollection<BsonDocument>(collectionName);
public MongoDBSQLStream(string connection_string, string db_name, string collection_name)
: this(connection_string, db_name, collection_name, ViewStreamDatabase.instance) {
// Fetch all documents from the collection
var documents = collection.Find(Builders<BsonDocument>.Filter.Empty).ToCursor();
}
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);
}
// 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 myOnFirstSubscriber() {
MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(1));
}
// Print a separator between documents
Console.WriteLine(new string('-', 20));
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);
}
}
}
[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,8 +3,7 @@ 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.
@@ -12,11 +11,12 @@ 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,
Platform,
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,
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>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<Nullable>disable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -11,7 +11,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.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 = true;
bool run_echo_tcp_client_only = false;
static void init(string[] args) {
Func<string, string, Task> login = async (l, p) => {
var login = l;
@@ -31,6 +31,7 @@ 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;
@@ -58,14 +59,16 @@ 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\")"},
// {"bbo_ts","timeSeries(bbo())"}
};
{"mongodb_testcollection", "mongodb(\"mongodb://192.168.1.80:27017\",\"local\",\"test_collection\")"}
};
// {"bbo_ts","timeSeries(bbo())"}
foreach (var kv in streams_to_publish) {
var name = kv.Key;
var exp = kv.Value;
@@ -77,6 +80,7 @@ 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>net7.0-windows</TargetFramework>
<TargetFramework>net8.0-windows</TargetFramework>
<Nullable>enable</Nullable>
<UseWindowsForms>true</UseWindowsForms>
</PropertyGroup>
+6
View File
@@ -38,6 +38,8 @@ 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}"
@@ -131,6 +133,10 @@ 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>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0-windows</TargetFramework>
<TargetFramework>net8.0-windows</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<UseWindowsForms>true</UseWindowsForms>
@@ -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[]> getUpdates(TimeSpan timeout, int maxElements);
IList<byte[]> get_updates(TimeSpan timeout, int maxElements);
/// <summary>
/// server URL to which the connection is made
/// </summary>
string serverURL { get; }
string server_url { 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 clientURL { get; }
string client_url { get; }
/// <summary>
/// registers the callback to receive
/// connection notifications
/// </summary>
void registerCallback(ConnectionStatusChange del);
void register_connection_callback(PubSubServerConnectionStatusChange del);
/// <summary>
/// registers callbacks to reveive data
/// </summary>
/// <param name="del"></param>
void registerDataCallback(DataDelegate del);
void register_data_callback(PubSubServerDataDelegate del);
void registerSendConfirmationCallback(SendConfirmationDelegate del);
void register_send_confirmation_callback(SendConfirmationDelegate del);
}
}
@@ -8,6 +8,7 @@ using System.IO;
//using BCG.Utilities.General;
using System.Threading;
using mbts;
using bts;
namespace BCG {
/// <summary>
@@ -20,26 +21,27 @@ namespace BCG {
/// ITCPConnectionHandler.onConnection will be called on any arbitrary thread
///
/// </summary>
public class PubSubServerManagedTCPConnection : IPubSubServerConnection,
ITCPMessageHandler,
ITCPConnectionHandler {
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
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
// in onData, data is queued here, for polling
private SyncQueue<byte[]> _in_data_queue = new SyncQueue<byte[]>();
private DateTime _last_hb = DateTime.Now;
private ConnectionStatusChange _conn_callback;
private DataDelegate _data_callback;
private SendConfirmationDelegate _on_send_onfirm_callback;
private PubSubServerConnectionStatusChange _conn_callback;
private PubSubServerDataDelegate _data_callback;
private SendConfirmationDelegate _on_send_confirm_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 serverURL)
: this(new BlutoSubAddress(serverURL)) { }
public PubSubServerManagedTCPConnection(string server_url)
: this(new BlutoSubAddress(server_url)) { }
public PubSubServerManagedTCPConnection(BlutoSubAddress server_address) {
_server_address = server_address;
@@ -48,17 +50,17 @@ namespace BCG {
private void _connect_to_ps_server() {
bts.Log.Gen.info($"PubSubServerManagedTCPConnection._connect_to_ps_server server: {_server_address}");
ClientToServerManagedTCPConnector.connect_to(_server_address, this, TimeSpan.FromSeconds(1));
AsyncTCPConnector_APM.connect_to(_server_address, this);
}
#region IPubSubServerConnection Members
public void sendWithConfirmation(byte[] data, ref string conf_id) {
bts.Log.Gen.info($"PubSubServerManagedTCPConnection.sendWithConfirmation server: {_server_address} data: {data.Length} myConnection : {_connection != null}");
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}");
if (_connection != null) {
_connection.sendWithConfirmation(data, ref conf_id, (msg_id, confirmed) => {
if (_on_send_onfirm_callback != null)
_on_send_onfirm_callback(msg_id, confirmed);
if (_on_send_confirm_callback != null)
_on_send_confirm_callback(msg_id, confirmed);
});
} else {
bts.Log.Gen.warn($"PubSubServerManagedTCPConnection.sendWithConfirmation server: {_server_address} no connection, buffering send");
@@ -86,18 +88,18 @@ namespace BCG {
public void send_with_confirmation(PubSubMessage msg, ref string conf_id) {
var bytes = msg.to_bytes();
sendWithConfirmation(bytes, ref conf_id);
send_with_confirmation(bytes, ref conf_id);
}
public IList<byte[]> getUpdates(TimeSpan timeout, int maxElements) {
public IList<byte[]> get_updates(TimeSpan timeout, int maxElements) {
return _in_data_queue.dequeueMultipleElements(Convert.ToInt32(timeout.TotalMilliseconds), maxElements);
}
public string serverURL {
public string server_url {
get { return _server_address.ToString(); }
}
public string clientURL {
public string client_url {
get {
string rv = "";
lock (this) {
@@ -118,16 +120,16 @@ namespace BCG {
}
}
public void registerCallback(ConnectionStatusChange del) {
public void register_connection_callback(PubSubServerConnectionStatusChange del) {
_conn_callback = del;
}
public void registerDataCallback(DataDelegate del) {
public void register_data_callback(PubSubServerDataDelegate del) {
_data_callback = del;
}
public void registerSendConfirmationCallback(SendConfirmationDelegate del) {
_on_send_onfirm_callback = del;
public void register_send_confirmation_callback(SendConfirmationDelegate del) {
_on_send_confirm_callback = del;
}
#endregion
@@ -185,18 +187,6 @@ 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
@@ -204,8 +194,7 @@ namespace BCG {
#region ITCPConnectionHandler Members
/// <summary>
/// This function is called on the main thread (as all other functions
/// except onData)
/// will be called on ana rbitrary .Net thread
/// </summary>
/// <param name="connection"></param>
public void onConnection(ITCPConnection connection) {
@@ -246,10 +235,10 @@ namespace BCG {
_connection.register(this);
if (_conn_callback != null)
_conn_callback(_server_address.ToString(), true);
if (_on_send_onfirm_callback != null) {
if (_on_send_confirm_callback != null) {
var msg_ids = _negative_unsent_confs.dequeueAll();
foreach (var id in msg_ids)
_on_send_onfirm_callback(id, false);
_on_send_confirm_callback(id, false);
}
}
}
@@ -264,47 +253,6 @@ 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,6 +16,7 @@ namespace BCG {
// SubMetaData
// TransactionMetaData
// UnSubMetaData
//
public class MetaData : ICloneable {
private IDictionary _name_value_map = new Dictionary<string, string>();
int mySize = -1;
@@ -1,10 +1,8 @@
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>
@@ -13,20 +11,18 @@ namespace BCG
/// identification of the client and the connection on which to receive data
/// machine; ip; port; clientURL
/// </summary>
public string subscribingApp;
public readonly string subscribing_app = ClientInfo.thisClient.processString;
/// <summary>
/// tags that are to be subscribed
/// </summary>
public string[] tags;
public SubMetaData()
{ }
public SubMetaData() { }
public SubMetaData(string theConnection, params string[] theTags)
{
connection = theConnection;
tags = theTags;
subscribingApp = ClientInfo.thisClient.processString;
public SubMetaData(string the_connection, params string[] the_tags) {
connection = the_connection;
tags = the_tags;
//subscribing_app = ClientInfo.thisClient.processString;
}
}
}
@@ -1,21 +1,17 @@
using System;
namespace BCG
{
[Serializable]
public class UnSubMetaData
{
namespace BCG {
//[Serializable]
public class UnSubMetaData {
public string connection;
public string[] tags;
public string unsubscribingApp = ClientInfo.thisClient.processString;
public readonly string[] tags;
public readonly string unsubscribing_app = ClientInfo.thisClient.processString;
public UnSubMetaData()
{}
public UnSubMetaData() { }
public UnSubMetaData(string theConnection, params string[] theTags)
{
connection = theConnection;
tags = theTags;
public UnSubMetaData(string the_connection, params string[] the_tags) {
connection = the_connection;
tags = the_tags;
}
}
}
@@ -6,12 +6,18 @@ 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,7 +8,9 @@ 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() {}
@@ -21,7 +23,9 @@ 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
@@ -31,7 +35,9 @@ 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 )
@@ -42,10 +48,14 @@ 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 () { }
@@ -55,9 +65,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,7 +328,8 @@ 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,23 +5,21 @@ 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>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup>
+8 -5
View File
@@ -4,19 +4,22 @@ 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 tagType, Stream body) { this.tag = tag; this.body = body; this.tagType = tagType; }
public PubSubUpdate(string tag, string the_tag_type, Stream body) {
this.tag = tag;
this.body = body;
this.tag_type = the_tag_type;
}
public string tag;
public Stream body;
public string tagType;
public string tag_type;
public override string ToString() {
return $"PubSubUpdate tag: {tag} tag_type: {tagType} len: {body.Position}";
return $"PubSubUpdate tag: {tag} tag_type: {tag_type} len: {body.Position}";
}
};
}
@@ -5,18 +5,16 @@ using System.IO;
using System.Collections.Generic;
using mbts;
namespace BCG
{
public delegate void ConnectionStatusChange(string peerURL, bool connection);
public delegate void DataDelegate(string sender, byte[] data);
namespace BCG {
public delegate void PubSubServerConnectionStatusChange(string peerURL, bool connection);
public delegate void PubSubServerDataDelegate(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
@@ -29,14 +27,14 @@ namespace BCG
/// getUpdates. The delegate will be fired on an arbitrary thread.
/// </summary>
/// <param name="?"></param>
void registerDataCallback(DataDelegate del);
void registerDataCallback(PubSubServerDataDelegate 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(ConnectionStatusChange del);
void registerConnectionCallback(PubSubServerConnectionStatusChange 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 ConnectionStatusChange myConnCb;
private DataDelegate myDataCb;
private PubSubServerConnectionStatusChange myConnCb;
private PubSubServerDataDelegate myDataCb;
/// <summary>
/// client to connection
@@ -77,12 +77,12 @@ namespace BCG {
#region IPubsubMessageHub Members
public void registerDataCallback(DataDelegate del) {
public void registerDataCallback(PubSubServerDataDelegate del) {
// BlueEventLog.logAssert(!myActivated, "connection callbacks should be registered before the call to activate");
myDataCb = del;
}
public void registerConnectionCallback(ConnectionStatusChange del) {
public void registerConnectionCallback(PubSubServerConnectionStatusChange del) {
//BlueEventLog.logAssert(!myActivated, "data callbacks should be registered before the call to activate");
myConnCb = del;
}
@@ -4,25 +4,22 @@ 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 ConnectionStatusChange myConnectionCb;
private DataDelegate myDataCb;
private PubSubServerConnectionStatusChange myConnectionCb;
private PubSubServerDataDelegate 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());
@@ -35,19 +32,14 @@ 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();
}
@@ -60,47 +52,39 @@ 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(DataDelegate del) { myDataCb = del; }
public void registerDataCallback(PubSubServerDataDelegate del) { myDataCb = del; }
public void registerConnectionCallback(ConnectionStatusChange del) { myConnectionCb = del; }
public void registerConnectionCallback(PubSubServerConnectionStatusChange 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 DataDelegate(myOnData));
hub.registerConnectionCallback(new ConnectionStatusChange(myOnConnection));
foreach (IPubsubMessageHub hub in myHubs.Values) {
hub.registerDataCallback(new PubSubServerDataDelegate(myOnData));
hub.registerConnectionCallback(new PubSubServerConnectionStatusChange(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))
@@ -109,8 +93,7 @@ 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))
@@ -126,8 +109,7 @@ 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();
}
@@ -138,21 +120,18 @@ 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 ConnectionStatusChange(myHandleConnections));
myMessageHub.registerConnectionCallback(new PubSubServerConnectionStatusChange(myHandleConnections));
}
@@ -55,7 +55,7 @@ namespace BCG {
myTagToPubsubMap[tag].onSub(sub);
if (onSingleTagChange != null)
onSingleTagChange(new PubSubSingleTagStateEvent(tag, PubSubSingleTagStateEvent.EventType.subscription, sub.connection, sub.subscribingApp));
onSingleTagChange(new PubSubSingleTagStateEvent(tag, PubSubSingleTagStateEvent.EventType.subscription, sub.connection, sub.subscribing_app));
}
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.unsubscribingApp));
onSingleTagChange(new PubSubSingleTagStateEvent(tag, PubSubSingleTagStateEvent.EventType.unsubscription, unsub.connection, unsub.unsubscribing_app));
} 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.unsubscribingApp);
" the unsub call by app {1} is voided", tag, unsub.unsubscribing_app);
//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,8 +29,7 @@ namespace BCG
/// </summary>
List<Pair<string, string>> myDeadPubConnToApp = new List<Pair<string, string>>();
public SingleTagPubSubMap(string tag)
{
public SingleTagPubSubMap(string tag) {
myTag = tag;
}
@@ -38,8 +37,7 @@ 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; }
}
@@ -47,8 +45,7 @@ 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; }
}
@@ -57,8 +54,7 @@ namespace BCG
/// connections as key and the application
/// as value
/// </summary>
public Dictionary<string, string> subscribers
{
public Dictionary<string, string> subscribers {
get { return mySubConnToApp; }
}
@@ -66,8 +62,7 @@ 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; }
}
@@ -75,8 +70,7 @@ 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; }
}
@@ -87,18 +81,14 @@ 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.subscribingApp);
+ "it should unsubscribe before resubscribing", myTag, sub.connection, sub.subscribing_app);
//BlueEventLog.logEvent(warningMsg, BlueEventLevel.Warning);
return false;
}
else
{
mySubConnToApp.Add(sub.connection, sub.subscribingApp);
} else {
mySubConnToApp.Add(sub.connection, sub.subscribing_app);
return true;
}
}
@@ -110,17 +100,13 @@ 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.unsubscribingApp);
myTag, unsub.connection, unsub.unsubscribing_app);
//BlueEventLog.logEvent(warningMsg, BlueEventLevel.Warning);
return false;
}
@@ -133,25 +119,20 @@ 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;
}
@@ -162,14 +143,10 @@ 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,92 +6,79 @@ 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
/// pubsub server. Once again, theres no sharing of conenctions
/// Deals with the subscriptions (and getting data) from one sole
/// pubsub server. 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> getUpdates(int timeout);
IList<PubSubUpdate> get_updates(int timeout);
event PubSubConnectionChange onPubSubConnectionChange;
void subscribe(params string[] tags);
void unsubscribe(params string[] tags);
}
internal class SingleServerSubscriberCore : IDisposable, ISingleServerSubscriberCore {
private HighPerformanceTimer myTimer = new HighPerformanceTimer();
private PubSubServerConnectionFactory myConnectionFactory;
private IPubSubServerConnection myConnection;
private List<string> myTags = new List<string>();
private readonly IPubSubServerConnection _connection;
private HashSet<string> _tags = new HashSet<string>();
public event PubSubConnectionChange onPubSubConnectionChange;
public SingleServerSubscriberCore(string serverURL, PubSubServerConnectionFactory connectionFactory) {
myConnectionFactory = connectionFactory;
myConnection = myConnectionFactory.getConnection(serverURL);
myConnection.registerCallback(new ConnectionStatusChange(myHandleServerReboot));
public SingleServerSubscriberCore(string server_url, PubSubServerConnectionFactory connection_factory) {
_connection = connection_factory.getConnection(server_url);
_connection.register_connection_callback(new PubSubServerConnectionStatusChange(_handle_server_reboot));
}
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);
// 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 retVal;
return rv;
}
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 (myTags.Contains(tag)) {
string errMsg = String.Format("tag {0} was already subscribed from server {1}",
tag, myConnection.serverURL);
// BlueEventLog.logEvent(errMsg, BlueEventLevel.Warning);
if (_tags.Contains(tag)) {
bts.Log.Gen.warn($"SingleServerSubscriberCore.subscribe tag: {tag} has already " +
$"been subscribed to server: {_connection.server_url}");
} else {
myTags.Add(tag);
_tags.Add(tag);
this_call_tags.Add(tag);
}
}
mySubscribe(tags);
_subscribe(this_call_tags.ToArray());
}
public void unsubscribe(params string[] tags) {
List<string> unsubTags = new List<string>();
List<string> unsub_tags = new List<string>();
foreach (string tag in tags) {
if (myTags.Contains(tag)) {
myTags.Remove(tag);
unsubTags.Add(tag);
if (_tags.Contains(tag)) {
_tags.Remove(tag);
unsub_tags.Add(tag);
} else {
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);
bts.Log.Gen.warn($"tag {tag} is being unsubscribed from server {_connection.server_url}, eventhough " +
"it was never subscribed");
}
}
UnSubMetaData unsub = new UnSubMetaData(myConnection.clientURL, unsubTags.ToArray());
UnSubMetaData unsub = new UnSubMetaData(_connection.client_url, unsub_tags.ToArray());
Stream str = new MemoryStream();
new MetaData(unsub).save(str);
str.Position = 0;
// this is where data is actually sent out
myConnection.send(str);
_connection.send(str);
}
#region IDisposable Members
@@ -99,44 +86,40 @@ namespace BCG {
public void Dispose() {
///remove the subscriber
//myConnection.send(new PubSubMessage(new MetaData(new RemoveSubscriberMetaData(myConnection.clientURL))).toStream());
myConnection.Dispose();
_connection.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.
myTags.Clear();
_tags.Clear();
}
#endregion
private void mySubscribe(params string[] tags) {
SubMetaData sub = new SubMetaData(myConnection.clientURL, tags);
// 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);
Stream str = new MemoryStream();
new MetaData(sub).save(str);
str.Position = 0;
myConnection.send(str);
_connection.send(str);
}
private void myHandleServerReboot(string serverURL, bool connection) {
private void _handle_server_reboot(string server_url, bool connection) {
bts.Log.Gen.info($"SingleServerSubscriberCore._handle_server_reboot connection: {connection} server_url: {server_url}");
if (connection) {
//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());
_subscribe(_tags.ToArray());
} else {
// BlueEventLog.logEvent("Connection to " + serverURL + " broke....attempting to reconnect...",
// PS3LogLevels.disconnection);
bts.Log.Gen.warn($"SingleServerSubscriberCore._handle_server_reboot connection to {server_url} broke....attempting to reconnect...");
}
//let anyone watching you know that your connection has changed.
if (onPubSubConnectionChange != null)
onPubSubConnectionChange(connection, serverURL);
onPubSubConnectionChange(connection, server_url);
}
internal static void processMessage(byte[] data, List<PubSubUpdate> updates) {
bts.Log.Gen.info($"SingleServerSubscriberCore.processMessage data_len: {data.Length}");
// 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}");
if (data != null && data.Length > 0) {
MemoryStream str = new MemoryStream(data);
MetaData md = new MetaData(str);
@@ -156,9 +139,8 @@ namespace BCG {
if (p.isCompressed) {
long compressedSize = bodyStream.Length - bodyStream.Position;
if (compressedSize == 0) {
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);
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);
return;
}
realBody = StreamDecompressor.decompressStream(bodyStream);
@@ -174,8 +156,9 @@ namespace BCG {
bodyStream.Read(trConstituent, 0, trConstituent.Length);
updates.Add(new PubSubUpdate(pub.tag, pub.type, new MemoryStream(trConstituent)));
}
} else { }
//BlueEventLog.logAssert(false, "Unknown message received: " + md.ToString() + " , only Pub and TransactionMetaData are legal");
} else {
bts.Log.assert_d(false, $"Unknown message received: {md} 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.serverURL, ClientInfo.thisClient.processString);
pub_md = new PubMetaData(tagType, tag, (int)contentToPublish.Length, (int)content.Length, serverConn.server_url, 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.serverURL, ClientInfo.thisClient.processString);
pub_md = new PubMetaData(tagType, tag, (int)content.Length, serverConn.server_url, 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.serverURL);
Console.WriteLine("StaticPublisherCore.mySend serverConn.sendWithConfirmation tag: " + tag + " server: " + serverConn.server_url);
// 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.registerCallback(new ConnectionStatusChange(myHandleServerReboot));
sender.registerSendConfirmationCallback(new SendConfirmationDelegate(myOnSendConfirm));
sender.register_connection_callback(new PubSubServerConnectionStatusChange(myHandleServerReboot));
sender.register_send_confirmation_callback(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.serverURL == serverURL) {
if (p.Value.server_url == 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.getUpdates(timeShare));
retVal.AddRange(subCore.get_updates(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 = true;
bool run_echo_tcp_server_only = false;
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>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,26 +1,20 @@
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;
@@ -28,47 +22,36 @@ 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));
@@ -79,32 +62,25 @@ 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_covalues<double>(rows, _output_col);
IEnumerable<double> vals = BlueDataUtility.get_colvalues<double>(rows, _output_col);
OHLC ohlc = compute_ohlc(vals);
MutableBlueRow nr = new MutableBlueRow(_out_schema, reprow);
nr.setValue("id", kv.Key);
@@ -118,13 +94,11 @@ 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();
@@ -133,28 +107,22 @@ 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>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<Version>1.0.0</Version>
<AssemblyVersion>1.0.0</AssemblyVersion>
</PropertyGroup>
@@ -2,7 +2,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFrameworks>net7.0</TargetFrameworks>
<TargetFrameworks>net8.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>net7.0-windows</TargetFramework>
<TargetFramework>net8.0-windows</TargetFramework>
<UseWPF>true</UseWPF>
<Authors>Blue Trading Systems</Authors>
<Company>Blue Trading Systems</Company>
@@ -320,7 +320,6 @@
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>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<Authors>Blue Trading Systems</Authors>
<Company>Blue Trading Systems</Company>
<Description />
@@ -2,7 +2,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable>
<Platforms>x64</Platforms>
<PlatformTarget>x64</PlatformTarget>
@@ -215,9 +215,6 @@ 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>net7.0</TargetFramework>
<TargetFramework>net8.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>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<OutputType>Exe</OutputType>
<GeneratePackageOnBuild Condition="$(Configuration) == 'Release' or $(Configuration) == 'Release'">true</GeneratePackageOnBuild>
<PackageOutputPath>$(OutputPath)\net7.0</PackageOutputPath>
<PackageOutputPath>$(OutputPath)\net8.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)..\net7.0\libspark.dll" />
<BuildOutputInPackage Include="$(OutputPath)..\net8.0\libspark.dll" />
</ItemGroup>
</Target>
<!-- deployed sdk build net7.0 -->
<ItemGroup Condition=" '$(SparkRoot)' == '' and '$(TargetFramework)' == 'net7.0'">
<!-- deployed sdk build net8.0 -->
<ItemGroup Condition=" '$(SparkRoot)' == '' and '$(TargetFramework)' == 'net8.0'">
<Reference Include="spark_api">
<HintPath>..\..\lib\net7.0\spark_api.dll</HintPath>
<HintPath>..\..\lib\net8.0\spark_api.dll</HintPath>
</Reference>
</ItemGroup>
<Target Name="copy_lib_deployed" AfterTargets="PostBuildEvent" Condition=" '$(SparkRoot)' == '' ">
<Copy SourceFiles="..\..\lib\net7.0\libspark.dll" DestinationFolder="$(OutputPath)\" SkipUnchangedFiles="true" />
<Copy SourceFiles="..\..\lib\net8.0\libspark.dll" DestinationFolder="$(OutputPath)\" SkipUnchangedFiles="true" />
</Target>
</Project>
-1
View File
@@ -155,7 +155,6 @@ 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>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<Platforms>x64</Platforms>
<PlatformTarget>x64</PlatformTarget>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable>
<Platforms>x64</Platforms>
<PlatformTarget>x64</PlatformTarget>
@@ -7,13 +7,6 @@ HistoricMarketDataConfig {
}
}
PlatformMap {
value {
glbx_platform: "PLATFORM_MDP3ONIXS"
ice_platform: "PLATFORM_ICEONIXS"
}
}
PublisherOptions {
value {
network_interface: "0.0.0.0"
@@ -209,14 +202,6 @@ HistoricMarketDataConfig {
}
}
PlatformMap {
rule: "DeploymentType.type == 'prod'"
value {
glbx_platform: "PLATFORM_MDP3ONIXS"
ice_platform: "PLATFORM_ICEONIXS"
}
}
MarketDataConfig {
value {
emumba_deribit {
@@ -237,4 +222,4 @@ MarketDataConfig {
value{
onixs_license: "{default}"
}
}
}
-7
View File
@@ -12,13 +12,6 @@ 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;
+45 -26
View File
@@ -150,34 +150,53 @@ bts::api::SecurityType DomainSecurityUtils::get_type(
}
std::string DomainSecurityUtils::get_exchange_id(const bts::api::Security &s) {
switch (s.ExchangeData_case()) {
case (bts::api::Security::ExchangeDataCase::EXCHANGEDATA_NOT_SET):
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:
bts_assert(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();
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;
}
bts_assert(false);
return "";
return true;
}
} // namespace bts::domain
+2
View File
@@ -27,6 +27,8 @@ 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
+18 -40
View File
@@ -78,33 +78,20 @@ bool is_heart_beat_ok(const ColdHeader &);
class MDConsumerInst {
public:
MDConsumerInst(
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),
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),
_queues() {
LOG(md, Diag,
"MDConsumerInst::MDConsumerInst symbol: {} venue: {} type: {} "
"platform: {}",
symbol, static_cast<int>(exchange), static_cast<int>(type),
static_cast<int>(platform));
"MDConsumerInst::MDConsumerInst symbol: {} venue: {} type: {}",
symbol, static_cast<int>(exchange), static_cast<int>(type));
if (enable_depth(symbol, type)) {
_hot_depth.reset(new utils::SharedListReader<HotHeader, HotDepthEntry>(
ShmSegmentName::hot_name(symbol, type, exchange, _platform),
shared));
ShmSegmentName::hot_name(symbol, type, exchange), shared));
} else {
_hot.reset(new utils::SharedListReader<HotHeader, HotEntry>(
ShmSegmentName::hot_name(symbol, type, exchange, _platform),
shared));
ShmSegmentName::hot_name(symbol, type, exchange), shared));
}
}
~MDConsumerInst() {
@@ -265,7 +252,6 @@ 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
@@ -341,7 +327,6 @@ 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;
@@ -592,7 +577,6 @@ class MDConsumerImpl {
map<security_key, shared_ptr<MDConsumerHandle>> _deferred_handles;
vector<SecurityCallback> _callbacks;
mutable SecurityCache _cache;
// co2::Platform _platform;
public:
MDConsumerImpl(
@@ -600,8 +584,7 @@ class MDConsumerImpl {
SecurityCache &cache,
api::ISyntheticSecurityService *ss,
MBOReader *mbo,
bool shared,
co2::Platform platform)
bool shared)
: _workmutex(), _instances(), _sci(), _mbo(mbo), _deferred_handles(),
_callbacks(), _cache(cache) {
std::ostringstream msg;
@@ -612,15 +595,14 @@ class MDConsumerImpl {
msg << symbol.symbol() << "-" << SecurityType_Name(symbol.type())
<< " " << Venue_Name(symbol.exchange());
}
LOG(md, Info, "Created MDConsumerImpl for securities: {} on {}",
msg.str(), co2::Platform_Name(platform));
LOG(md, Info, "Created MDConsumerImpl for securities: {}", msg.str());
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, platform));
s.symbol(), s.type(), s.exchange(), shared));
}
// TODO, add options here
}
@@ -632,13 +614,12 @@ class MDConsumerImpl {
// ISyntheticSecurityService?
if (has_synthetic_instance == false) {
_instances.push_back(make_shared<MDConsumerInst>(
SYNTHETIC_SYMBOL, s.type(), api::VENUE_BTS, shared,
platform));
SYNTHETIC_SYMBOL, s.type(), api::VENUE_BTS, shared));
}
has_synthetic_instance = true;
} else {
_instances.push_back(make_shared<MDConsumerInst>(
s.symbol(), s.type(), s.exchange(), shared, platform));
s.symbol(), s.type(), s.exchange(), shared));
}
}
}
@@ -1022,7 +1003,7 @@ class MDConsumerImpl {
// MDConsumer public interface, forwards to pimple
///////////////////////////////////////////////////////////////////////////////////////////////////
void log_cons(const std::set<UserSymbol> *symbols, co2::Platform platform) {
void log_cons(const std::set<UserSymbol> *symbols) {
std::string log_str = "";
if (symbols != nullptr) {
for (const auto &s : *symbols) {
@@ -1030,8 +1011,7 @@ void log_cons(const std::set<UserSymbol> *symbols, co2::Platform platform) {
Venue_Name(s.exchange()) + ",";
}
}
LOG(md, Info, "MDConsumer::MDConsumer symbols: {} platform: {} ", log_str,
static_cast<int>(platform));
LOG(md, Info, "MDConsumer::MDConsumer symbols: {}", log_str);
}
MDConsumer::MDConsumer(
@@ -1039,17 +1019,16 @@ MDConsumer::MDConsumer(
api::ISyntheticSecurityService *ss,
MBOReader *mbo,
const std::set<UserSymbol> *symbols,
bool shared,
co2::Platform platform)
bool shared)
: _pimpl() {
log_cons(symbols, platform);
log_cons(symbols);
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, platform));
_pimpl.reset(new MDConsumerImpl(*symbols, cache, ss, mbo, shared));
}
MDConsumer::~MDConsumer() {}
@@ -1114,7 +1093,6 @@ 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();
+1 -3
View File
@@ -36,8 +36,7 @@ class MDConsumer {
api::ISyntheticSecurityService *ss,
MBOReader *mbo = nullptr,
const std::set<UserSymbol> *sym = nullptr,
bool shared = true,
co2::Platform platform = co2::PLATFORM_UNSET);
bool shared = true);
~MDConsumer();
bool all_heartbeats_ok() const;
@@ -89,7 +88,6 @@ 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;
+10 -15
View File
@@ -24,16 +24,15 @@ 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, platform),
ShmSegmentName::hot_name(symbol, type, exchange),
config::get<PublisherOptions>().hot_initial(),
config::get<PublisherOptions>().hot_growth(),
flags & MDProducerFlags::Shared),
_cold(
ShmSegmentName::cold_name(symbol, type, exchange, platform),
ShmSegmentName::cold_name(symbol, type, exchange),
config::get<PublisherOptions>().cold_initial(),
config::get<PublisherOptions>().cold_growth(),
flags & MDProducerFlags::Shared),
@@ -49,7 +48,6 @@ 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");
@@ -130,22 +128,19 @@ class MDProducerImpl {
// MDProducer
//
MDProducer::MDProducer(Platform platform, MDProducerFlags flags)
: _mutex(new std::mutex()), _producers(), _id_recorder(),
_platform(platform), _flags(flags),
MDProducer::MDProducer(MDProducerFlags flags)
: _mutex(new std::mutex()), _producers(), _id_recorder(), _flags(flags),
_smapper(flags & MDProducerFlags::SecurityMapper) {}
#ifndef _WIN32
MDProducer::MDProducer(
Platform platform, SecurityIDRecorder recorder, MDProducerFlags flags)
MDProducer::MDProducer(SecurityIDRecorder recorder, MDProducerFlags flags)
: _mutex(new std::mutex()), _producers(),
_id_recorder(std::make_unique<SecurityIDRecorder>(recorder)),
_platform(platform), _flags(flags),
_smapper(flags & MDProducerFlags::SecurityMapper) {}
_flags(flags), _smapper(flags & MDProducerFlags::SecurityMapper) {}
#endif
MDProducer::MDProducer(MDProducer &&other)
: _mutex(std::move(other._mutex)), _producers(std::move(other._producers)),
_platform(other._platform), _flags(other._flags),
_smapper(_flags & MDProducerFlags::SecurityMapper) {}
_flags(other._flags), _smapper(_flags & MDProducerFlags::SecurityMapper) {
}
MDProducer::~MDProducer() {}
@@ -169,9 +164,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,
_platform, _flags));
_flags));
} else {
newp.reset(new MDProducerImpl(symbol, type, exchange, _platform, _flags));
newp.reset(new MDProducerImpl(symbol, type, exchange, _flags));
}
_producers.push_back(std::move(newp));
return *_producers.back();
+1 -4
View File
@@ -42,10 +42,8 @@ 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);
@@ -72,7 +70,6 @@ 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;
};
+10 -15
View File
@@ -1,5 +1,4 @@
#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"
@@ -40,6 +39,7 @@
#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,7 +97,6 @@ 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) {}
};
@@ -107,8 +106,7 @@ struct CCMDSourceData {
bts::md::GeneralMarketDataListener listener;
CCMDSourceData(bts::api::Venue venue)
: sec_index(), md_producer(bts::co2::PLATFORM_SIM),
listener(sec_index, md_producer, venue) {}
: sec_index(), listener(sec_index, md_producer, venue) {}
};
struct MarketDataGateway::PImpl {
@@ -142,7 +140,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));
}
@@ -244,18 +242,17 @@ 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
@@ -271,9 +268,8 @@ 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{}",
@@ -281,7 +277,6 @@ void MarketDataGateway::_start_onix_glbx( bts::api::md::GlbxMarketDataConfig cfg
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, "Found duplicate entry for platform id '{} - ", key);
LOG(md, Error, "Exchange key '{}' has already been added.", key);
bts_assert(result.second);
}
result.first->second = std::move(entry);
+2 -4
View File
@@ -83,12 +83,10 @@ void apply_work(SettlementRecorderImpl &me, Work work) {
}
}
SettlementRecorder::SettlementRecorder(Platform platform)
: _self(new SettlementRecorderImpl()) {
SettlementRecorder::SettlementRecorder() : _self(new SettlementRecorderImpl()) {
string db_dir =
config::get<config::SystemConfig>().data_path() + "databases/md/";
string db_path =
db_dir + "/settlements_" + Platform_Name(platform) + ".sqlite";
string db_path = db_dir + "/settlements.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(Platform platform);
SettlementRecorder();
~SettlementRecorder();
// Put settlement update in a queue
+6 -8
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,9 +137,8 @@ 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), platform(co2::PLATFORM_UNSET),
trade_count(0), coconut_status_count(0), exchange_status_count(0),
rfq_count(0) {}
security_count(0), has_depth(false), trade_count(0),
coconut_status_count(0), exchange_status_count(0), rfq_count(0) {}
uint32_t version;
size_t count;
@@ -152,7 +151,6 @@ 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