clean up: Sort includes, undef macros in headers

In general, it's best not to export macros from headers

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/client/spark_api_server/Program.cs

@@ -38,9 +38,9 @@
          var gwy = new bts.websocket_core.WebsocketConnectionGateway();
          var prms = new WebsocketServerConfig {
             DefaultEndPoint = "/",
-            Host = "titan",
+            Host = "0.0.0.0",
             Scheme = "http",
-            Port = "8080",
+            Port = "3000",
             KeepAliveIntervalSecs = 0.1,
          };
          gwy.init(sync);

src/client/spark_websocket/Core/WebsocketConnectionGateway.cs

@@ -424,6 +424,7 @@ namespace bts.websocket_core {
       /// </summary>
       /// <param name="server_params">The parameters for the websocket server includind the URL and endpoint,</param>
       public Task start_listening(WebsocketServerConfig server_params) {
+         LoggerContext logger = new(Log.Gen, $"WebsocketGateway.start_listening ", "");
          // start listening, assert that init has been called already
          bts.Log.notnull(_sync);
          bts.Log.isnull(_server_params);
@@ -433,6 +434,7 @@ namespace bts.websocket_core {
          _server_params = server_params;
          var builder = WebApplication.CreateBuilder();
          builder.WebHost.UseUrls(_server_params.Uri());
+         logger.warn(_server_params.Uri());
          // var MyAllowSpecificOrigins = "_myAllowSpecificOrigins";
          builder.Services.AddLogging(builder => {
             builder.AddFilter("Microsoft", Microsoft.Extensions.Logging.LogLevel.Warning)

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/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/server/sparkd/main.cpp

@@ -8,7 +8,11 @@
 #include "server/sparkd/cmds/Run.h"
 #include "server/sparkd/cmds/Web.h"
 #include "server/sparkd/cmds/crypto/CryptoCmd.h"
+#include "server/sparkd/md/cme/CMEActivSession.h"
+
 #include <CLI11/CLI11.hpp>
+#include <iostream>
+#include <string>
 
 using namespace bts;
 
@@ -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

@@ -1,5 +1,6 @@
 #include "MarketDataGateway.h"
 #include "lib/api/md/EmumbaDeribitMarketDataConfig.pb.h"
+#include "lib/api/md/GlbxMarketDataConfig.pb.h"
 #include "lib/api/md/MarketDataConfig.pb.h"
 #include "lib/api/md/MarketDataConfigUtils.h"
 #include "lib/api/md/MarketDataReplayConfig.pb.h"
@@ -25,6 +26,7 @@
 #include "server/sparkd/md/binance/BinanceSubscription.h"
 #include "server/sparkd/md/binance_us/BinanceUsSubscription.h"
 #include "server/sparkd/md/binance_usds_futures/BinanceUsdsFuturesSubscription.h"
+#include "server/sparkd/md/cme/CMEActivSession.h"
 #include "server/sparkd/md/coinbase/CoinbaseSubscription.h"
 #include "server/sparkd/md/deribit/DeribitSubscription.h"
 #include "server/sparkd/md/emumba/EmumbaDeribitData.h"
@@ -133,7 +135,7 @@ void MarketDataGateway::run() {
    }
    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) {
@@ -230,7 +232,33 @@ 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,6 +287,27 @@ static void run_onixs() {
    disp.run();
 }
 
+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,

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/CMEActivQueryHandler.h

@@ -0,0 +1,116 @@
+#pragma once
+
+#include "lib/md/GeneralMarketDataListener.h"
+#include "server/sparkd/md/cme/CMEActivSubscriptionHandler.h"
+
+#include <ActivMiddleware/CxxYZ/unordered_map.h>
+#include <ActivOneApi/OneApi.h>
+
+// Be sure to undef this at the end of the file. As a general rule, headers
+// shouldn't export macros.
+#define UNUSED(x) (void)(x)
+
+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
+
+#undef UNUSED

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

@@ -0,0 +1,92 @@
+#include "CMEActivSession.h"
+#include "lib/logging/logging.h"
+#include "lib/md/GeneralMarketDataListener.h"
+#include "server/sparkd/md/cme/CMEActivQueryHandler.h"
+#include "server/sparkd/md/cme/CMEActivSessionHandler.h"
+#include "server/sparkd/md/cme/CMEActivSnapshotHandler.h"
+#include "server/sparkd/md/cme/CMEActivSubscriptionHandler.h"
+
+#include <ActivMiddleware/ActivBase/Exception.h>
+#include <ActivMiddleware/ActivBase/FieldTypes/ActivFieldTypes.h>
+#include <ActivMiddleware/ActivBase/FieldTypes/FieldMap.h>
+#include <ActivOneApi/Constants/DataSourceIds.h>
+#include <ActivOneApi/Constants/FieldIds.h>
+#include <ActivOneApi/Session.h>
+#include <exception>
+
+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,20 @@
+#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,34 @@
+#include "CMEActivSessionHandler.h"
+#include "server/sparkd/md/cme/CMEActivSubscriptionHandler.h"
+
+#include <ActivMiddleware/ActivBase/ActivBase.h>
+#include <ActivOneApi/Session.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,24 @@
+#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/CMEActivSnapshotHandler.h

@@ -0,0 +1,98 @@
+/**
+ *	@file	source/ActivOnePlatform/Samples/SnapshotFieldListValidatorSample/SnapshotHandler.h
+ *
+ *	@brief	SnapshotHandler definition.
+ */
+
+#if (!defined ACTIV_SAMPLE_SNAPSHOT_HANDLER_H)
+#define ACTIV_SAMPLE_SNAPSHOT_HANDLER_H
+
+#include <ActivOneApi/MarketData/MarketData.h>
+#include <ActivOneApi/OneApi.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/CMEActivSubscriptionHandler.h

@@ -0,0 +1,372 @@
+#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 "lib/orderman/orderman_utils/Book.h"
+#include "lib/persistence/IDGenerator.h"
+#include "lib/utils/bts_assert.h"
+#include "server/sparkd/md/MarketDataGateway.h"
+
+#include <ActivMiddleware/ActivBase/ActivBase.h>
+#include <ActivOneApi/ISessionHandler.h>
+#include <ActivOneApi/ISubscriptionHandler.h>
+#include <ActivOneApi/MarketData/EventTypes.h>
+#include <ActivOneApi/MarketData/FieldIds.h>
+#include <ActivOneApi/MarketData/MarketData.h>
+#include <ActivOneApi/MarketData/Separators.h>
+#include <ActivOneApi/Messages/RefreshMessage.h>
+#include <ActivOneApi/Messages/UpdateMessage.h>
+#include <ActivOneApi/OneApi.h>
+#include <cassert>
+#include <iostream>
+#include <string>
+
+using namespace Activ;
+using namespace Activ::OneApi;
+
+#define UNUSED(x) ((void)x)
+
+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
+
+#undef UNUSED