file restructuring for sdk

WARNING: This breaks the sparkd clang build (only works with gcc)

CMakePresets.json

@@ -113,8 +113,8 @@
       "description": "Debug build compiled w/ gcc",
       "binaryDir": "${sourceDir}/build/gcc/debug",
       "cacheVariables": {
-        "CMAKE_C_COMPILER": "gcc",
-        "CMAKE_CXX_COMPILER": "g++",
+        "CMAKE_C_COMPILER": "gcc-12",
+        "CMAKE_CXX_COMPILER": "g++-12",
         "CMAKE_CXX_FLAGS_DEBUG": "-g -fPIC -Wall -pedantic -Wextra -Werror -Wno-missing-braces -Wno-ignored-qualifiers --std=c++20 -fdiagnostics-color -Wno-stringop-overread -DDEBUG",
         "CMAKE_INSTALL_PREFIX": "${sourceDir}/install/gcc/debug/sparkd"
       }

src/lib/api/md/GlbxMarketDataConfig.proto

@@ -6,17 +6,19 @@ package bts.api.md;
 
 option csharp_namespace = "bts.api.md";
 
+message GlbxActivePlatform {
+  string host = 1;
+  string user = 2;
+  string password = 3;
+}
+message GlbxOnixPlatform {
+  string temp =1; // will add everything related to onix here.
+}
 
 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;
+  bts.api.domain.SecurityCriteria security_criteria = 2;
+  oneof Platform {
+    GlbxActivePlatform glbx_active_platform = 10;
+    GlbxOnixPlatform glbx_onix_platform = 11;
+  }
 }

src/lib/cmake/GetLinuxServerLinkLibraries.cmake

@@ -25,8 +25,23 @@ function(get_link_libraries dest)
     Packet++
     Common++
     ssh2
+    ACE-5.7.5
+    ActivBase_x86-64_ubuntu22-04_gcc123_mdid
+    ActivOneApi_x86-64_ubuntu22-04_gcc123_mdid
+    ActivAgentApplication_x86-64_ubuntu22-04_gcc123_mdid
+    ActivDatabase_x86-64_ubuntu22-04_gcc123_mdid
+    ActivFramework_x86-64_ubuntu22-04_gcc123_mdid
+    ActivServices_x86-64_ubuntu22-04_gcc123_mdid
+    BoostAtomic_x86-64_ubuntu22-04_gcc123_mdid
+    BoostChrono_x86-64_ubuntu22-04_gcc123_mdid
+    BoostProgramOptions_x86-64_ubuntu22-04_gcc123_mdid
+    BoostFilesystem_x86-64_ubuntu22-04_gcc123_mdid
+    -lm
+    -lgcc_s
+    -lc
     -lrt
     -ldl
+    -lstdc++
     -lpcap
     -static-libgcc
     -static-libstdc++

src/lib/core/TimeStamp.cpp

@@ -23,7 +23,18 @@ TimeStamp::TimeStamp(uint32_t year, uint32_t month, uint32_t day) {
    value_ =
        std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count();
 }
+bool TimeStamp::from_ymd(
+    std::string yearMonthDay, TimeStamp *result) {
+   uint32_t year, month, day;
+   std::istringstream ss(yearMonthDay);
+    char dash; // to consume the dash character in the input
 
+    // Use stringstream to extract year, month, and day from the input string
+    ss >> year >> dash >> month >> dash >> day;
+    if(ss.fail())
+      return false;
+    return TimeStamp::from_ymd(year,month,day,result);
+}
 bool TimeStamp::from_ymd(
     std::uint32_t year, std::uint32_t month, std::uint32_t day, TimeStamp *result) {
    if (!is_valid_ymd(year, month, day)) {

src/lib/core/TimeStamp.h

@@ -41,6 +41,9 @@ struct TimeStamp {
        std::uint32_t month,
        std::uint32_t day,
        TimeStamp *result);
+   static bool from_ymd(
+      std::string yearMonthDay,
+       TimeStamp *result);
 
    constexpr std::uint64_t nanos() const;
    constexpr std::uint64_t micros() const;

src/lib/orderman/actors/MDSource.h

@@ -5,7 +5,7 @@
 #include "lib/md/IntPrice.h"
 #include "lib/md/MDConsumer.h"
 #include "lib/md/MDConsumerHandle.h"
-#include "lib/md/cme/MDP3Utils.h"
+#include "server/sparkd/md/cme/MDP3Utils.h"
 #include "lib/mercury/actors/Actor.h"
 #include "lib/mercury/actors/IPollableActor.h"
 #include "lib/mercury/events/TimerEvent.h"

src/lib/orderman/ccapi/CCAPIDeribitOrderMux.cpp

@@ -1,3 +1,5 @@
+#ifdef SPARK_ENABLE_CCAPI
+
 #include "CCAPIDeribitOrderMux.h"
 #include "CCAPIOrderState.h"
 #include "CCAPISingleOrderManager.h"
@@ -953,4 +955,6 @@ void CCAPIDeribitOrderMux::_send_buffered_orders(uint32_t sec_id) {
        "n_buffered_orders: {}",
        me->buffered_creates.size());
 }
-} // namespace bts::orderman
+} // namespace bts::orderman
+
+#endif

src/lib/orderman/ccapi/CCAPIDeribitOrderMux.h

@@ -1,4 +1,6 @@
 #pragma once
+#ifdef SPARK_ENABLE_CCAPI
+
 #include "ICCAPIOrderMux.h"
 #include "lib/md/SecurityOracle.h"
 #include "lib/orderman/CreateOrderMsg.h"
@@ -72,4 +74,6 @@ class CCAPIDeribitOrderMux : public ICCAPIOrderMux {
    void _handle_generic_pvt_request(const nlohmann::json &json_obj);
 };
 
-} // namespace bts::orderman
+} // namespace bts::orderman
+
+#endif

src/lib/orderman/gateways/CCAPIExecutionGateway.cpp

@@ -1,3 +1,5 @@
+#ifdef SPARK_ENABLE_CCAPI
+
 #include "CCAPIExecutionGateway.h"
 #include "lib/api/service_helpers/AdminServiceUtils.h"
 #include "lib/api/services/AdminService.pb.h"
@@ -492,4 +494,6 @@ SingleOrderMux *CCAPIExecutionGateway::_find_mux_from_orderkey(OrderKey key) {
    return _find_mux(it->second.routing_record);
 }
 
-} // namespace bts::orderman
+} // namespace bts::orderman
+
+#endif

src/lib/orderman/gateways/CCAPIExecutionGateway.h

@@ -1,4 +1,5 @@
 #pragma once
+#ifdef SPARK_ENABLE_CCAPI
 
 #include "lib/orderman/gateways/IExecutionGateway.h"
 #include "lib/api/services/AdminService.pb.h"
