scratch: spark clang build check pt

* This is a new preset, so won't break existing/default build
* Now we can use clangd on windows!

.gitmodules

@@ -1,3 +1,6 @@
 [submodule "externals"]
 	path = externals
 	url = git@bitbucket.org:bluetradesys/externals.git
+[submodule "vcpkg"]
+	path = vcpkg
+	url = https://github.com/microsoft/vcpkg.git

CMakeLists.txt

@@ -2,6 +2,10 @@ cmake_minimum_required(VERSION 3.12)
 project(spark)
 set(CMAKE_CXX_STANDARD 20)
 
+if(MY_TEST_VCPKG)
+  add_subdirectory(src/lib_system)
+  add_subdirectory(src/app_tester)
+else()
 if(WIN32)
   set(platform "windows")
 else()
@@ -154,3 +158,4 @@ else()
       DESTINATION "${SPARK_INSTALL_PREFIX}/bin")
   endif()
 endif()
+endif()

CMakePresets.json

@@ -26,6 +26,7 @@
         "CMAKE_CXX_STANDARD": "20",
         "CMAKE_SHARED_LINKER_FLAGS_DEBUG": "/DEBUG",
         "CMAKE_SHARED_LINKER_FLAGS_RELEASE": "/OPT:REF /OPT:ICF /LTCG",
+        "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",
@@ -33,6 +34,42 @@
       }
     },
 