@@ -73,3 +74,5 @@ class CCAPIExecutionGateway : public IExecutionGateway {
 
 } // namespace orderman
 } // namespace bts
+
+#endif

src/lib/orderman/gateways/ExecutionGatewayFactory.cpp

@@ -34,9 +34,11 @@ std::unique_ptr<IExecutionGateway> ExecutionGatewayFactory::create_gateway(
          return std::make_unique<ilink3::GLBXiLink3ExecutionGateway>(
              conn, pl, o);
       return std::make_unique<GLBXExecutionGateway>(conn, pl, o);
-   } else if (conn.venue() == bts::api::VENUE_DERIBIT)
-      //   return std::make_unique<EMExecutionGateway>(o);
+   } else if (conn.venue() == bts::api::VENUE_DERIBIT) {
+#ifdef SPARK_ENABLE_CCAPI
       return std::make_unique<CCAPIExecutionGateway>(conn.venue(), o);
+#endif
+   }
    return std::make_unique<FakeExecutionGateway>(o);
 }
 

src/lib/orderman/gateways/GLBXiLink3ExecutionGatewayState.h

@@ -1,5 +1,4 @@
 #pragma once
-#ifdef SPARK_ENABLE_CCAPI
 
 #include "lib/core/EnumMacros.h"
 #include "lib/core/FMT_IMPL.h"
@@ -26,7 +25,6 @@ ENUM_CLASS_DEF(
 #undef FOR_EACH_GLBX_ILINK3_EXECUTION_GATEWAY_STATE
 
 } // namespace bts::orderman::ilink3
-#endif
 
 FMT_IMPL(
     bts::orderman::ilink3::GLBXiLink3ExecutionGatewayState, s, "{}", s.name());

src/lib/orderman/ilink3/IL3State.h

@@ -1,5 +1,4 @@
 #pragma once
-#ifdef SPARK_ENABLE_CCAPI
 
 #include "lib/core/EnumMacros.h"
 #include "lib/core/FMT_IMPL.h"
@@ -27,6 +26,5 @@ ENUM_CLASS_DEF(IL3State, FOR_EACH_IL3_STATE)
 #undef FOR_EACH_IL3_STATE
 
 } // namespace bts::orderman::ilink3
-#endif
 
 FMT_IMPL(bts::orderman::ilink3::IL3State, s, "{}", s.name());

src/sdk/examples/md_publisher/md_publisher_example.cpp

@@ -1,6 +1,6 @@
 #include "spark/lib/logging/logging.h"
 #include "spark/lib/md/MDProducer.h"
-#include "spark/lib/md/cme/MDP3Utils.h"
+#include "server/sparkd/md/cme/MDP3Utils.h"
 #include "spark/lib/tracing/bts_trace.h"
 #include "spark/lib/utils/PBUtils.h"
 #include "spark/lib/utils/bts_chrono.h"

src/server/sparkd/cmds/md/cmds/Channels.cpp

@@ -1,5 +1,5 @@
 #include "Channels.h"
-#include "lib/md/cme/MDP3Utils.h"
+#include "server/sparkd/md/cme/MDP3Utils.h"
 
 namespace bts::cmds::md::cmds {
 

src/server/sparkd/main.cpp

@@ -1,3 +1,4 @@
+#include "server/sparkd/md/cme/CMEActivSession.h"
 #include "lib/testing/catch_test_runner.h"
 #include "lib/utils/init_spark_lib.h"
 #include "server/sparkd/cmds/AdminReport.h"
@@ -10,6 +11,9 @@
 #include "server/sparkd/cmds/crypto/CryptoCmd.h"
 #include <CLI11/CLI11.hpp>
 
+#include <iostream>
+#include <string>
+
 using namespace bts;
 
 int main(int argc, char **argv) {
@@ -30,6 +34,9 @@ int main(int argc, char **argv) {
       app.add_subcommand(CLI::App_p(new cmds::Run()));
       app.add_subcommand(CLI::App_p(new cmds::KeyCmd()));
       app.add_subcommand(CLI::App_p(new cmds::Web()));
+
+     
+
       app.add_subcommand(
           "test", "Run unit test(s)"); // so this shows up in help
       CLI11_PARSE(app, argc, argv);

src/server/sparkd/md/MarketDataGateway.cpp

@@ -32,6 +32,7 @@
 #include "server/sparkd/md/okx/OkxSubscription.h"
 #include "server/sparkd/md/platforms/mdp3_onixs/MDP3Main.h"
 #include "server/sparkd/md/recording/MDReplayer.h"
+#include "server/sparkd/md/cme/CMEActivSession.h"
 
 #include <bits/stdint-uintn.h>
 #include <boost/filesystem.hpp>
@@ -126,15 +127,19 @@ void MarketDataGateway::run() {
    }
    LOG(md, Info, "MarketDataGateway::run starting t_id:{} t_name:{}",
        os::get_thread_id(), os::get_thread_name());
+
    _start_cc(me_->md_config);
+   
    if (!me_->md_config.emumba_deribit().empty()) {
       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_glbx(me_->md_config.glbx().at(0));
    }
+
    _start_replays(me_->md_config);
    while (true) {
       std::this_thread::sleep_for(std::chrono::seconds(1));
@@ -230,7 +235,26 @@ void MarketDataGateway::unsubscribe(MarketDataListenerID id) {
    LOG(md, Info, "Unsubscribing listener {}", id);
    me_->listeners.erase(id);
 }
-/// glbx //////////////////////////////////////////////////////////////////////
+/// onix active //////////////////////////////////////////////////////////////////////
+//////////////////////////////////////////////////////////////////////////////
+static void run_active(bts::api::md::GlbxMarketDataConfig cfg, persistence::IDGenerator *sec_id_store) {
+   LOG(md, Info, "MarketDataGateway::run_active t_id:{} t_name", os::get_thread_id(),
+       os::get_thread_name());
+      bts::md::GeneralSecurityIndex sec_index;
+      bts::md::MDProducer md_producer;
+      std::shared_ptr<GeneralMarketDataListener> listener =std::make_shared<bts::md::GeneralMarketDataListener>(sec_index,md_producer,bts::api::VENUE_GLBX);
+      bts::md::cme::CMEActivSession::run(cfg.glbx_active_platform().host(), cfg.glbx_active_platform().user(), cfg.glbx_active_platform().password(), listener, sec_id_store);
+   // Timers
+//    auto &disp = utils::Dispatcher::current();
+//    utils::DispatcherTimer heartbeat_timer(disp, 250, [&]() {
+//       for (auto &p : platforms)
+//          p->send_heartbeats();
+//    });
+//    heartbeat_timer.start();
+
+//    disp.run();
+}
+/// onix glbx //////////////////////////////////////////////////////////////////////
 //////////////////////////////////////////////////////////////////////////////
 static void run_onixs() {
    LOG(md, Info, "MarketDataGateway::run_onixs t_id:{} t_name",
@@ -259,8 +283,29 @@ static void run_onixs() {
    disp.run();
 }
 
-void MarketDataGateway::_start_onix_glbx(
-    bts::api::md::GlbxMarketDataConfig cfg) {
+void MarketDataGateway::_start_active_glbx(bts::api::md::GlbxMarketDataConfig cfg){
+   LOG(md, Info,
+       "MarketDataGateway::_start_active_glbx starting t_id:{} t_name:{}. "
+       "cfg:\n{}",
+       os::get_thread_id(), os::get_thread_name(), cfg.DebugString());
+LOG(md, Info,
+       "MarketDataGateway::_start_active_glbx");
+   bts::md::run_active(cfg, &me_->sec_id_store);
+}
+void MarketDataGateway::_start_glbx( bts::api::md::GlbxMarketDataConfig cfg){
+   LOG(md, Info,
+       "MarketDataGateway::_start_glbx starting t_id:{} t_name:{}. "
+       "cfg:\n{}",
+       os::get_thread_id(), os::get_thread_name(), cfg.DebugString());
+       if(cfg.has_glbx_active_platform()){
+            _start_active_glbx(cfg);
+       }
+       else if(cfg.has_glbx_onix_platform()){
+            _start_onix_glbx(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{}",

src/server/sparkd/md/MarketDataGateway.h

@@ -35,6 +35,10 @@ class MarketDataGateway {
        bts::api::md::EmumbaDeribitMarketDataConfig cfg);
    static void _start_onix_glbx(
        bts::api::md::GlbxMarketDataConfig cfg);
+   void _start_glbx(
+       bts::api::md::GlbxMarketDataConfig cfg);
+   void _start_active_glbx(
+       bts::api::md::GlbxMarketDataConfig cfg);
    void _start_replays(const bts::api::md::MarketDataConfig &cfg);
 };
 

src/server/sparkd/md/cme/CMEActivSession.cpp

@@ -0,0 +1,91 @@
+#include "CMEActivSession.h"
+#include "lib/logging/logging.h"
+#include "lib/md/GeneralMarketDataListener.h"
+#include "server/sparkd/md/cme/CMEActivSessionHandler.h"
+#include "server/sparkd/md/cme/CMEActivSubscriptionHandler.h"
+#include "server/sparkd/md/cme/CMEActiveQueryHandler.h"
+#include "server/sparkd/md/cme/CMEActiveSnapshotHandler.h"
+#include <ActivMiddleware/ActivBase/Exception.h>
+#include <ActivMiddleware/ActivBase/FieldTypes/ActivFieldTypes.h>
+#include <ActivMiddleware/ActivBase/FieldTypes/FieldMap.h>
+#include <ActivOneApi/Constants/FieldIds.h>
+#include <ActivOneApi/Session.h>
+#include <exception>
+#include <ActivOneApi/Constants/DataSourceIds.h>
+
+namespace bts::md::cme {
+
+using Activ::FieldMap;
+using Activ::OneApi::Session;
+using Activ::OneApi::Handle;
+using std::string;
+void CMEActivSession::run(const string &host, const string &user, const string &password,std::shared_ptr<GeneralMarketDataListener> listner, persistence::IDGenerator *sec_id_store) {
+   try {
+      // Session callback handler.
+      CMEActivSessionHandler sessionHandler;
+
+      FieldMap fields;
+      // Construct the Session with a default configuration passing in our sessionHandler.
+      Session session(fields, sessionHandler);
+
+      FieldMap connectParameters;
+      
+      connectParameters.Set(Activ::OneApi::FID_HOST, host.c_str());
+      connectParameters.Set(Activ::OneApi::FID_USER_ID, user.c_str());
+      connectParameters.Set(Activ::OneApi::FID_PASSWORD, password.c_str());
+
+      // Connect synchronously with a 10 second timeout (will throw on failure).
+      session.Connect(connectParameters, 10000);
+      std::cerr << "connected" << std::endl;
+   
+      CMEActivQueryHandler query_handler(listner, sec_id_store);
+      // Specifying a data source and symbol.
+      // Activ::OneApi::Handle handle = session.Subscribe(Activ::OneApi::DATA_SOURCE_ACTIV,"CSIQ.Q",subscriptionHandler);
+      
+      // Activ::OneApi::Handle handle = session.Query(DATA_SOURCE_ACTIV,
+      //             "type=option and country=US and symbol=KC*",
+      //             subscriptionHandler);
+
+      md::cme::SnapshotHandler snapshotHandler;
+
+		// Request snapshot of the Microsoft US Equity topic.
+		// Returned Handle for a Snapshot is only required to cancel the request prior to a response.
+		// (void) session.Snapshot(DATA_SOURCE_ACTIV, "type=option and country=US and symbol=CSIQ/*", snapshotHandler);
+
+      // (void) session.Query(DATA_SOURCE_ACTIV, "symbol=CSIQ/*", query_handler);
+      (void) session.Query(DATA_SOURCE_ACTIV, "type=future and symbol=NK/*", query_handler);
+      // QueryResults queryResults;
+      // std::string query;
+      // while(true)
+      // {
+      //    query ="";
+      //    std::getline(std::cin, query);
+      //    if(query.substr(0,3) == "end")
+      //       break;
+      //    session.Query(DATA_SOURCE_ACTIV,
+      //                query,
+      //                queryResults);
+
+      //       // Prin  t out the topics.
+      //       for (const auto &message : queryResults)
+      //       {
+      //          std::cout << "Found symbol " << message.GetSymbol() << std::endl;
+      //       }
+      //    std::cout<<"END "<<std::endl;
+      // }
+      // Alternatively an asynchronous connection can be used.
+      // Connection completion is notified via ISessionHandler::OnSessionConnect().
+      // session.AsyncConnect(connectParameters);
+      session.Run();
+
+   } catch (const Activ::Exception &ex) {
+      std::cerr << "Error running CMEActivSession (activ): " << ex.GetMessage()
+                << ex.ToString()<< std::endl;
+   } catch (const std::exception &ex) {
+      std::cerr << "Error running CMEActivSession: " << ex.what() << std::endl;
+      // LOG(md, Error, "Error running CMEActivSession: {}",
+      //     std::string{ex.what()});
+   }
+}
+
+} // namespace bts::md::cme

src/server/sparkd/md/cme/CMEActivSession.h

@@ -0,0 +1,19 @@
+#pragma once
+
+#include "lib/md/GeneralMarketDataListener.h"
+#include "lib/persistence/IDGenerator.h"
+#include <string>
+
+namespace bts::md::cme {
+
+class CMEActivSession {
+ public:
+   static void run(
+       const std::string &host,
+       const std::string &user,
+       const std::string &password,
+       std::shared_ptr<GeneralMarketDataListener> listner,
+      persistence::IDGenerator *sec_id_store);
+};
+
+} // namespace bts::md::cme

src/server/sparkd/md/cme/CMEActivSessionHandler.cpp

@@ -0,0 +1,33 @@
+#include "CMEActivSessionHandler.h"
+#include <ActivMiddleware/ActivBase/ActivBase.h>
+#include <ActivOneApi/Session.h>
+#include "CMEActivSubscriptionHandler.h"
+
+namespace bts::md::cme {
+
+using Activ::LogType;
+using Activ::StatusCode;
+using Activ::String;
+using Activ::OneApi::Session;
+
+void CMEActivSessionHandler::OnSessionLogMessage(
+    Session & /*session*/, const LogType log_type, const String &message) {
+   std::cout << Activ::LogTypeToString(log_type) << ": " << message
+             << std::endl;
+}
+
+void CMEActivSessionHandler::OnSessionConnect(Session & /*session*/) {
+   std::cerr << "connected" << std::endl;
+}
+
+void CMEActivSessionHandler::OnSessionDisconnect(Session & /*session*/) {
+   std::cout << "disconnected" << std::endl;
+}
+
+void CMEActivSessionHandler::OnSessionError(
+    Session & /*session*/, const StatusCode statusCode) {
+   std::string err{Activ::StatusCodeToString(statusCode)};
+   std::cerr << "CMEActivSessionHandler::OnSessionError. " << err << std::endl;
+}
+
+} // namespace bts::md::cme

src/server/sparkd/md/cme/CMEActivSessionHandler.h

@@ -0,0 +1,25 @@
+#pragma once
+
+
+#include <ActivMiddleware/ActivBase/ActivBase.h>
+#include <ActivOneApi/ISessionHandler.h>
+
+namespace bts::md::cme {
+
+class CMEActivSessionHandler : public Activ::OneApi::ISessionHandler {
+ public:
+   static void run();
+
+   void OnSessionLogMessage(
+       Activ::OneApi::Session &session,
+       const Activ::LogType log_type,
+       const Activ::String &message) override;
+
+   void OnSessionConnect(Activ::OneApi::Session &session) override;
+   void OnSessionDisconnect(Activ::OneApi::Session &session) override;
+   void OnSessionError(
+       Activ::OneApi::Session &session,
+       const Activ::StatusCode statusCode) override;
+};
+
+} // namespace bts::md::cme

src/server/sparkd/md/cme/CMEActivSubscriptionHandler.h

@@ -0,0 +1,365 @@
+#pragma once
+#include "lib/core/TimeStamp.h"
+#include "lib/domain/AggressorSide.h"
+#include "lib/domain/Book.h"
+#include "lib/domain/BookLevel.h"
+#include "lib/domain/FutureSecurity.h"
+#include "lib/domain/Quantity.h"
+#include "lib/domain/SecurityID.h"
+#include "lib/domain/Side.h"
+#include "lib/domain/TradeRecord.h"
+#include "lib/md/GeneralMarketDataListener.h"
+#include "server/sparkd/md/MarketDataGateway.h"
+#include "lib/orderman/orderman_utils/Book.h"
+#include "lib/persistence/IDGenerator.h"
+#include "lib/utils/bts_assert.h"
+#include <ActivOneApi/MarketData/EventTypes.h>
+#include <ActivOneApi/MarketData/FieldIds.h>
+#include <ActivOneApi/MarketData/Separators.h>
+#include <cassert>
+#include <iostream>
+#include <string>
+#define UNUSED(x) (void)(x)
+#include "ActivOneApi/MarketData/MarketData.h"
+#include <ActivMiddleware/ActivBase/ActivBase.h>
+
+#include <ActivOneApi/OneApi.h>
+#include <ActivOneApi/ISessionHandler.h>
+#include <ActivOneApi/ISubscriptionHandler.h>
+#include <ActivOneApi/Messages/RefreshMessage.h>
+#include <ActivOneApi/Messages/UpdateMessage.h>
+using namespace Activ;
+using namespace Activ::OneApi;
+namespace bts::md::cme {
+class CMEActivSubscriptionHandler : public Activ::OneApi::ISubscriptionHandler {
+
+   FieldListValidator m_fieldListValidator;
+
+   Activ::OneApi::TopicSubscriptionState *m_state;
+
+ public:
+   std::shared_ptr<GeneralMarketDataListener> listner;
+   persistence::IDGenerator *sec_id_store;
+
+   void init(
+       std::shared_ptr<GeneralMarketDataListener> listner,
+       persistence::IDGenerator *id_store,
+       Activ::OneApi::TopicSubscriptionState *m_state) {
+      this->listner = listner;
+      sec_id_store = id_store;
+      this->m_state = m_state;
+   }
+   CMEActivSubscriptionHandler() {}
+   CMEActivSubscriptionHandler(
+       std::shared_ptr<GeneralMarketDataListener> listner,
+       persistence::IDGenerator *sec_id_store,
+       Activ::OneApi::TopicSubscriptionState *m_state)
+       : m_state(m_state), listner(listner), sec_id_store(sec_id_store) {}
+
+   // ---------------------------------------------------------------------------------------------------------------------------------
+   std::string removeCharacters(const std::string &input) {
+      std::string result;
+
+      for (char c : input) {
+         if (c != '/' && c != '.') {
+            result += c;
+         } else if (c == '.') {
+            break; // Stop adding characters after encountering a dot
+         }
+      }
+
+      return result;
+   }
+   void OnSubscriptionRefresh(
+       const RefreshMessage &refreshMessage, const Context &context) {
+      UNUSED(context);
+      std::cout << "REFRESH received for " << refreshMessage.m_symbol
+                << std::endl;
+      std::cout << "Symbol .................... " << refreshMessage.m_symbol
+                << std::endl;
+      std::cout << "Topic subscription state .. "
+                << TopicSubscriptionStateToString(
+                       refreshMessage.m_topicSubscriptionState)
+                << std::endl;
+      StatusCode statusCode = m_fieldListValidator.Initialize(refreshMessage);
+
+      for (const Activ::OneApi::FlvDetail::Field &field :
+           m_fieldListValidator) {
+            if(field.IsDefined())
+         std::cout<<"onsub Feild: "<<MarketData::GetFieldName(field.GetId())<<", Value: "<<field.GetValue<std::string>() <<std::endl;
+      }
+
+      if (STATUS_CODE_SUCCESS != statusCode) {
+         std::cout << "ERROR decoding field data for "
+                   << refreshMessage.m_symbol << " - "
+                   << StatusCodeToString(statusCode) << std::endl;
+         return;
+      }
+      const Activ::OneApi::FlvDetail::Field *pField;
+      statusCode = m_fieldListValidator.GetField(
+          MarketData::FID_EXPIRATION_DATE, pField);
+
+      if (STATUS_CODE_SUCCESS != statusCode) {
+         std::cout << "ERROR fetching "
+                   << "exp date"
+                   << " field for " << refreshMessage.m_symbol << " - "
+                   << StatusCodeToString(statusCode) << std::endl;
+         return;
+      }
+      std::string exp_date = pField->GetValue<String>();
+      core::TimeStamp exp_time;
+      if (!core::TimeStamp::from_ymd(exp_date, &exp_time)) {
+         return;
+      }
+
+      String result = removeCharacters(refreshMessage.m_symbol);
+      domain::SecurityID sec_id{
+          sec_id_store->get_or_add(api::domain::VENUE_GLBX, result)};
+      std::vector<domain::Security> complete_security_list;
+      domain::Security &sec = complete_security_list.emplace_back();
+      sec.security_common.venue = api::domain::VENUE_GLBX;
+      sec.security_common.id = sec_id;
+      sec.security_common.exchange_sec_id =
+          refreshMessage.ToString(); // Need to fix this
+      sec.security_common.desc = result;
+      sec.security_common.symbol = result;
+      sec.security_common.asset = "a1";
+      sec.security_common.group = result;
+
+      sec.security_common.min_price_increment = 0.1;
+      sec.security_common.min_quantity = 1;
+      sec.security_common.min_quantity_increment = 0.1;
+      sec.security_common.currency = "name";
+      auto &t = sec.security.emplace<domain::FutureSecurity>();
+      t.expiration = exp_time;
+      std::vector<const domain::Security *> const_security_list;
+      for (const auto &security : complete_security_list) {
+         const_security_list.push_back(&security);
+      }
+      listner->on_securities(const_security_list);
+
+      
+      // Alternatively use random access.
+
+      // Print out the current value for FID_TRADE. FID_TRADE is defined in the Activ::MarketData namespace.
+      // ProcessField(
+      //     MarketData::FID_EXPIRATION_DATE, "Exp ", refreshMessage.m_symbol);
+   }
+   template <typename T>
+   bool _get_field(String field_name, Activ::OneApi::FieldId fieldId, T &res ){
+      const Activ::OneApi::FlvDetail::Field *pField;
+      auto statusCode =
+          m_fieldListValidator.GetField(fieldId, pField);
+      if (STATUS_CODE_SUCCESS != statusCode) {
+         std::cout << "ERROR fetching "
+                   << field_name
+                   << " field Error: "
+                   << StatusCodeToString(statusCode) << std::endl;
+         return false;
+      }
+      res = pField->GetValue<T>();
+      return true;
+   }
+   bool _parse_trade(std::vector<const domain::TradeRecord *> &trades, domain::SecurityID id) {
+      UNUSED(trades);
+      for (const Activ::OneApi::FlvDetail::Field &field :
+           m_fieldListValidator) {
+         if (field.IsDefined()) {
+
+            std::cout << "Has Feild: "
+                      << MarketData::GetFieldName(field.GetId())
+                      << ", Value: " << field.GetValue<std::string>()
+                      << std::endl;
+         }
+      }
+      std::vector<domain::TradeRecord> parsed_trades;
+      auto &t = parsed_trades.emplace_back();
+      double tp;
+      if(!_get_field<double>("FID_TRADE", MarketData::FID_TRADE, tp))
+         return false;
+      uint ts;
+      if(!_get_field<uint>("FID_TRADE_SIZE", MarketData::FID_TRADE_SIZE, ts))
+         return false;
+      String time, date;
+      if(!_get_field<String>("FID_TRADE_TIME", MarketData::FID_TRADE_TIME, time))
+         return false;
+      if(!_get_field<String>("FID_TRADE_DATE", MarketData::FID_TRADE_DATE, date))
+         return false;
+      
+      t.trade_price=bts::domain::Price::from_double(tp);
+      t.trade_quantity=bts::domain::Quantity(ts);
+      t.sec_id=id;
+      t.agg_side=bts::domain::AggressorSide::Buy;
+      t.time_stamp=core::TimeStamp::from_iso_string(date+"T"+time);;
+      trades.push_back(&parsed_trades[0]);
+      bts_assert(trades[0]->sec_id==id);
+      listner->on_trades(trades);
+      return true;
+   }
+   // void _parse_side()
+   bool _parse_book(domain::Book &result, size_t n_bid_levels = 5) {
+      UNUSED(n_bid_levels);
+      uint ask_size, bid_size;
+      if(!_get_field<uint>("FID_ASK_SIZE", MarketData::FID_ASK_SIZE, ask_size))
+         return false;
+      if(!_get_field<uint>("FID_BID_SIZE", MarketData::FID_BID_SIZE, bid_size))
+         return false;
+      double bid_val,ask_val;
+      if(!_get_field<double>("FID_BID", MarketData::FID_BID, bid_val))
+         return false;
+      if(!_get_field<double>("FID_ASK", MarketData::FID_ASK, ask_val))
+         return false;
+      domain::BookSideT<10> &b = result.bids();
+      domain::BookSideT<10> &a = result.asks();
+      domain::BookLevel b_l;
+      b_l.price=domain::Price::from_double(bid_val);
+      b_l.total_quantity=domain::Quantity(bid_size);
+      domain::BookLevel a_l;
+      a_l.price=domain::Price::from_double(ask_val);
+      a_l.total_quantity=domain::Quantity(ask_size);
+      b[0]=b_l;
+      a[0]=a_l;
+
+      return true;
+   }
+   // ---------------------------------------------------------------------------------------------------------------------------------
+   void ProcessField(
+       const FieldId fieldId, const String &fieldName, const String &symbol) {
+      const FieldListValidator::Field *pField;
+
+      // FieldListValidator::GetField() provides access to the internally stored cached field.
+      StatusCode statusCode = m_fieldListValidator.GetField(fieldId, pField);
+      if (STATUS_CODE_SUCCESS != statusCode) {
+         std::cout << "ERROR fetching " << fieldName << " field for " << symbol
+                   << " - " << StatusCodeToString(statusCode) << std::endl;
+         return;
+      }
+
+      std::cout << fmt::format("{:.<27s} ", fieldName + " ");
+
+      if (!pField->IsDefined()) {
+         std::cout << "undefined" << std::endl;
+      } else {
+         // Field contents are only valid if the Field is defined.
+         std::cout << pField->GetValue<String>() << std::endl;
+      }
+   }
+   void OnSubscriptionUpdate(
+       const UpdateMessage &updateMessage, const Context &context) {
+      UNUSED(context);
+      m_fieldListValidator.ApplyUpdate(updateMessage);
+      String result = removeCharacters(updateMessage.m_symbol);
+      auto id = sec_id_store->get_id(api::domain::VENUE_GLBX, result);
+      assert(id.has_value());
+
+      for (const Activ::OneApi::FlvDetail::Field &field :
+           m_fieldListValidator) {
+            if(field.IsDefined())
+         std::cout<<"Update Feild: "<<MarketData::GetFieldName(field.GetId())<<", Value: "<<field.GetValue<std::string>() <<std::endl;
+      }
+      switch (updateMessage.m_eventType) {
+      case MarketData::EVENT_TYPE_BBO_QUOTE:
+      case MarketData::EVENT_TYPE_BBO_DEPTH: {
+         domain::Book book;
+         book.set_security_id(domain::SecurityID(id.value()));
+         if (_parse_book(book)) {
+            listner->on_book(book);
+            std::cout << "Book UPDATE received for " << updateMessage.m_symbol
+                      << std::endl;
+         }
+         break;
+      }
+      case MarketData::EVENT_TYPE_TRADE:
+      {
+         std::vector<const domain::TradeRecord *> trades;
+         if(_parse_trade(trades,domain::SecurityID(id.value()))){
+            
+            std::cout << "trade UPDATE received for " << updateMessage.m_symbol
+                      << std::endl;
+         }
+         break;
+      }
+      default:
+         std::cout << "case not defined \n";
+      }
+   }
+
+   // ---------------------------------------------------------------------------------------------------------------------------------
+
+   void OnSubscriptionTopicStatus(
+       const TopicStatusMessage &topicStatusMessage, const Context &context) {
+      UNUSED(context);
+      std::cout << "TOPIC STATUS received for " << topicStatusMessage.m_symbol
+                << " "
+                << TopicSubscriptionStateToString(
+                       topicStatusMessage.m_topicSubscriptionState)
+                << std::endl;
+
+      *m_state = topicStatusMessage.m_topicSubscriptionState;
+   }
+
+   // ---------------------------------------------------------------------------------------------------------------------------------
+
+   void OnSubscriptionError(
+       const StatusCode statusCode, const Context &context) {
+      UNUSED(context);
+      std::cout << "Subscription error " << StatusCodeToString(statusCode)
+                << std::endl;
+   }
+
+   // ---------------------------------------------------------------------------------------------------------------------------------
+
+   void OnSubscriptionFailure(
+       const StatusCode statusCode, const Context &context) {
+      UNUSED(context);
+      std::cout << "Subscription failure " << StatusCodeToString(statusCode)
+                << std::endl;
+   }
+   // void OnSubscriptionUpdate(
+   //     const Activ::OneApi::UpdateMessage &updateMessage,
+   //     const Activ::OneApi::Context &context) {
+   //    UNUSED(context);
+   //    std::cout << "UPDATE: " << updateMessage.m_symbol
+   //              << updateMessage.ToString() << std::endl;
+
+   //    // ApplyUpdate is used here, unlike Initialize, ApplyUpdate does not clear existing
+   //    // field content.
+   //    // m_fieldListValidator.ApplyUpdate(updateMessage);
+
+   //    // ... use m_fieldListValidator to access content.
+   // }
+   // void OnSubscriptionRefresh(
+   //     const Activ::OneApi::RefreshMessage &refreshMessage,
+   //     const Activ::OneApi::Context &context) {
+   //    UNUSED(context);
+   //    std::cout << "REFRESH: " << refreshMessage.m_symbol << std::endl;
+
+   //    // Access to field data is obtained using a FieldListValidator.
+   //    // m_fieldListValidator.Initialize(refreshMessage);
+
+   //    // ... use m_fieldListValidator to access content.
+   // }
+
+   // void OnSubscriptionTopicStatus(
+   //     const TopicStatusMessage &topicStatusMessage, const Context &context) {
+   //    UNUSED(context);
+   //    // If topicStatusMessage.m_topicSubscriptionState is not TOPIC_SUBSCRIPTION_STATE_OK
+   //    // then any cached fields may be stale.
+   //    std::cout << "TOPIC STATUS: " << topicStatusMessage.m_symbol
+   //              << ", state=" << topicStatusMessage.ToString() << std::endl;
+   // }
+   // void OnSubscriptionError(
+   //     const StatusCode statusCode, const Context &context) {
+   //    std::cout << "error :" << StatusCodeToString(statusCode) << std::endl;
+   //    UNUSED(statusCode);
+   //    UNUSED(context);
+   // }
+   // void OnSubscriptionFailure(
+   //     const StatusCode statusCode, const Context &context) {
+   //    std::cout << "Failure :" << StatusCodeToString(statusCode) << std::endl;
+   //    UNUSED(statusCode);
+   //    UNUSED(context);
+   // }
+};
+
+} // namespace bts::md::cme

src/server/sparkd/md/cme/CMEActiveQueryHandler.h

@@ -0,0 +1,108 @@
+#pragma once
+#include "server/sparkd/md/cme/CMEActivSubscriptionHandler.h"
+#include "lib/md/GeneralMarketDataListener.h"
+#include <ActivMiddleware/CxxYZ/unordered_map.h>
+#define UNUSED(x) (void)(x)
+#include "ActivOneApi/OneApi.h"
+namespace bts::md::cme {
+class CMEActivQueryHandler : public Activ::OneApi::ITopicQueryHandler {
+   class TopicDetails {
+    public:
+      //! The subscription handle for this topic
+      Activ::OneApi::ScopedHandle m_handle;
+      CMEActivSubscriptionHandler sub_handler;
+      //! The current topic subscription state. Is this topic waiting for a refresh.
+      Activ::OneApi::TopicSubscriptionState m_state;
+
+      /**
+		 *	@brief	Has this topic been seen for the current topic query. Whilst a query is in an error state there may be missed
+		 *			topic deletions, these need to be cleaned up once the OnTopicQueryComplete callback is received.
+		 */
+      size_t m_queryVersion;
+   };
+   //! Used to track how many times the query has restarted.
+   // size_t m_queryVersion;
+   //! Map of subscriptions.
+   typedef Activ::CxxYZ::unordered_map<Activ::String, TopicDetails> SubscriptionMap;
+   SubscriptionMap m_subscriptionMap;
+   //! Used to track how many times the query has restarted.
+   size_t m_queryVersion;
+   public:
+   std::shared_ptr<GeneralMarketDataListener> listner;
+   persistence::IDGenerator *sec_id_store;
+   CMEActivQueryHandler(
+       std::shared_ptr<GeneralMarketDataListener> listner,
+       persistence::IDGenerator *sec_id_store)
+       : m_queryVersion(0), listner(listner), sec_id_store(sec_id_store) {}
+
+   // ---------------------------------------------------------------------------------------------------------------------------------
+
+   void OnTopicQueryStart(const Context &context) {
+      UNUSED(context);
+      std::cout << "QUERY start" << std::endl;
+      ++m_queryVersion;
+   }
+
+   // ---------------------------------------------------------------------------------------------------------------------------------
+
+   void OnTopicQueryAdd(Topic &topic, const Context &context) {
+      const String &symbol = topic.GetSymbol();
+      TopicDetails &topicDetails = m_subscriptionMap[symbol];
+
+      if (!topicDetails.m_handle) {
+         topicDetails.m_queryVersion = m_queryVersion;
+         topicDetails.m_state = TOPIC_SUBSCRIPTION_STATE_PENDING;
+         topicDetails.sub_handler = CMEActivSubscriptionHandler(listner, sec_id_store, &topicDetails.m_state);
+         topicDetails.m_handle =
+             context.GetSession().Subscribe(DATA_SOURCE_ACTIV, symbol, topicDetails.sub_handler);
+      } else {
+         topicDetails.m_queryVersion = m_queryVersion;
+      }
+   }
+   // ---------------------------------------------------------------------------------------------------------------------------------
+
+   void OnTopicQueryRemove(const TopicId &topicId, const Context &context) {
+      UNUSED(context);
+      std::cout << "QUERY topic removed: " << topicId.m_symbol << std::endl;
+
+      SubscriptionMap::iterator iter = m_subscriptionMap.find(topicId.m_symbol);
+      if (m_subscriptionMap.end() != iter)
+         m_subscriptionMap.erase(iter);
+   }
+
+   // ---------------------------------------------------------------------------------------------------------------------------------
+
+   void OnTopicQueryComplete(const Context &context) {
+      std::cout << "QUERY complete" << std::endl;
+      UNUSED(context);
+      // Close the subscriptions for any topics that haven't been received.
+      SubscriptionMap::iterator iter = m_subscriptionMap.begin();
+      while (m_subscriptionMap.end() != iter) {
+         if (iter->second.m_queryVersion != m_queryVersion)
+            m_subscriptionMap.erase(iter++);
+         else
+            ++iter;
+      }
+   }
+
+   // ---------------------------------------------------------------------------------------------------------------------------------
+
+   void OnTopicQueryError(const StatusCode statusCode, const Context &context) {
+      std::cout << "QUERY error " << StatusCodeToString(statusCode)
+                << std::endl;
+      UNUSED(context);
+   }
+
+   // ---------------------------------------------------------------------------------------------------------------------------------
+
+   void OnTopicQueryFailure(
+       const StatusCode statusCode, const Context &context) {
+      std::cout << "QUERY failure " << StatusCodeToString(statusCode)
+                << std::endl;
+
+      // Shutdown session.
+      Session::Shutdown(context.GetSession());
+      UNUSED(context);
+   }
+};
+} // namespace bts::md::cme

src/server/sparkd/md/cme/CMEActiveSnapshotHandler.h

@@ -0,0 +1,100 @@
+/**
+ *	@file	source/ActivOnePlatform/Samples/SnapshotFieldListValidatorSample/SnapshotHandler.h
+ *
+ *	@brief	SnapshotHandler definition.
+ */
+#pragma once
+#if (!defined ACTIV_SAMPLE_SNAPSHOT_HANDLER_H)
+#define ACTIV_SAMPLE_SNAPSHOT_HANDLER_H
+
+#include "ActivOneApi/OneApi.h"
+#include "ActivOneApi/MarketData/MarketData.h"
+
+
+#include <ostream>
+
+using namespace Activ;
+using namespace Activ::OneApi;
+namespace bts::md::cme {
+//! SnapshotHandler.
+class SnapshotHandler : public ISnapshotHandler
+{
+public:
+	virtual void OnSnapshot(const SnapshotMessage &snapshotMessage, const Context &context) ACTIV_OVERRIDE;
+	virtual void OnSnapshotFailure(const SnapshotFailureMessage &snapshotFailureMessage, const Context &context) ACTIV_OVERRIDE;
+
+private:
+	void ProcessField(const FieldId fieldId, const String &fieldName, const String &symbol);
+
+	// FieldListValidator will parse SnapshotMessage, if possible reuse FieldListValidator objects.
+	FieldListValidator	m_fieldListValidator;
+};
+
+// ---------------------------------------------------------------------------------------------------------------------------------
+
+inline void SnapshotHandler::OnSnapshot(const SnapshotMessage &snapshotMessage, const Context &context)
+{
+	std::cout << "SNAPSHOT received for " << snapshotMessage.m_symbol << std::endl;
+	std::cout << "Symbol .................... " << snapshotMessage.m_symbol << std::endl;
+
+	// Initialize will clear any existing cache fields. SnapshotMessage contains a complete topic field list if no filters have
+	// been requested.
+	StatusCode statusCode = m_fieldListValidator.Initialize(snapshotMessage);
+
+	if (STATUS_CODE_SUCCESS != statusCode)
+	{
+		std::cout << "ERROR decoding field data for " << snapshotMessage.m_symbol << " - " << StatusCodeToString(statusCode) << std::endl;
+		return;
+	}
+     for (const Activ::OneApi::FlvDetail::Field &field :
+           m_fieldListValidator) {
+            if(field.IsDefined())
+         std::cout<<"Has snapshot Feild: "<<MarketData::GetFieldName(field.GetId())<<", Value: "<<field.GetValue<std::string>() <<std::endl;
+      }
+
+	// Print out the current value for FID_TRADE. FID_TRADE is defined in the Activ::MarketData namespace.
+	ProcessField(MarketData::FID_TRADE, "FID_TRADE", snapshotMessage.m_symbol);
+
+	std::cout << std::endl;
+
+	context.GetSession().CloseHandle(context.GetHandle());
+}
+
+// ---------------------------------------------------------------------------------------------------------------------------------
+
+inline void SnapshotHandler::OnSnapshotFailure(const SnapshotFailureMessage &snapshotFailureMessage, const Context &context)
+{
+	std::cout << "SNAPSHOT FAILURE for " << snapshotFailureMessage.m_symbol << std::endl;
+	std::cout << snapshotFailureMessage.ToString() << std::endl;
+
+	context.GetSession().CloseHandle(context.GetHandle());
+}
+
+// ---------------------------------------------------------------------------------------------------------------------------------
+
+inline void SnapshotHandler::ProcessField(const FieldId fieldId, const String &fieldName, const String &symbol)
+{
+	const FieldListValidator::Field *pField;
+
+	// FieldListValidator::GetField() provides access to the internally stored cached field.
+	StatusCode statusCode = m_fieldListValidator.GetField(fieldId, pField);
+	if (STATUS_CODE_SUCCESS != statusCode)
+	{
+		std::cout << "ERROR fetching " << fieldName << " field for " << symbol << " - " << StatusCodeToString(statusCode) << std::endl;
+		return;
+	}
+
+
+	if (!pField->IsDefined())
+	{
+		std::cout << "undefined" << std::endl;
+	}
+	else
+	{
+		// Field contents are only valid if the Field is defined.
+		std::cout << pField->GetValue<String>() << std::endl;
+	}
+}
+
+}
+#endif // (!defined ACTIV_SAMPLE_SNAPSHOT_HANDLER_H)

src/server/sparkd/md/cme/CMEChannel.h

@@ -2,7 +2,7 @@
 #include "lib/core/FMT.h"
 #include "lib/core/FMT_IMPL.h"
 #include "lib/md/UserSymbol.h"
-#include "lib/md/cme/CMEConnection.h"
+#include "server/sparkd/md/cme/CMEConnection.h"
 
 #include <map>
 #include <ostream>

src/server/sparkd/md/cme/MDP3Utils.cpp

@@ -15,7 +15,7 @@
 #include <boost/filesystem.hpp>
 #include <iostream>
 #include <mutex>
-#include <pugixml/pugixml.hpp>
+// #include <pugixml/pugixml.hpp>
 #include <sstream>
 
 using namespace std;
@@ -132,62 +132,63 @@ std::string to_str(const std::vector<T> &vec) {
    return ss.str();
 }
 const vector<CMEChannel> parse_channels(const vector<UserSymbol> &symbols) {
-   LOG(md, Info, "parse_channels symbols: {}", FMT_JOIN(symbols, ", "));
-   await_cme_config();
+   // LOG(md, Info, "parse_channels symbols: {}", FMT_JOIN(symbols, ", "));
+   // await_cme_config();
    vector<CMEChannel> result;
-   bool skip_options = config::get<PublisherOptions>().skip_options();
-
-   pugi::xml_document doc;
-   bts_assert_msg(
-       doc.load_file(config_path().c_str()), "Error parsing config.xml as xml");
-
-   auto configuration = doc.root().child("configuration");
-   bts_assert(configuration);
-
-   LOG(md, Info, "CME: Loaded '{}' config. Last updated: {}",
-       std::string(configuration.attribute("environment").as_string("??")),
-       std::string(configuration.attribute("updated").as_string("??")));
-   auto channels = configuration.children("channel");
-   for (const auto &chan : channels) {
-      CMEChannel channel;
-      channel.id = chan.attribute("id").as_string();
-      auto channel_type =
-          infer_security_type(chan.attribute("label").as_string());
-      if (channel_type == SECURITYTYPE_OPTION && skip_options)
-         continue;
-      if (!(channel.id.empty())) {
-         auto product_list = chan.child("products").children("product");
-         for (const auto &product : product_list) {
-            auto symbol = product.attribute("code").as_string();
-            for (auto s : symbols) {
-               if ((s.type() == channel_type ||
-                    s.type() == SECURITYTYPE_SPREAD) &&
-                   s.symbol() == symbol) {
-                  channel.symbols.insert(s);
-               }
-            }
-         }
-         if (!channel.symbols.empty()) {
-            auto connections = chan.child("connections").children("connection");
-            for (const auto &srccon : connections) {
-               std::string id = srccon.attribute("id").as_string();
-               if (!id.empty()) {
-                  CMEConnection connection;
-                  connection.feed_type =
-                      srccon.child("type").attribute("feed-type").as_string();
-                  connection.ip = srccon.child_value("ip");
-                  connection.port = srccon.child("port").text().as_uint();
-                  connection.feed = srccon.child_value("feed")[0];
-                  channel.connections.insert(std::make_pair(id, connection));
-               }
-            }
-
-            result.push_back(channel);
-         }
-      }
-   }
-   LOG(md, Info, "parse_channels channels: {}", to_str(result));
+   (void)(symbols);
+   // bool skip_options = config::get<PublisherOptions>().skip_options();
    return result;
+   // pugi::xml_document doc;
+   // bts_assert_msg(
+   //     doc.load_file(config_path().c_str()), "Error parsing config.xml as xml");
+
+   // auto configuration = doc.root().child("configuration");
+   // bts_assert(configuration);
+
+   // LOG(md, Info, "CME: Loaded '{}' config. Last updated: {}",
+   //     std::string(configuration.attribute("environment").as_string("??")),
+   //     std::string(configuration.attribute("updated").as_string("??")));
+   // auto channels = configuration.children("channel");
+   // for (const auto &chan : channels) {
+   //    CMEChannel channel;
+   //    channel.id = chan.attribute("id").as_string();
+   //    auto channel_type =
+   //        infer_security_type(chan.attribute("label").as_string());
+   //    if (channel_type == SECURITYTYPE_OPTION && skip_options)
+   //       continue;
+   //    if (!(channel.id.empty())) {
+   //       auto product_list = chan.child("products").children("product");
+   //       for (const auto &product : product_list) {
+   //          auto symbol = product.attribute("code").as_string();
+   //          for (auto s : symbols) {
+   //             if ((s.type() == channel_type ||
+   //                  s.type() == SECURITYTYPE_SPREAD) &&
+   //                 s.symbol() == symbol) {
+   //                channel.symbols.insert(s);
+   //             }
+   //          }
+   //       }
+   //       if (!channel.symbols.empty()) {
+   //          auto connections = chan.child("connections").children("connection");
+   //          for (const auto &srccon : connections) {
+   //             std::string id = srccon.attribute("id").as_string();
+   //             if (!id.empty()) {
+   //                CMEConnection connection;
+   //                connection.feed_type =
+   //                    srccon.child("type").attribute("feed-type").as_string();
+   //                connection.ip = srccon.child_value("ip");
+   //                connection.port = srccon.child("port").text().as_uint();
+   //                connection.feed = srccon.child_value("feed")[0];
+   //                channel.connections.insert(std::make_pair(id, connection));
+   //             }
+   //          }
+
+   //          result.push_back(channel);
+   //       }
+   //    }
+   // }
+   // LOG(md, Info, "parse_channels channels: {}", to_str(result));
+   // return result;
 }
 
 } // namespace md

src/server/sparkd/md/cme/MDP3Utils.h

@@ -1,8 +1,8 @@
 #pragma once
 
 #include "lib/core/FMT_IMPL.h"
-#include "lib/md/cme/CMEChannel.h"
-#include "lib/md/cme/CMEConnection.h"
+#include "server/sparkd/md/cme/CMEChannel.h"
+#include "server/sparkd/md/cme/CMEConnection.h"
 #include "lib/md/coconut2.h"
 #include <set>
 #include <string>

src/server/sparkd/md/platforms/mdp3_onixs/MDP3ListenerHelpers.h

@@ -1,4 +1,5 @@
-#include "lib/md/cme/MDP3Utils.h"
+#pragma once
+#include "server/sparkd/md/cme/MDP3Utils.h"
 #include "lib/md/coconut2.h"
 #include "lib/utils/bts_assert.h"
 

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

@@ -4,7 +4,7 @@
 #include "lib/logging/logging.h"
 #include "lib/md/MBOProcessor.h"
 #include "lib/md/SecurityDB.h"
-#include "lib/md/cme/MDP3Utils.h"
+#include "server/sparkd/md/cme/MDP3Utils.h"
 #include "lib/md/coconut2.h"
 #include "lib/md/protos/co2.pb.h"
 #include "lib/md/vendor/OnixsUtils.h"