+    {
+      "name": "spark-clang",
+      "binaryDir": "${sourceDir}/build/${presetName}",
+      "generator": "'Ninja Multi-Config'",
+      "cacheVariables": {
+        "CMAKE_C_COMPILER": "clang",
+        "CMAKE_CXX_COMPILER": "clang++",
+        "CMAKE_EXPORT_COMPILE_COMMANDS": {
+          "type": "BOOL",
+          "value": "ON"
+        },
+        "CMAKE_CXX_STANDARD": "20",
+        "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"
+      }
+    },
+
+    {
+      "name": "tester",
+      "generator": "Visual Studio 17 2022",
+      "binaryDir": "${sourceDir}/build/tester",
+      "cacheVariables": {
+        "CMAKE_CXX_STANDARD": "20",
+        "CMAKE_SHARED_LINKER_FLAGS_DEBUG": "/DEBUG",
+        "CMAKE_SHARED_LINKER_FLAGS_RELEASE": "/OPT:REF /OPT:ICF /LTCG",
+        "CMAKE_TOOLCHAIN_FILE": "${sourceDir}/vcpkg/scripts/buildsystems/vcpkg.cmake",
+        "MY_TEST_VCPKG": {
+          "type": "BOOL",
+          "value": "ON"
+        }
+      }
+    },
+
     {
       "name": "sparkd",
       "condition": {
@@ -257,6 +294,35 @@
       "configuration": "Release"
     },
 
+    {
+      "name": "spark-debug-clang",
+      "displayName": "Debug (clang)",
+      "inherits": "spark",
+      "configurePreset": "spark-clang",
+      "configuration": "Debug"
+    },
+
+    {
+      "name": "spark-release-clang",
+      "displayName": "Release (clang)",
+      "inherits": "spark",
+      "configurePreset": "spark-clang",
+      "configuration": "Release"
+    },
+
+    {
+      "name": "tester",
+      "hidden": true
+    },
+
+    {
+      "name": "tester-debug",
+      "displayName": "Debug",
+      "inherits": "tester",
+      "configurePreset": "tester",
+      "configuration": "Debug"
+    },
+
     {
       "name": "sparkd",
       "condition": {

cmake/InitExecutable.cmake

@@ -0,0 +1,8 @@
+function(init_executable source_dir)
+  file(
+    GLOB_RECURSE ${PROJECT_NAME}_Sources
+    RELATIVE ${CMAKE_CURRENT_SOURCE_DIR}
+    ${source_dir}/*.cpp)
+  add_executable(${PROJECT_NAME} ${${PROJECT_NAME}_Sources})
+  target_include_directories(${PROJECT_NAME} PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
+endfunction(init_executable)

cmake/InitLibrary.cmake

@@ -0,0 +1,29 @@
+function(init_library source_dir)
+  file(
+    GLOB_RECURSE ${PROJECT_NAME}_Sources
+    RELATIVE ${CMAKE_CURRENT_SOURCE_DIR}
+    ${source_dir}/*.cpp)
+
+  # Note: cpp files inside 'tests' dir(s) are considered tests.
+  # Remove tests from main sources, and add them to test sources
+  set(${PROJECT_NAME}_Test_Sources)
+  foreach(item ${${PROJECT_NAME}_Sources})
+    if(${item} MATCHES "tests/*.cpp")
+      list(REMOVE_ITEM ${PROJECT_NAME}_Sources ${item})
+      list(APPEND ${PROJECT_NAME}_Test_Sources ${item})
+    endif()
+  endforeach()
+
+  # Add main library
+  add_library(${PROJECT_NAME} ${${PROJECT_NAME}_Sources})
+  target_include_directories(${PROJECT_NAME} PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
+
+  # Add tests executable (if there are any)
+  if (${PROJECT_NAME}_Test_Sources)
+    add_executable(${PROJECT_NAME}-tests ${${PROJECT_NAME}_Test_Sources})
+    target_link_libraries(${PROJECT_NAME}-tests PRIVATE ${PROJECT_NAME})
+    # Catch2
+    find_package(Catch2 CONFIG REQUIRED COMPONENTS Catch2WithMain)
+    target_link_libraries(${PROJECT_NAME}-tests PRIVATE Catch2::Catch2WithMain)
+  endif()
+endfunction(init_library)

src/DataStreamSuite/DataStream/Primitives/AddColumns/AddColumnsStream.cs

@@ -1,13 +1,13 @@
-// using System;
-// using System.Data;
-// using System.Collections;
-// using BCG.Utilities;
-// using BCG.Utilities.General;
-// namespace BCG
-// {
+using System;
+using System.Data;
+using System.Collections;
+using BCG.Utilities;
+using BCG.Utilities.General;
+namespace BCG
+{
 
-//    public class AddColumnsStream : MapStream
-//    {
-//       public AddColumnsStream(ViewStream vs, string columnDef) : base (vs, new ColumnAdder (columnDef))  {}    
-//    }
-// }
+   public class AddColumnsStream : MapStream
+   {
+      public AddColumnsStream(ViewStream vs, string columnDef) : base (vs, default)  {}    
+   }
+}

src/DataStreamSuite/DataStream/Primitives/FileStreams/PlaybackStream.cs

@@ -104,13 +104,14 @@ namespace BCG {
             if (!p.replayed)
                return p;
          }
-         return null;
-      }
-
-      private PlaybackPacket _get_previous_packet(PlaybackPacket p) {
-         if (p.idx == 0)
-            return null;
-         return _packets[p.idx - 1];
+         return default;
+         // else
+         // {
+         //    double secondsElapsed = new BinaryReader(myFileReader).ReadDouble();
+         //    myNextPacketTime = myPlaybackStart + new TimeSpan(0, 0, 0, 0, (int)(1000 * secondsElapsed / myPlaybackSpeed));
+         //    if (onProgress != null)
+         //       onProgress(myFileReader.Position, myFileReader.Length);
+         // }
       }
 
       #region IWorker Members
@@ -130,7 +131,8 @@ namespace BCG {
             _fire_packet(ops, row_array);
             curr_packet.replay_timestamp = PersistorStream.epoch_now();
          } else {
-            PlaybackPacket prev_packet = _get_previous_packet(curr_packet);
+            PlaybackPacket prev_packet = new(0, 0, Array.Empty<byte>());
+            // PlaybackPacket prev_packet = _get_previous_packet(curr_packet);
             // this is the time we need to fire (replay) this packet
             long time_to_fire = prev_packet.replay_timestamp + curr_packet.offset_from_previous;
             long now_epoch = PersistorStream.epoch_now();

src/DataStreamSuite/PubSub/Testing/SimpleTCPClient.cs

@@ -102,14 +102,6 @@ namespace BCG
             System.Threading.Thread.Sleep(1000);
          }
 
-         // Task.Run(async () =>
-         // {
-         //    while (!_conn.isDisposed)
-         //    {
-         //       await _conn.send_async(buff);
-         //       await Task.Delay(100);
-         //    }
-         // });
       }
 
       public void onData(RemoteMessage msg)

src/app_tester/CMakeLists.txt

@@ -0,0 +1,6 @@
+cmake_minimum_required(VERSION 3.23)
+project(bts-tester)
+include(${CMAKE_SOURCE_DIR}/cmake/InitExecutable.cmake)
+init_executable(tester)
+target_link_libraries(${PROJECT_NAME} PUBLIC
+  bts-system)

src/app_tester/tester/main.cpp

@@ -0,0 +1,9 @@
+#include "system/foo.h"
+
+#include <string>
+
+int main(int argc, char ** argv) {
+   bts::system::Foo f{"jules"};
+   f.say_hello();
+   return 0;
+}

src/client/spark/App.xaml.cs

@@ -84,7 +84,12 @@ namespace mbts.ui.shell {
 
          base.OnStartup(e);
          _runtime = new ShellBootstrapper(
-            _opts?.user, _opts?.password, _opts?.sparkd, _opts?.restore_workspace);
+            _opts?.user,
+            _opts?.password,
+            _opts?.sparkd,
+            _opts?.restore_workspace,
+            _opts.verbose ? LogLevel.Diag : LogLevel.Warn,
+            _opts.verbose ? LogLevel.Diag : LogLevel.Info);
          _runtime.Run();
 
          // Select text on tab in

src/client/spark/ShellBootstrapper.cs

@@ -1,4 +1,6 @@
 namespace mbts.ui.shell {
+   using bts;
+
    using mbts.config;
    using mbts.ui.extensions;
 
@@ -23,10 +25,13 @@
          string user = null,
          string password = null,
          string config_source = null,
-         bool? restore_workspace = null
+         bool? restore_workspace = null,
+         LogLevel config_console_log_level = LogLevel.Warn,
+         LogLevel config_file_log_level = LogLevel.Info
       ) : base(config_source) {
          // Get the LoginModel instance on this thread to make sure it performs syncronous config initialization.
-         var login_model = LoginModel.init(config_source);
+         var login_model = LoginModel.init(
+            config_source, config_console_log_level, config_file_log_level);
          // Show the dialog on a background thread so animations aren't blocked during the main application startup
          var thread = new Thread(thread_start) {
             IsBackground = true,

src/client/spark/Spark.Cli/SparkCmdOpts.cs

@@ -42,6 +42,11 @@ namespace Spark.Cli {
       [Option(_SPARKD_OPT_NAME, MetaValue = "URL", HelpText = "The config server url")]
       public string? sparkd { get; set; }
 
+      [Option(
+         "verbose",
+         HelpText ="Verbose logging while initializing config (once initialized, the log levels in config are used)")]
+      public bool verbose { get; set; }
+
       /// <summary>
       /// Conform opts to modern cli for backwards compatibility
       /// </summary>

src/client/spark/view/Shell/Login/LoginModel.cs

@@ -62,14 +62,19 @@ namespace mbts.ui.shell {
       static Lazy<LoginModel> _instance;
 
       readonly string _config_source;
+      readonly LogLevel _config_console_log_level;
+      readonly LogLevel _config_file_log_level;
       private string _workspace;
       private PreloginWorkspaceEnumerator _enum_workspace;
 
       ServiceContext _ctx;
       DispatcherFrame _frame = new DispatcherFrame();
 
-      LoginModel(string config_source) {
+      LoginModel(
+         string config_source, LogLevel config_console_log_level, LogLevel config_file_log_level) {
          _config_source = config_source;
+         _config_console_log_level = config_console_log_level;
+         _config_file_log_level = config_file_log_level;
          init().ignore();
       }
 
@@ -84,10 +89,12 @@ namespace mbts.ui.shell {
       public bool HasFatalError { get; set; }
       public User[] Users { get; private set; }
 
-      public static LoginModel init(string config_source) {
+      public static LoginModel init(
+         string config_source, LogLevel config_console_log_level, LogLevel config_file_log_level) {
          // '_instance' is lazy to guarentee it is only instantiated once
          Interlocked.CompareExchange(
-            ref _instance, new(() => new(config_source)), null);
+            ref _instance,
+            new(() => new(config_source, config_console_log_level, config_file_log_level)), null);
          return _instance.Value;
       }
 
@@ -119,7 +126,7 @@ namespace mbts.ui.shell {
       public async Task<bool> init() {
          // Make sure config is initialized and working
          try {
-            Configuration.initialize(_config_source);
+            Configuration.initialize(_config_source, _config_console_log_level, _config_file_log_level);
          } catch (ApplicationException e) {
             ErrorMessage = (e.Message ?? string.Empty) + " " + (App.stunnel_job_error ?? string.Empty);
             HasFatalError = true;

src/client/spark_api/api/ServiceContext.cs

@@ -28,7 +28,7 @@ namespace bts.api.core {
 
       public static ServiceContext create(string? config_server = null) {
          var args = Environment.GetCommandLineArgs();
-         BTS.init(config_server, args);
+         BTS.init(config_server, LogLevel.Warn, LogLevel.Info, args);
          var qbus = ApiNativeMethods.api_create_quickbus(null, 0);
          return new ServiceContext(qbus);
       }

src/client/spark_core/Collections/Extensions/EnumerableExtensions.cs

@@ -1197,7 +1197,6 @@ namespace System.Collections.Generic {
       /// A tuple whose items each represent the sum of their repsective items in
       /// <paramref name="source"/>
       /// </returns>
-      [RequiresPreviewFeatures]
       public static (T1, T2) Sum<T1, T2>(this IEnumerable<(T1, T2)> source)
          where T1 : INumber<T1>
          where T2 : INumber<T2> =>
@@ -1210,7 +1209,6 @@ namespace System.Collections.Generic {
       /// A tuple whose items each represent the sum of their repsective items in
       /// <paramref name="source"/>
       /// </returns>
-      [RequiresPreviewFeatures]
       public static (T1, T2) Sum<T1, T2>(this IEnumerable<(T1, T2)> source, T1 seed1, T2 seed2)
          where T1 : IAdditionOperators<T1, T1, T1>
          where T2 : IAdditionOperators<T2, T2, T2> =>
@@ -1224,7 +1222,6 @@ namespace System.Collections.Generic {
       /// A tuple whose items each represent the sum of their repsective items in
       /// <paramref name="source"/>
       /// </returns>
-      [RequiresPreviewFeatures]
       public static (T1, T2, T3) Sum<T1, T2, T3>(this IEnumerable<(T1, T2, T3)> source)
          where T1 : INumber<T1>
          where T2 : INumber<T2>
@@ -1238,7 +1235,6 @@ namespace System.Collections.Generic {
       /// A tuple whose items each represent the sum of their repsective items in
       /// <paramref name="source"/>
       /// </returns>
-      [RequiresPreviewFeatures]
       public static (T1, T2, T3) Sum<T1, T2, T3>(
          this IEnumerable<(T1, T2, T3)> source, T1 seed1, T2 seed2, T3 seed3)
          where T1 : IAdditionOperators<T1, T1, T1>
@@ -1255,7 +1251,6 @@ namespace System.Collections.Generic {
       /// A tuple whose items each represent the sum of their repsective items in
       /// <paramref name="source"/>
       /// </returns>
-      [RequiresPreviewFeatures]
       public static (T1, T2, T3, T4) Sum<T1, T2, T3, T4>(this IEnumerable<(T1, T2, T3, T4)> source)
          where T1 : INumber<T1>
          where T2 : INumber<T2>
@@ -1270,7 +1265,6 @@ namespace System.Collections.Generic {
       /// A tuple whose items each represent the sum of their repsective items in
       /// <paramref name="source"/>
       /// </returns>
-      [RequiresPreviewFeatures]
       public static (T1, T2, T3, T4) Sum<T1, T2, T3, T4>(
          this IEnumerable<(T1, T2, T3, T4)> source, T1 seed1, T2 seed2, T3 seed3, T4 seed4)
          where T1 : IAdditionOperators<T1, T1, T1>

src/client/spark_core/config/Configuration.cs

@@ -50,14 +50,18 @@ namespace mbts.config {
          }
       }
 
-      public static void initialize(string? config_source = null) {
+      public static void initialize(
+         string? config_source = null,
+         LogLevel config_console_log_level = LogLevel.Warn,
+         LogLevel config_file_log_level = LogLevel.Info) {
          if (_state == State.OK) {
             return;
          }
+         var cli_args = Environment.GetCommandLineArgs();
          // Considered doing this async on a background thread but it wouldn't do any good. The next
          // guy down the line to ask for config will just block until this completes.
          try {
-            BTS.init(config_source);
+            BTS.init(config_source, config_file_log_level, config_console_log_level, cli_args);
          } catch (ApplicationException) {
             _state = State.Error;
             throw;

src/client/spark_core/utils_api/BTS.cs

@@ -9,17 +9,20 @@ namespace bts
    {
       static int _needs_init_spark_lib = 0;
 
-      public static void init(string config_source = null) {
-         var args = Environment.GetCommandLineArgs();
-         init(config_source, args);
-      }
-
-      public static void init(string config_source = null, params string[] args) {
+      public static void init(
+         string config_source = null,
+         LogLevel console_log_level = LogLevel.Warn,
+         LogLevel file_log_level = LogLevel.Info,
+         params string[] args) {
          utils.BTSNativeMethods.provide_command_line(args, args.Length);
 
          var buffer = new StringBuilder(4096);
          bool config_ok = utils.BTSNativeMethods.initialize_nothrow(
-            config_source ?? string.Empty, buffer, buffer.Capacity);
+            config_source ?? string.Empty,
+            buffer,
+            buffer.Capacity,
+            (short)console_log_level,
+            (short)file_log_level);
          if (!config_ok) {
             throw new ApplicationException(buffer.ToString());
          }

src/client/spark_core/utils_api/BTSNativeMethods.cs

@@ -27,6 +27,10 @@ namespace bts.utils
 
       [DllImport("libspark")]
       public static extern bool initialize_nothrow(
-         string config_source, StringBuilder buffer, int length);
+         string config_source,
+         StringBuilder buffer,
+         int length,
+         short console_log_level,
+         short file_log_level);
    }
 }

src/lib/api/core/ApiClientInterop.cpp

@@ -37,7 +37,7 @@ CAPI Quickbus *api_create_quickbus(const char *qbconfig, long length) {
    // If config hasn't been initialized yet do it now and suppress exceptions.
    // When we're called from a nuget package we may not have defaults which can
    // cause errors
-   config::initialize_nothrow("", nullptr, 0);
+   config::initialize_nothrow("", nullptr, 0, bts::Warn.ordinal, bts::Info.ordinal);
 
    config::QuickbusNodeCfg cfg;
    if (length > 0 && qbconfig != nullptr &&

src/lib/api/md/GlbxMarketDataConfig.proto

@@ -0,0 +1,22 @@
+syntax = "proto3";
+
+import "lib/api/domain/SecurityCriteria.proto";
+
+package bts.api.md;
+
+option csharp_namespace = "bts.api.md";
+
+
+message GlbxMarketDataConfig {
+  uint32 unique_check_value = 1;
+  //uint32 num_depth_levels = 1;
+  //uint32 depth_throttle_ms = 2;
+  bts.api.domain.SecurityCriteria security_criteria = 3;
+  // Market data isn't subscribed for securities listed here (even if they are
+  // included by 'security_criteria'). There are some securities that don't
+  // fit into our system just yet, such as 'BTCBIDR', which trades in prices
+  // larger than our cpp 'domain::Price' class supports.
+  //repeated string excluded_securities = 4;
+  //string url_rest_base = 5;
+  //string url_ws_base = 6;
+}

src/lib/api/md/MarketDataConfig.proto

@@ -10,7 +10,7 @@ import "lib/api/md/MarketDataReplayConfig.proto";
 import "lib/api/md/DeribitMarketDataConfig.proto";
 import "lib/api/md/EmumbaDeribitMarketDataConfig.proto";
 import "lib/api/md/OkxMarketDataConfig.proto";
-
+import "lib/api/md/GlbxMarketDataConfig.proto";
 
 option csharp_namespace = "bts.api.md";
 
@@ -24,6 +24,7 @@ message MarketDataConfig {
   repeated DeribitMarketDataConfig deribit = 5;
   repeated CoinbaseMarketDataConfig coinbase =6;
   repeated OkxMarketDataConfig okx = 7;
+  repeated GlbxMarketDataConfig glbx = 8;
   // Add an entry here for all the exchanges that need their market data to be recorded
 
   repeated EmumbaDeribitMarketDataConfig emumba_deribit = 100;

src/lib/api/md/MarketDataConfigUtils.cpp

@@ -19,6 +19,8 @@ std::set<bts::api::Venue> MarketDataConfigUtils::get_live_exchanges(
       rv.insert(bts::api::VENUE_BINANCE_US);
    if (cfg.deribit_size() > 0)
       rv.insert(bts::api::VENUE_DERIBIT);
+   if (cfg.glbx_size()>0)
+      rv.insert(bts::api::VENUE_GLBX);
    return rv;
 }
 

src/lib/api/md/MarketDataRecordingConfig.proto

@@ -11,4 +11,5 @@ option cc_enable_arenas = true;
 message MarketDataRecordingConfig {
    uint32 num_of_seconds_to_run = 10;
    bts.api.Venue venue = 20;
+   uint32 n_book_levels = 30;
 }

src/lib/cmake/GetWindowsClientLinkLibraries.cmake

@@ -1,8 +1,8 @@
 function(get_link_libraries dest)
+  find_package(fmt CONFIG REQUIRED)
   set(${dest}
     crypt32
-    debug fmtd
-    optimized fmt
+    fmt::fmt-header-only
     libcrypto
     libcurl
     libboost_system

src/lib/config/ConfigLog.h

@@ -15,7 +15,9 @@ namespace internal {
 // To allow config diagnostic logging
 class ConfigLog {
  public:
-   ConfigLog();
+   explicit ConfigLog(
+      bts::log::Level console_log_level = bts::Warn,
+      bts::log::Level file_log_level = bts::Info);
    ~ConfigLog();
 
    void info(const std::string &msg);
@@ -50,7 +52,7 @@ class ConfigLog {
 // Definition
 //////////////////////////////////////////////////
 
-inline ConfigLog::ConfigLog()
+inline ConfigLog::ConfigLog(bts::log::Level console_log_level, bts::log::Level file_log_level)
     : print_errors_(true), logger_hub_("config_logger"),
       logger_(logger_hub_.make_logger()) {
    log::LoggingConfig cfg;
@@ -66,9 +68,9 @@ inline ConfigLog::ConfigLog()
    }
 
    cfg.deployment_type = "Config";
-   f[log::Cfg.ordinal] = Warn;
+   f[log::Cfg.ordinal] = console_log_level;
    cfg.sinks["console"] = std::make_pair(log::console_sink(cfg), f);
-   f[log::Cfg.ordinal] = Info;
+   f[log::Cfg.ordinal] = file_log_level;
    cfg.sinks["syslog"] = std::make_pair(log::syslog_sink(cfg), f);
    logger_hub_.config(cfg);
 
@@ -126,7 +128,7 @@ inline bool ConfigLog::enqueue(
       return false;
    }
    return logger_.enqueue(
-       log::log_ts(), log::Cfg, level, file, line, msg_template, msg_args...);
+       log::log_ts(), log::Cfg, level, file, line, msg_template, std::move(msg_args)...);
 }
 
 } // namespace internal

src/lib/config/ConfigQuerier.cpp

@@ -97,7 +97,8 @@ void run_config_query(
 void ConfigQuerier::init_(bool verbose) {
    char buffer[4096] = {0};
    utils::TimeStamp start = utils::TimeStamp::now();
-   bool success = config::initialize_nothrow("", buffer, sizeof(buffer));
+   bool success = config::initialize_nothrow(
+      "", buffer, sizeof(buffer), bts::Warn.ordinal, bts::Info.ordinal);
    double time = (utils::TimeStamp::now() - start).seconds();
 
    if (success) {

src/lib/config/ConfigurationState.cpp

@@ -362,7 +362,7 @@ void check_rule(
 }
 
 void ConfigurationState::check_rules() {
-   initialize("");
+   initialize("", bts::Warn.ordinal, bts::Info.ordinal);
    std::vector<std::shared_ptr<proto::Message>>::iterator iter;
    for (iter = _config_chain.begin(); iter != _config_chain.end(); ++iter) {
       proto::Message *message = (*iter).get();
@@ -371,12 +371,13 @@ void ConfigurationState::check_rules() {
    check_rule(_defaults.get(), _rule_engine);
 }
 
-void ConfigurationState::initialize_settings(string config_source) {
+void ConfigurationState::initialize_settings(
+   string config_source, bts::log::Level console_log_level, bts::log::Level file_log_level) {
    std::lock_guard<std::recursive_mutex> lock(_initialization_mutex);
    if (_is_initialized)
       return;
 
-   ConfigLog log;
+   ConfigLog log{console_log_level, file_log_level};
    CFG_LOG(log, Info, "Initializing settings from '{}'", config_source);
    // For verbose config logs, uncomment line below
    // log.set_print_errors(true);

src/lib/config/ConfigurationState.h

@@ -60,7 +60,10 @@ class ConfigurationState {
     * @param config_source The source from which to load values. Can be a path
     * (file or dir), or URL (config server).
     */
-   void initialize_settings(std::string config_source = "");
+   void initialize_settings(
+      std::string config_source = "",
+      bts::log::Level console_log_level = bts::Warn,
+      bts::log::Level file_log_level = bts::Info);
 
    int load(const std::string &url, ConfigLog &);
    int load_string(const std::string &values, bool print_errors = false);

src/lib/config/config.cpp

@@ -27,14 +27,24 @@ internal::ConfigurationState config_state;
 // Public API
 //////////////////////////////////////////
 
-extern "C" BTS_EXPORT void initialize(const char *config_source) {
-   config_state.initialize_settings(string{config_source});
+extern "C" BTS_EXPORT void initialize(
+    const char *config_source,
+    std::int16_t console_log_level,
+    std::int16_t file_log_level) {
+   config_state.initialize_settings(
+      string{config_source},
+      bts::log::Level{static_cast<std::uint8_t>(console_log_level)},
+      bts::log::Level{static_cast<std::uint8_t>(file_log_level)});
 }
 
 extern "C" BTS_EXPORT bool initialize_nothrow(
-    const char *config_source, char *error_message, size_t len) {
+    const char *config_source,
+    char *error_message,
+    size_t len,
+    std::int16_t console_log_level,
+    std::int16_t file_log_level) {
    try {
-      initialize(config_source);
+      initialize(config_source, console_log_level, file_log_level);
    } catch (std::exception &e) {
       std::string msg = e.what();
       if (msg.size() < len) {

src/lib/config/config.h

@@ -27,12 +27,19 @@ namespace config {
 // config can result in network requests this function is available for
 // applications that want to manage pauses in execution. Ex: initialize() on a
 // background thread so that config::get<T>() is fast on a UI thread
-extern "C" BTS_EXPORT void initialize(const char *config_path);
+extern "C" BTS_EXPORT void initialize(
+   const char *config_path,
+   std::int16_t console_log_level,
+   std::int16_t file_log_level);
 
 // Calls initialize() but catches any thrown exception and copies it to the
 // provided buffer
 extern "C" BTS_EXPORT bool initialize_nothrow(
-   const char *config_source, char *error_msg, size_t length);
+   const char *config_source,
+   char *error_msg,
+   size_t length,
+   std::int16_t console_log_level,
+   std::int16_t file_log_level);
 
 std::vector<const google::protobuf::FieldDescriptor *> get_fields(
     const std::vector<std::string> keys = {},

src/lib/config/profiles/default/market_data.conf

@@ -9,7 +9,7 @@ HistoricMarketDataConfig {
 
 PlatformMap {
   value {
-    glbx_platform: "PLATFORM_SIM"
+    glbx_platform: "PLATFORM_MDP3ONIXS"
     ice_platform: "PLATFORM_ICEONIXS"
   }
 }
@@ -24,6 +24,8 @@ PublisherOptions {
     hot_growth: 1024
     cold_initial: 8192
     cold_growth: 2048
+    channel_prefix: ""
+    disable_heartbeat: false
   }
 }
 
@@ -97,11 +99,18 @@ CMEOptions {
     feed_engine {
       name: "mdp3feed"
       spin_time: 0
+      channels_per_thread: 3
+      data_wait: 10
+      upp_socket_buffer_size: 268435456
     }
     channel {
       id: "*"
       feed_engine: "mdp3feed"
       use_tcp_replay: false
+      recovery_queue_size: 640000
+      lost_packet_wait_timeout_ms: 10
+      use_natural_refresh: false
+      join_snapshot_on_gap: false
     }
   }
 }
@@ -222,3 +231,10 @@ MarketDataConfig {
     }
   }
 }
+
+
+ OnixsOptions {
+  value{
+      onixs_license: "{default}"
+  }
+}

src/lib/config/profiles/default/toggles.conf

@@ -9,7 +9,11 @@ OrdersSoftCleanup {
       enabled: true
    }
 }
-
+PlacementInQueue {
+  vlaue{
+    enabled: true;
+  }
+}
 RiskLimit {
 value{
   risk_limits{

src/lib/datastreams/Tables/StreamData.h

@@ -0,0 +1,185 @@
+#include "lib/utils/bts_assert.h"
+#include <any>
+#include <iostream>
+#include <map>
+#include <string>
+#include <typeindex>
+#include <unordered_map>
+#include <variant>
+#include <vector>
+// this file has all the classes that represent streaming tabular data
+// schema ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+namespace bts::streams {
+
+using RowID = int;
+
+// Represents a column of a schema (tuple of type, name and col number). For the type we use
+// std::type_index
+
+class ColumnInfo {
+ public:
+   ColumnInfo(
+       const std::string &column_name,
+       const std::type_index &column_type,
+       size_t col_index)
+       : _name(column_name), _type(column_type), _index(col_index) {}
+   const std::string &column_name() const { return _name; }
+   size_t col_index() const { return _index; }
+   const std::type_index &column_type() const { return _type; }
+
+ private:
+   const std::string _name;
+   const std::type_index _type;
+   const size_t _index;
+};
+class StreamSchema {
+ public:
+   bool add_column(
+       const std::string &column_name, const std::type_index &column_type) {
+      auto it = _col_map.find(column_name);
+      if (it != _col_map.end()) {
+         return false;
+      } else {
+         _col_map.insert(
+             {column_name, {column_name, column_type, _col_map.size()}});
+         return true;
+      }
+   }
+   ColumnInfo const *get_column(const std::string &col_name) const {
+      auto it = _col_map.find(col_name);
+      if (it == _col_map.end())
+         return nullptr;
+      return &it->second;
+   }
+   ColumnInfo *get_column(const std::string &col_name) {
+      auto it = _col_map.find(col_name);
+      if (it == _col_map.end())
+         return nullptr;
+      return &it->second;
+   }
+   size_t size() const { return _col_map.size(); }
+
+ private:
+   std::map<std::string, ColumnInfo> _col_map = {};
+};
+// row ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+class StreamRow {
+ public:
+   StreamRow(const StreamSchema &schema) : schema(schema) {
+      data.reserve(schema.size());
+      for (size_t i = 0; i < schema.size(); i++) {
+         std::any any;
+         data.push_back(any);
+      }
+   }
+   RowID id() const { return get_value<int>("id"); }
+   template <typename T>
+   bool set_value(const std::string &columnName, const T &value) {
+      ColumnInfo const *col_info = schema.get_column(columnName);
+      if (col_info == nullptr) {
+         throw std::invalid_argument(
+             "Column " + columnName + " not found in the schema");
+         return false;
+      }
+      if (col_info->column_type() != typeid(T)) {
+         throw std::invalid_argument("Column " + columnName + " type mismatch");
+         return false;
+      }
+      data[col_info->col_index()] = value;
+      return true;
+   }
+   template <typename T>
+   T get_value(const std::string &columnName) const {
+      ColumnInfo const *col_info = schema.get_column(columnName);
+      if (col_info == nullptr) {
+         throw std::invalid_argument(
+             "Column " + columnName + " not found in the schema");
+      }
+      if (col_info->column_type() != typeid(T)) {
+         throw std::invalid_argument("Column " + columnName + " type mismatch");
+      }
+      return std::any_cast<T>(data[col_info->col_index()]);
+   }
+   // template <typename T> // T operator[](const std::string &column_name) const { // return get_value<T>(column_name); // }
+ private:
+   const StreamSchema &schema; // this is where the actual memory is held
+   std::vector<std::any> data;
+};
+
+// table /////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+class StreamTable {
+ private:
+   std::map<RowID, StreamRow const *> _row_map;
+
+ public:
+   StreamTable(const StreamSchema &schema_) : _schema(schema_) {}
+   bool add_row(StreamRow const *row) {
+      RowID id = row->id();
+      auto it = _row_map.find(id);
+      if (it != _row_map.end()) {
+         return false;
+      } else {
+         _row_map.insert({id, row});
+         return true;
+      }
+   }
+   StreamRow const *get_row(RowID id) const {
+      const auto &it = _row_map.find(id);
+      if (it == _row_map.end())
+         return nullptr;
+      return it->second;
+   }
+   StreamRow const *operator[](RowID id) const { return get_row(id); }
+   size_t size() const { return _row_map.size(); }
+   const StreamSchema &schema() const { return _schema; }
+
+ private:
+   const StreamSchema &_schema;
+};
+// stream updates and subscription ////////////////////////////////////////////
+class StreamUpdate {
+ public:
+   const StreamTable &complete_state;
+   std::set<RowID> deleted_ids;
+   std::vector<StreamRow const *> added_rows;
+   StreamUpdate(
+       const StreamTable &complete_state_,
+       const std::set<RowID> deleted_ids_,
+       const std::vector<StreamRow const *> &added_rows_)
+       : complete_state(complete_state_), deleted_ids(deleted_ids_),
+         added_rows(added_rows_) {}
+};
+
+class DirectSubsription {};
+class CloudSubscirption {};
+// template <
+//     typename T,
+//     typename = std::enable_if_t<
+//         std::is_same<T, DirectSubsription>::value ||
+//         std::is_same<T, CloudSubscirption>::value>>
+
+class StreamSubscription {
+ private:
+   std::variant<DirectSubsription, CloudSubscirption> subscription;
+   const size_t _sub_id;
+
+ public:
+   StreamSubscription(const DirectSubsription &, size_t subscription_id_)
+       : _sub_id(subscription_id_) {}
+   StreamSubscription(const CloudSubscirption &, size_t subscription_id_)
+       : _sub_id(subscription_id_) {}
+   size_t sub_id() const { return _sub_id; }
+};
+
+class IStreamSync {
+ public:
+   virtual void on_stream_update(
+       const StreamSubscription &, const StreamUpdate &stream_update) = 0;
+};
+
+class StreamSource {
+ public:
+   void subscribe_stream_expression(const StreamSubscription &, IStreamSync &) {
+   }
+};
+} // namespace bts::streams

src/lib/datastreams/Tables/tests/stream_tests.cpp

@@ -0,0 +1,97 @@
+
+#include "catch/catch.hpp"
+#include "lib/api/service_helpers/AdminServiceUtils.h"
+#include "lib/api/services/AdminService.server.pb.h"
+#include "lib/config/config.h"
+#include "lib/datastreams/Tables/StreamData.h"
+#include "lib/utils/bts_chrono.h"
+#include "lib/utils/helper_macros.h"
+#include <memory>
+
+using namespace bts::streams;
+
+TEST_CASE("schema", "[streams]") {
+   // create schema
+   bts::streams::StreamSchema schema = {};
+   schema.add_column("id", std::type_index(typeid(int)));
+   schema.add_column("sec_id", std::type_index(typeid(uint32_t)));
+   schema.add_column("bid_price", std::type_index(typeid(double)));
+   schema.add_column("bid_qty", std::type_index(typeid(uint32_t)));
+   schema.add_column("ask_price", std::type_index(typeid(double)));
+   schema.add_column("ask_qty", std::type_index(typeid(uint32_t)));
+   ColumnInfo const *ci = schema.get_column("bid_price");
+   REQUIRE(ci != nullptr);
+   REQUIRE(ci->col_index() == 2);
+   REQUIRE(ci->column_type() == typeid(double));
+}
+
+TEST_CASE("row", "[streams]") {
+   // create schema
+   bts::streams::StreamSchema schema = {};
+   schema.add_column("id", std::type_index(typeid(int)));
+   schema.add_column("sec_id", std::type_index(typeid(uint32_t)));
+   schema.add_column("bid_price", std::type_index(typeid(double)));
+   schema.add_column("bid_qty", std::type_index(typeid(uint32_t)));
+   schema.add_column("ask_price", std::type_index(typeid(double)));
+   schema.add_column("ask_qty", std::type_index(typeid(uint32_t)));
+   ColumnInfo const *ci = schema.get_column("bid_price");
+   REQUIRE(ci != nullptr);
+   REQUIRE(ci->col_index() == 2);
+   REQUIRE(ci->column_type() == typeid(double));
+   bts::streams::StreamRow r(schema);
+   r.set_value("id", 0);
+   r.set_value("sec_id", (uint)0);
+   r.set_value("bid_price", 1.0);
+   r.set_value("bid_qty", (uint32_t)42);
+   r.set_value("ask_price", 1.0);
+   r.set_value("ask_qty", (uint32_t)42);
+   REQUIRE(r.get_value<uint32_t>("bid_qty") == 42);
+   REQUIRE(r.get_value<double>("bid_price") == 1.0);
+}
+
+TEST_CASE("table", "[streams]") {
+   // create schema
+   bts::streams::StreamSchema schema = {};
+   schema.add_column("id", std::type_index(typeid(int)));
+   schema.add_column("sec_id", std::type_index(typeid(uint32_t)));
+   schema.add_column("bid_price", std::type_index(typeid(double)));
+   schema.add_column("bid_qty", std::type_index(typeid(uint32_t)));
+   schema.add_column("ask_price", std::type_index(typeid(double)));
+   schema.add_column("ask_qty", std::type_index(typeid(uint32_t)));
+   ColumnInfo const *ci = schema.get_column("bid_price");
+   REQUIRE(ci != nullptr);
+   REQUIRE(ci->col_index() == 2);
+   REQUIRE(ci->column_type() == typeid(double));
+   bts::streams::StreamRow r1(schema);
+   r1.set_value("id", 0);
+   r1.set_value("sec_id", (uint)0);
+   r1.set_value("bid_price", 1.0);
+   r1.set_value("bid_qty", (uint32_t)42);
+   r1.set_value("ask_price", 1.0);
+   r1.set_value("ask_qty", (uint32_t)42);
+   REQUIRE(r1.get_value<uint32_t>("bid_qty") == 42);
+   REQUIRE(r1.get_value<double>("bid_price") == 1.0);
+   bts::streams::StreamTable table(schema);
+   table.add_row(&r1);
+   bts::streams::StreamRow r2(schema);
+   r2.set_value("id", 1);
+   r2.set_value("sec_id", (uint)1);
+   r2.set_value("bid_price", 1.0);
+   r2.set_value("bid_qty", (uint32_t)52);
+   r2.set_value("ask_price", 1.0);
+   r2.set_value("ask_qty", (uint32_t)52);
+   // REQUIRE(r2["ask_qty"]<uint32_t> == 52);
+   REQUIRE(r2.get_value<uint32_t>("bid_qty") == 52);
+   REQUIRE(r2.get_value<double>("bid_price") == 1.0);
+   table.add_row(&r2);
+   auto r1_ret = table[0];
+   REQUIRE(r1_ret != nullptr);
+   REQUIRE(r1_ret->id() == 0);
+   REQUIRE(r1_ret->get_value<uint32_t>("bid_qty") == 42);
+   REQUIRE(r1_ret->get_value<double>("bid_price") == 1.0);
+   auto r2_ret = table[1];
+   REQUIRE(r2_ret != nullptr);
+   REQUIRE(r2_ret->id() == 1);
+   REQUIRE(r2_ret->get_value<uint32_t>("bid_qty") == 52);
+   REQUIRE(r2_ret->get_value<double>("bid_price") == 1.0);
+}

src/lib/md/MarketDataGateway.cpp

@@ -1,4 +1,5 @@
 #if !defined(_WIN32) && defined(SPARK_ENABLE_CCAPI)
+#include "server/sparkd/md/platforms/mdp3_onixs/MDP3Main.h"
 #include "MarketDataGateway.h"
 #include "lib/api/md/EmumbaDeribitMarketDataConfig.pb.h"
 #include "lib/api/md/MarketDataConfig.pb.h"
@@ -141,6 +142,10 @@ 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()){
+      bts_assert(me_->md_config.glbx().size() == 1);
+      _start_onix_glbx(me_->md_config.glbx().at(0));
+   }
    _start_replays(me_->md_config);
    while (true) {
       std::this_thread::sleep_for(std::chrono::seconds(1));
@@ -236,6 +241,46 @@ void MarketDataGateway::unsubscribe(MarketDataListenerID id) {
    LOG(md, Info, "Unsubscribing listener {}", id);
    me_->listeners.erase(id);
 }
+/// glbx //////////////////////////////////////////////////////////////////////
+//////////////////////////////////////////////////////////////////////////////
+static void run_onixs() {
+   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}));
+   if (!cme_symbols.empty()) {
+      auto *main = new bts::md::MDP3Main();
+      main->start(
+          {cme_symbols.begin(), cme_symbols.end()}, mbo, tssd);
+      platforms.emplace_back(main);
+   }
+   // deribit
+   // auto *d_platform = new bts::md::EmMDListener()
+   // Timers
+   auto &disp = utils::Dispatcher::current();
+   utils::DispatcherTimer heartbeat_timer(disp, 250, [&]() {
+      for (auto &p : platforms)
+         p->send_heartbeats();
+   });
+   heartbeat_timer.start();
+
+   disp.run();
+}
+
+
+
+void MarketDataGateway::_start_onix_glbx( bts::api::md::GlbxMarketDataConfig cfg){
+   LOG(md, Info,
+       "MarketDataGateway::_start_onix_glbx starting t_id:{} t_name:{}. "
+       "cfg:\n{}",
+       os::get_thread_id(), os::get_thread_name(), cfg.DebugString());
+   bts::md::run_onixs();
+}
+
 
 // deribit ////////////////////////////////////////////////////////////////////
 ///////////////////////////////////////////////////////////////////////////////

src/lib/md/MarketDataGateway.h

@@ -1,5 +1,7 @@
 #pragma once
+
 #if !defined(_WIN32) && defined(SPARK_ENABLE_CCAPI)
+#include "lib/api/md/GlbxMarketDataConfig.pb.h"
 #include "MarketDataListenerID.h"
 #include <memory>
 
@@ -31,6 +33,8 @@ class MarketDataGateway {
    MarketDataRegistration _start_cc(const api::md::MarketDataConfig &) noexcept;
    static void _start_emumba_deribit(
        bts::api::md::EmumbaDeribitMarketDataConfig cfg);
+   static void _start_onix_glbx(
+       bts::api::md::GlbxMarketDataConfig cfg);
    void _start_replays(const bts::api::md::MarketDataConfig &cfg);
 };
 

src/lib/md/recording/MDRecorder.cpp

@@ -143,7 +143,7 @@ struct MDRecorder::PImpl {
        const bts::api::md::MarketDataRecordingConfig &recording_config)
        : venue(venue),
          settings(MDRecordingSettings{
-             10,
+             recording_config.n_book_levels(),
              static_cast<uint64_t>(recording_config.num_of_seconds_to_run()) *
                  1000 * 1000 * 1000,
              100000, 1000000000}),
@@ -184,7 +184,8 @@ void MDRecorder::_poll_md() {
       uint64_t now_ns = core::TimeStamp::now_ns();
       for (auto &kv : me->handles) {
          MDRecordingSecEntry &md_entry = kv.second;
-         _poll_books(md_entry);
+         if(me->settings.n_book_levels > 0)
+            _poll_books(md_entry);
          _poll_trades(md_entry);
          _poll_misc(md_entry);
       }

src/lib/md/vendor/OnixsUtils.cpp

@@ -41,7 +41,7 @@ OnixsTemplLicense onixs_create_temporary_license() {
    const auto &options = config::get<OnixsOptions>();
    OnixsTemplLicense linfo;
    linfo.path = options.onixs_license();
-   if (linfo.path == "{default}") {
+   if (linfo.path == "{default}" || linfo.path =="") {
       linfo.path = Path{config::get<SystemConfig>().data_path()} / "licenses";
       auto ident = boost::filesystem::unique_path("%%%%-%%%%-%%%%-%%%%.lic");
       linfo.license = {linfo.path / ident};

src/lib_system/CMakeLists.txt

@@ -0,0 +1,8 @@
+cmake_minimum_required(VERSION 3.23)
+project(bts-system)
+include(${CMAKE_SOURCE_DIR}/cmake/InitLibrary.cmake)
+init_library(system)
+
+# fmt
+find_package(fmt CONFIG REQUIRED)
+target_link_libraries(${PROJECT_NAME} PUBLIC fmt::fmt)

src/lib_system/system/foo.cpp

@@ -0,0 +1,8 @@
+#include "foo.h"
+#include <fmt/format.h>
+
+namespace bts::system {
+   void Foo:: say_hello() {
+      std::cout << fmt::format("Hi, my name is {}", name) << std::endl;
+   }
+}

src/lib_system/system/foo.h

@@ -0,0 +1,13 @@
+#pragma once
+
+#include <iostream>
+#include <string>
+
+namespace bts::system {
+
+struct Foo {
+   std::string name;
+   void say_hello();
+};
+
+}

src/server/sparkd/md/platforms/mdp3_onixs/MDP3Main.cpp

@@ -86,6 +86,7 @@ const onixs::FeedEngineSettings feed_engine_settings_from_config(
    if (settings.num_threads() != 0) {
       result.threadCount(settings.num_threads());
    } else {
+      bts_assert(settings.channels_per_thread()>0);
       // If num_threads is absent determine thread count based on #channels
       result.threadCount((feed_count / settings.channels_per_thread()) + 1);
    }

vcpkg.json

@@ -0,0 +1,5 @@
+{
+  "dependencies": [
+    "fmt"
+  ]
+}