refactor: Increase md logging, clean up/refactor (check pt)

src/lib/core/ProtoPrinter.h

@@ -1,6 +1,9 @@
 #pragma once
+#include "lib/utils/OSTREAM_FMT_IMPL.h"
+#include <fmt/format.h>
 #include <google/protobuf/text_format.h>
 #include <iostream>
+#include <sstream>
 
 namespace bts::core {
 
@@ -10,6 +13,7 @@ class ProtoPrinter {
    explicit ProtoPrinter(const T *message) noexcept;
    static void print(const T &message, std::ostream &out = std::cout);
 
+   std::string to_string() const;
    const T *message;
 };
 
@@ -21,6 +25,9 @@ class ProtoPrinterFactory {
  public:
    template <class T>
    static ProtoPrinter<T> create(const T *message);
+
+   template <class T>
+   static std::string to_string(const T *message);
 };
 
 //////////////////////////////////////////////////
@@ -42,6 +49,18 @@ ProtoPrinter<T> ProtoPrinterFactory::create(const T *message) {
    return ProtoPrinter<T>(message);
 }
 
+template <class T>
+std::string ProtoPrinterFactory::to_string(const T *message) {
+   return ProtoPrinter<T>(message).to_string();
+}
+
+template <class T>
+std::string ProtoPrinter<T>::to_string() const {
+   std::stringstream out;
+   out << *this;
+   return out.str();
+}
+
 template <class T>
 std::ostream &operator<<(
     std::ostream &out, const ProtoPrinter<T> &printer) noexcept {
@@ -52,3 +71,13 @@ std::ostream &operator<<(
 }
 
 } // namespace bts::core
+
+template <class T>
+struct fmt::formatter<bts::core::ProtoPrinter<T>> {
+   constexpr auto parse(fmt ::format_parse_context &ctx) { return ctx.end(); }
+
+   template <typename FormatContext>
+   auto format(const bts ::core ::ProtoPrinter<T> &t, FormatContext &ctx) {
+      return fmt::format_to(ctx.out(), "{}", t);
+   }
+};

src/lib/domain/AggressorSide.h

@@ -8,6 +8,14 @@ class AggressorSide {
    enum Value { None = 0, Buy = 1, Sell = 2 };
    Value value;
 
+   AggressorSide() = default;
+   constexpr AggressorSide(AggressorSide::Value value) noexcept; // NOLINT
+   AggressorSide(const AggressorSide &) noexcept = default;
+   AggressorSide(AggressorSide &&) noexcept = default;
+   AggressorSide &operator=(const AggressorSide &) noexcept = default;
+   AggressorSide &operator=(AggressorSide &&) noexcept = default;
+   ~AggressorSide() = default;
+
    constexpr const char *name() const noexcept;
 };
 
@@ -18,6 +26,9 @@ std::ostream &operator<<(
 // Definition
 //////////////////////////////////////////////////
 
+constexpr AggressorSide::AggressorSide(AggressorSide::Value value) noexcept
+    : value(value) {}
+
 constexpr const char *AggressorSide::name() const noexcept {
    switch (value) {
    case AggressorSide::None:

src/lib/domain/OptionSecurity.h

@@ -4,6 +4,11 @@
 
 namespace bts::domain {
 
+enum class CallPut {
+   Call,
+   Put
+};
+
 class OptionSecurity {
  public:
    enum CallPut { Call = 0, Put };

src/lib/domain/TradeRecord.h

@@ -1,20 +1,19 @@
 #pragma once
+#include "AggressorSide.h"
+#include "Price.h"
+#include "Quantity.h"
 #include "SecurityID.h"
 #include "lib/core/TimeStamp.h"
-#include "AggressorSide.h"
-#include "Quantity.h"
-#include "Price.h"
 
 namespace bts::domain {
 
 class TradeRecord {
  public:
    SecurityID sec_id = {};
-   TimeStamp time_stamp = {};
+   core::TimeStamp time_stamp = {};
    AggressorSide agg_side = {};
    Quantity trade_quantity;
    Price trade_price;
 };
 
-
 } // namespace bts::domain

src/lib/domain/Venue.h

@@ -1,14 +1,51 @@
 #pragma once
+#include <ostream>
 
 namespace bts::domain {
 
 enum class Venue {
    None = 0,
-   Glbx = 1,
-   Deribit = 2,
-   Huobi = 3,
-   Binance = 4,
-   FTX = 5
+   Sim = 1,
+   Glbx = 2,
+   Deribit = 3,
+   Huobi = 4,
+   Binance = 5,
+   FTX = 6
 };
 
+constexpr const char *Venue_NAME(Venue venue);
+
+std::ostream &operator<<(std::ostream &out, const Venue &venue) noexcept;
+
+//////////////////////////////////////////////////
+// Definition
+//////////////////////////////////////////////////
+
+constexpr const char *Venue_NAME(Venue venue) {
+   switch (venue) {
+   case Venue::None:
+      return "None";
+   case Venue::Sim:
+      return "Sim";
+   case Venue::Glbx:
+      return "Glbx";
+   case Venue::Deribit:
+      return "Deribit";
+   case Venue::Huobi:
+      return "Huobi";
+   case Venue::Binance:
+      return "Binance";
+   case Venue::FTX:
+      return "FTX";
+   default:
+      return "Unknown";
+   }
+}
+
+inline std::ostream &operator<<(
+    std::ostream &out, const Venue &venue) noexcept {
+   out << Venue_NAME(venue);
+   return out;
+}
+
 } // namespace bts::domain

src/lib/md/MDClient.cpp

@@ -7,6 +7,7 @@
 #include "lib/utils/TimeStamp.h"
 #include <set>
 #include <string>
+#include <thread>
 
 namespace bts {
 namespace md {
@@ -40,6 +41,8 @@ struct MDClientImpl {
        : _dispatcher(d), _ss(s), _security_timer(), _recovery_timer(),
          _symbols(symbols ? *symbols : get_system_md_symbols()), _mbo(mbo),
          _consumer(_cache, _ss, mbo, &_symbols), _event() {
+      LOG(md, Info, "MDClient initializing with symbols: {}",
+          fmt::to_string(fmt::join(_symbols, ", ")));
       if (d) {
          _log_cons();
          _security_timer = std::make_unique<utils::DispatcherTimer>(
@@ -73,9 +76,10 @@ struct MDClientImpl {
          TimeStamp start = TimeStamp::now();
          auto count = _consumer.scan_securities();
          auto delta = (TimeStamp::now() - start);
-         if (delta.milliseconds() > 10) {
-            LOG(md, Diag, "Scanned handles: {}ms  found: {} securities. ",
-                delta.milliseconds(), count);
+         auto thread_id = std::this_thread::get_id();
+         if (delta.milliseconds() > 10 || count > 0) {
+            LOG(md, Info, "Scanned handles: {}ms  found: {} securities; thread: {}",
+                delta.milliseconds(), count, thread_id);
          }
          return count;
       }
@@ -105,4 +109,4 @@ SecurityOracle MDClient::oracle() const {
 std::set<UserSymbol> MDClient::symbols() const { return _self->_symbols; }
 
 } // namespace md
-} // namespace bts
+} // namespace bts

src/lib/md/MDConsumer.cpp

@@ -17,6 +17,7 @@
 #include <algorithm>
 #include <btree/map.h>
 #include <map>
+#include <range/v3/view.hpp>
 #include <set>
 #include <string>
 #include <tbb/queuing_mutex.h>
@@ -95,17 +96,31 @@ class MDConsumerInst {
           "platform: {}",
           symbol, static_cast<int>(exchange), static_cast<int>(type),
           static_cast<int>(platform));
+      std::string hot_seg_name(
+          hot_segment_name(symbol, type, exchange, _platform));
       if (enable_depth(symbol, type)) {
+         LOG(md, Info,
+             "MDConsumerInst in {}. init hot-depth shm for {}-{}-{}. segment "
+             "name: {}",
+             std::string(__FUNCTION__), symbol, SecurityType_Name(type),
+             Venue_Name(exchange), hot_seg_name);
          _hot_depth.reset(new utils::SharedListReader<HotHeader, HotDepthEntry>(
-             hot_segment_name(symbol, type, exchange, _platform), shared));
+             hot_seg_name, shared));
       } else {
+         LOG(md, Info,
+             "MDConsumerInst in {}. init hot shm for {}-{}-{}. segment name: "
+             "{}",
+             std::string(__FUNCTION__), symbol, SecurityType_Name(type),
+             Venue_Name(exchange), hot_seg_name);
          _hot.reset(new utils::SharedListReader<HotHeader, HotEntry>(
-             hot_segment_name(symbol, type, exchange, _platform), shared));
+             hot_seg_name, shared));
       }
    }
    ~MDConsumerInst() {
-      if (_queues)
+      if (_queues) {
+         LOG(md, Info, "MDConsumerInst in destructor. deleting queues");
          delete _queues;
+      }
    }
 
    template <typename CB>
@@ -600,16 +615,9 @@ class MDConsumerImpl {
        co2::Platform platform)
        : _workmutex(), _instances(), _sci(), _mbo(mbo), _deferred_handles(),
          _callbacks(), _cache(cache) {
-      std::ostringstream msg;
-      for (const auto &symbol : symbols) {
-         if (msg.tellp() > 0) {
-            msg << ", ";
-         }
-         msg << symbol.symbol() << "-" << SecurityType_Name(symbol.type())
-             << " " << Venue_Name(symbol.exchange());
-      }
-      LOG(md, Info, "Created MDConsumerImpl for securities: {} on {}",
-          msg.str(), co2::Platform_Name(platform));
+      LOG(md, Info, "Created MDConsumer for securities on {}: {}",
+          co2::Platform_Name(platform),
+          fmt::to_string(fmt::join(symbols, ", ")));
 
       for (auto &s : symbols) {
          if (s.type() == SECURITYTYPE_FUTURE ||
@@ -895,6 +903,13 @@ class MDConsumerImpl {
             }
             count += found_securities.size();
          }
+         if (!found_securities.empty()) {
+            auto found_syms = ranges::transform_view(
+                found_securities,
+                [](const MutSecurityPtr &s) { return s->symbol(); });
+            LOG(md, Info, "MDConsumer found new securities: {}",
+                fmt::to_string(fmt::join(found_syms, ", ")));
+         }
       }
 
       if (_sci) {
@@ -916,6 +931,12 @@ class MDConsumerImpl {
       }
 
       _cache.flush_to_disk();
+      if (count != 0) {
+         auto thread_id = std::this_thread::get_id();
+         LOG(md, Info,
+             "scan_securities actually succeeded. found: {}; thread: {}", count,
+             thread_id);
+      }
       return count;
    }
 

src/lib/md/MDConsumerInterop.cpp

@@ -1,10 +1,12 @@
 #ifdef _WIN32
 #include "MDConsumer.h"
+#include "lib/utils/OSTREAM_FMT_IMPL.h"
 #include "SecurityCache.h"
 #include "UserSymbol.h"
 #include "common.h"
 #include "get_symbols.h"
 #include "lib/md/protos/co2.pb.h"
+#include "lib/domain_model/Security.pb.h"
 #include "lib/utils/Dispatcher.h"
 #include "lib/utils/defer.h"
 #include "relay/RelayClient.h"
@@ -12,8 +14,7 @@
 #include <set>
 #include <string>
 
-namespace bts {
-namespace md {
+namespace bts::md {
 
 #define CAPI extern "C" BTS_EXPORT
 
@@ -26,6 +27,17 @@ struct BlitUserSymbol {
    SecurityType type;
 };
 
+inline std::ostream &operator<<(std::ostream &out, const BlitUserSymbol &bus) noexcept {
+   out << bus.symbol << "-" << api::SecurityType_Name(type) << "-" << api::Venue_Name(bus.exchange);
+   return out;
+}
+
+}
+
+OSTREAM_FMT_IMPL(bts::md::BlitUserSymbol);
+
+namespace bts::md {
+
 CAPI MDConsumer *
 create_md_consumer(BlitUserSymbol *usersymbols, uint32_t count, bool shared) {
    SecurityCache cache;
@@ -169,6 +181,7 @@ CAPI int md_consumer_read_rfqs(
 CAPI void add_relay_symbol(
     md::RelayControl *prelay, BlitUserSymbol &symbol, uint32_t flags) {
    co2::SecurityChannel channel;
+   LOG(md, Info, "Adding relay symbol: {}", symbol);
    channel.set_symbol(symbol.symbol);
    channel.set_exchange(symbol.exchange);
    channel.set_type(symbol.type);
@@ -211,6 +224,5 @@ CAPI void dump_co2_memory_to_disk(MDConsumer *consumer, const char *path) {
    }
 }
 
-} // namespace md
 } // namespace bts
 #endif

src/lib/md/MDProducer.cpp

@@ -1,5 +1,10 @@
 #include "lib/domain_model/Security.pb.h"
 #include "lib/logging/logging.h"
+#include "lib/os/os.h"
+#include "lib/utils/ThreadDescriptor.h"
+#include <fmt/format.h>
+#include <spdlog/fmt/bundled/format.h>
+#include <thread>
 #ifndef NO_CPP0X
 #include "MDProducer.h"
 
@@ -85,6 +90,12 @@ class MDProducerImpl {
          // set_key() set it in hot and cold and make the security public.
          c_entry->security.key = invalid_security_key;
          handle.set_key(security.key);
+      } else {
+         LOG(md, Info,
+             "MDProducerHandle::add_security INVALID. thread: {}; has_depth: "
+             "{}; sec: {}",
+             threading::this_thread_descriptor, has_depth(),
+             fmt::format("{}", security));
       }
 
       return handle;

src/lib/md/Shm/FixedSecurity.cpp

@@ -1,4 +1,6 @@
 #include "lib/utils/bts_assert.h"
+#include "lib/domain_model/Security.pb.h"
+#include "lib/utils/OSTREAM_FMT_IMPL.h"
 #ifndef NO_CPP0X
 
 #include "../PriceSize.h"
@@ -13,8 +15,7 @@ using namespace std;
 using namespace bts::api;
 using std::size_t;
 
-namespace bts {
-namespace md {
+namespace bts::md {
 
 FixedSecurity::FixedSecurity()
     : size(sizeof(FixedSecurity)), key(invalid_security_key), numeric_id(0),
@@ -415,7 +416,44 @@ void convert_security(
    }
 }
 
-} // namespace md
-} // namespace bts
+std::ostream &operator<<(std::ostream &out, const FixedSecurity &/*security*/) {
+   // out << api::Venue_Name(security.exchange) << "_"
+   //     << api::SecurityType_Name(security.type) << "_" << security.symbol();
+   return out;
+}
+
+std::ostream &operator<<(std::ostream &out, const FixedStockSecurity &stock) {
+   out << api::Venue_Name(stock.exchange) << "_"
+       << api::SecurityType_Name(stock.type) << "_" << stock.symbol();
+   return out;
+}
+
+std::ostream &operator<<(std::ostream &out, const FixedFutureSecurity &future) {
+   out << api::Venue_Name(future.exchange) << "_"
+       << api::SecurityType_Name(future.type) << "_" << future.symbol() << "_"
+       << std::setfill('0') << std::setw(2) << future.expiration_month << "-"
+       << std::setw(2) << future.expiration_day << "-" << std::setw(2)
+       << future.expiration_year - 2000;
+   return out;
+}
+
+std::ostream &operator<<(std::ostream &out, const FixedOptionSecurity &option) {
+   out << api::Venue_Name(option.exchange) << "_"
+       << api::SecurityType_Name(option.type) << "_" << option.symbol() << "_"
+       << std::setfill('0') << std::setw(2) << option.expiration_month << "-"
+       << std::setw(2) << option.expiration_day << "-" << std::setw(2)
+       << option.expiration_year - 2000 << "_" << option.strike << "_"
+       << api::PutOrCall_Name(option.call_put);
+   return out;
+}
+
+std::ostream &operator<<(std::ostream &out, const FixedSpreadLeg &leg) {
+   out << "[" << leg.exchange_key << ":" << leg.key << "]"
+       << api::Side_Name(leg.leg_side) << "_" << leg.leg_quantity << "@"
+       << leg.leg_price;
+   return out;
+}
+
+} // namespace bts::md
 
 #endif

src/lib/md/Shm/FixedSecurity.h

@@ -1,18 +1,21 @@
 #pragma once
-
 #include "../common.h"
+#include "lib/domain_model/Security.pb.h"
+#include "lib/utils/OSTREAM_FMT_IMPL.h"
 #include "lib/utils/ValueWrapper.h"
 #include "lib/utils/bts_array_size.h"
 #include "lib/utils/bts_assert.h"
 #include "lib/utils/bts_export.h"
 #include "lib/utils/bts_no_copy.h"
 #include "lib/utils/warn_ctrl.h"
-
+#include <fmt/format.h>
+#include <iomanip>
+#include <ostream>
 #include <sstream>
 #include <string>
+#include <tuple>
 
-namespace bts {
-namespace md {
+namespace bts::md {
 
 using SecurityKey = utils::ValueWrapper<struct SecurityKeyTag, uint32_t>;
 using ExchangeKey = utils::ValueWrapper<struct ExchangeKeyTag, uint64_t>;
@@ -37,6 +40,9 @@ class BTS_EXPORT FixedSecurity {
 
  public:
    FixedSecurity(const FixedSecurity &) = delete;
+   FixedSecurity(FixedSecurity &&) = default;
+   FixedSecurity &operator=(const FixedSecurity &) = delete;
+   FixedSecurity &operator=(FixedSecurity &&) = default;
    ~FixedSecurity();
 
    STRING_PROPERTY(symbol);
@@ -124,7 +130,8 @@ struct BTS_EXPORT FixedSpreadLeg {
    double leg_delta;
 };
 
-template <int LEGCOUNT = 16> class FixedSpreadSecurity : public FixedSecurity {
+template <int LEGCOUNT = 16>
+class FixedSpreadSecurity : public FixedSecurity {
    char _summary[128];
 
  public:
@@ -162,7 +169,8 @@ template <int LEGCOUNT = 16> class FixedSpreadSecurity : public FixedSecurity {
 class FixedSyntheticSecurity : public FixedSecurity {
  public:
    FixedSyntheticSecurity(const api::Security &sec)
-       : _leg_count(sec.synthetic().legs_size()), _buffer_size(static_cast<std::uint32_t>(sec.ByteSizeLong())) {
+       : _leg_count(sec.synthetic().legs_size()),
+         _buffer_size(static_cast<std::uint32_t>(sec.ByteSizeLong())) {
       type = api::SECURITYTYPE_SYNTHETIC;
       sec.SerializeToArray(buffer(), _buffer_size);
       size = sizeof(FixedSyntheticSecurity) + _buffer_size;
@@ -263,7 +271,8 @@ void with_security(
       // buffer and only heap allocating if we exceed that but in general
       // FixedSynthetic is not performance sensitive
       std::vector<uint8_t> buffer;
-      buffer.resize(psynthetic->ByteSizeLong() + sizeof(FixedSyntheticSecurity));
+      buffer.resize(
+          psynthetic->ByteSizeLong() + sizeof(FixedSyntheticSecurity));
       auto *psec = new (&buffer[0]) FixedSyntheticSecurity(*psynthetic);
       cb(*psec);
       return;
@@ -282,5 +291,43 @@ void with_security_like(const api::Security &security, C cb) {
        securitytype_of(security), num_legs(security), &security, cb);
 }
 
-} // namespace md
-} // namespace bts
+std::ostream &operator<<(std::ostream &out, const FixedSecurity &security);
+std::ostream &operator<<(std::ostream &out, const FixedStockSecurity &stock);
+std::ostream &operator<<(std::ostream &out, const FixedFutureSecurity &future);
+std::ostream &operator<<(std::ostream &out, const FixedOptionSecurity &option);
+std::ostream &operator<<(std::ostream &out, const FixedSpreadLeg &leg);
+template <int LEGCOUNT>
+std::ostream &operator<<(
+    std::ostream &out, const FixedSpreadSecurity<LEGCOUNT> &spread);
+
+} // namespace bts::md
+
+namespace bts::md {
+
+template <int LEGSIZE>
+std::ostream &operator<<(
+    std::ostream &out, const FixedSpreadSecurity<LEGSIZE> &spread) {
+   out << api::Venue_Name(spread.exchange) << "_"
+       << api::SecurityType_Name(spread.type) << "_" << spread.symbol() << "_"
+       << std::setfill('0') << std::setw(2) << spread.expiration_month << "-"
+       << std::setw(2) << spread.expiration_day << "-" << std::setw(2)
+       << spread.expiration_year - 2000;
+   if (spread.leg_count() > 0) {
+      out << " (" << spread.legs[0];
+      for (auto i = 1; i < spread.leg_count(); i++) {
+         out << " " << spread.legs[i];
+      }
+      out << ")";
+   }
+   return out;
+}
+
+} // namespace bts::md
+
+OSTREAM_FMT_IMPL(bts::md::FixedSecurity);
+OSTREAM_FMT_IMPL(bts::md::FixedStockSecurity);
+OSTREAM_FMT_IMPL(bts::md::FixedFutureSecurity);
+OSTREAM_FMT_IMPL(bts::md::FixedOptionSecurity);
+OSTREAM_FMT_IMPL(bts::md::FixedSpreadLeg);
+OSTREAM_FMT_IMPL_TEMPLATED(
+    bts::md::FixedSpreadSecurity<LEGCOUNT>, int LEGCOUNT);

src/lib/md/UserSymbol.h

@@ -1,5 +1,7 @@
 #pragma once
 #include "common.h"
+#include "lib/domain/Venue.h"
+#include "lib/utils/OSTREAM_FMT_IMPL.h"
 
 namespace bts {
 namespace md {
@@ -66,3 +68,16 @@ std::ostream &operator<<(std::ostream &os, const UserSymbol &sym);
 
 } // namespace md
 } // namespace bts
+
+template <>
+struct fmt ::formatter<bts ::md ::UserSymbol> {
+   constexpr auto parse(fmt ::format_parse_context &ctx) { return ctx.end(); }
+
+   template <typename FormatContext>
+   auto format(const bts ::md ::UserSymbol &t, FormatContext &ctx) {
+      return fmt ::format_to(
+          ctx.out(), "{}-{}-{}", t.symbol(),
+          bts::api::SecurityType_Name(t.type()),
+          Venue_Name(t.exchange()));
+   }
+};

src/lib/md/cme/GlbxMarketDataListener.h

@@ -0,0 +1,9 @@
+#pragma once
+
+namespace bts::md::glbx {
+
+class GlbxMarketDataListener {
+
+};
+
+}

src/lib/md/cme/MDP3Utils.cpp

@@ -15,12 +15,12 @@
 #include <iostream>
 #include <mutex>
 #include <pugixml/pugixml.hpp>
+#include <vector>
 
 using namespace std;
 using namespace bts::api;
 
-namespace bts {
-namespace md {
+namespace bts::md {
 
 std::ostream &operator<<(std::ostream &out, md::CMEConnection const &c) {
    out << "feed_type: " << c.feed_type << " ip: " << c.ip << " port: " << c.port
@@ -145,11 +145,15 @@ const std::vector<CMEChannel> get_configured_cme_channels() {
    get_system_md_symbols(api::VENUE_GLBX, api::SECURITYTYPE_UNSET, &symbols);
    return parse_channels(symbols);
 }
-const vector<CMEChannel> parse_channels(const vector<UserSymbol> &symbols) {
+
+const vector<CMEChannel> parse_channels(
+    const vector<UserSymbol> &symbols,
+    std::vector<const UserSymbol *> *symbols_without_channels) {
    await_cme_config();
    vector<CMEChannel> result;
    bool skip_options = config::get<PublisherOptions>().skip_options();
 
+   vector<size_t> symbols_with_channels;
    pugi::xml_document doc;
    bts_assert_msg(
        doc.load_file(config_path().c_str()), "Error parsing config.xml as xml");
@@ -164,45 +168,64 @@ const vector<CMEChannel> parse_channels(const vector<UserSymbol> &symbols) {
    for (const auto &chan : channels) {
       CMEChannel channel;
       channel.id = chan.attribute("id").as_string();
+      if (channel.id.empty()) {
+         continue;
+      }
       auto channel_type =
           infer_security_type(chan.attribute("label").as_string());
-      if (channel_type == SECURITYTYPE_OPTION && skip_options)
+      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);
+      }
+      auto product_list = chan.child("products").children("product");
+      for (const auto &product : product_list) {
+         auto symbol = product.attribute("code").as_string();
+         std::size_t idx = 0;
+         for (auto s : symbols) {
+            if ((s.type() == channel_type || s.type() == SECURITYTYPE_SPREAD) &&
+                s.symbol() == symbol) {
+               channel.symbols.insert(s);
+               if (symbols_without_channels != nullptr) {
+                  auto it = std::lower_bound(symbols_with_channels.begin(), symbols_with_channels.end(), idx);
+                  if (it == symbols_with_channels.end() || *it != idx) {
+                     symbols_with_channels.insert(it, idx);
+                  }
                }
             }
+            idx++;
+         }
+      }
+      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));
+            }
          }
-         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);
-         }
+         result.push_back(channel);
       }
    }
 
+   if (symbols_without_channels != nullptr) {
+      for (std::size_t i = 0; i < symbols_with_channels.size(); i++) {
+         std::size_t stop = symbols.size();
+         if (i != symbols_with_channels.size() - 1) {
+            stop = symbols_with_channels[i + 1];
+         }
+         for (auto j = symbols_with_channels[i] + 1; j < stop; j++) {
+            symbols_without_channels->push_back(&symbols[j]);
+         }
+      }
+   }
    return result;
 }
 
-} // namespace md
-} // namespace bts
+} // namespace bts::md
 #endif

src/lib/md/cme/MDP3Utils.h

@@ -26,9 +26,12 @@ std::ostream &operator<<(std::ostream &out, md::CMEConnection const &c);
 std::ostream &operator<<(std::ostream &out, md::CMEChannel const &ch);
 
 void await_cme_config();
+
 const std::vector<CMEChannel> get_configured_cme_channels();
-const std::vector<CMEChannel>
-parse_channels(const std::vector<UserSymbol> &symbols);
+
+const std::vector<CMEChannel> parse_channels(
+    const std::vector<UserSymbol> &symbols,
+    std::vector<const UserSymbol *> *symbols_without_channels = nullptr);
 
 const std::string config_path();
 const std::string template_path();

src/lib/md/relay/QueryServer.cpp

@@ -1,3 +1,4 @@
+#include "lib/core/ProtoPrinter.h"
 #include "lib/domain_model/Security.pb.h"
 #ifndef _WIN32
 #include "QueryServer.h"
@@ -625,6 +626,8 @@ struct MDRelayServer : bts::co2::AuthenticatedMarketDataRelayBase {
          prod->set_type(i.type());
          prod->set_symbol(i.symbol());
       }
+      LOG(md, Info, "QueryServer::request_products.  list:\n{}",
+          core::ProtoPrinterFactory::to_string(&list));
       reply(list);
    }
 

src/lib/md/relay/RelayRecorder.cpp

@@ -29,13 +29,15 @@ namespace md {
 const char *magic = "Co2";
 const uint32_t recorder_version = 3;
 
-template <class T> void writebuf(ofstream &out, const T &pb) {
+template <class T>
+void writebuf(ofstream &out, const T &pb) {
    auto size = static_cast<std::uint32_t>(pb.ByteSizeLong());
    out.write(reinterpret_cast<char *>(&size), sizeof(size));
    bts_assert(pb.SerializeToOstream(&out));
 }
 
-template <class T> bool readbuf(istream &in, T *result) {
+template <class T>
+bool readbuf(istream &in, T *result) {
    if (result) {
       uint32_t size;
       in.read(reinterpret_cast<char *>(&size), sizeof(size));
@@ -120,7 +122,18 @@ void record_security_keys(
 }
 #endif
 
-void replay_recording(istream &in, IRelayWriter &writer, int32_t naggle_ms) {
+// static co2::MDReply get_bogus_msg() {
+//    co2::MDReply rv;
+//    co2::ChannelReply &ch_reply = *rv.add_channels();
+//    co2::ChannelUpdate &ch_upd = *ch_reply.add_updates();
+//    return rv;
+// }
+
+void replay_recording(
+    istream &in,
+    IRelayWriter &writer,
+    int32_t naggle_ms,
+    const std::vector<co2::MDReply> &prepend_msgs) {
    bts_assert(!in.bad());
    CoconutReplayHeader header;
    size_t count = 0;
@@ -129,6 +142,15 @@ void replay_recording(istream &in, IRelayWriter &writer, int32_t naggle_ms) {
 #endif
    if (readbuf(in, &header) && header.magic() == magic &&
        header.version() == recorder_version) {
+      for (const auto &m : prepend_msgs) {
+         bool has_securities = false;
+         LOG(md, Info, "BEGIN Prepending message to md-replay feed");
+         writer.write(m, &has_securities);
+         LOG(md, Info, "END Prepending message to md-replay feed");
+#ifndef _WIN32
+         record_security_keys(idrec, m);
+#endif
+      }
       co2::MDReply message;
       while (readbuf(in, &message)) {
          bool has_securities = false;
@@ -158,12 +180,13 @@ void replay_file(
     const std::string &filename,
     IRelayWriter &writer,
     int32_t naggle_ms,
-    bool loop) {
+    bool loop,
+    const std::vector<co2::MDReply> &prepend_msgs) {
    do {
       ifstream in;
       in.open(filename, ios::binary | ios::in);
       if (in.is_open()) {
-         replay_recording(in, writer, naggle_ms);
+         replay_recording(in, writer, naggle_ms, prepend_msgs);
       } else {
          LOG(md, Diag, "Cannot read file {}", filename);
          if (loop)
@@ -176,7 +199,8 @@ bool replay_url(
     const std::string &url,
     IRelayWriter &writer,
     int32_t naggle_ms,
-    bool loop) {
+    bool loop,
+    const std::vector<co2::MDReply> &prepend_msgs) {
    // split up url parts
    size_t http_len = BTS_ARRAYSIZE("http://") - 1;
    auto after_host = url.find('/', http_len);
@@ -187,7 +211,7 @@ bool replay_url(
    filename = os::sparkdata_path("data/replay/") + filename;
    utils::download_file_if_newer(host, path, filename);
    if (boost::filesystem::exists(filename)) {
-      replay_file(filename, writer, naggle_ms, loop);
+      replay_file(filename, writer, naggle_ms, loop, prepend_msgs);
       return true;
    }
    return false;
@@ -197,11 +221,12 @@ void replay_url_with_fallback(
     const std::string &url,
     IRelayWriter &writer,
     int32_t naggle_ms,
-    bool loop) {
-   if (!replay_url(url, writer, naggle_ms, loop)) {
+    bool loop,
+    const std::vector<co2::MDReply> &prepend_msgs) {
+   if (!replay_url(url, writer, naggle_ms, loop, prepend_msgs)) {
       const auto fallback = config::get<ReplayOptions>().fallback();
       if (boost::filesystem::exists(fallback)) {
-         replay_file(fallback, writer, naggle_ms, loop);
+         replay_file(fallback, writer, naggle_ms, loop, prepend_msgs);
       }
    }
 }
@@ -244,7 +269,10 @@ shared_ptr<IRelayWriter> mkrelayproducer(bool shared);
 
 /*this is the public one*/
 void replay_recording(
-    const string &filename, int32_t naggle_ms, const std::string fallback) {
+    const string &filename,
+    int32_t naggle_ms,
+    const std::string fallback,
+    const std::vector<co2::MDReply> &prepend_msgs) {
    auto writer = mkrelayproducer(true);
    if (config::get<ReplayOptions>().interactive()) {
       naggle_ms = -1;
@@ -255,16 +283,17 @@ void replay_recording(
 
    if (filename.find("http://") == 0) {
       // loops forever
-      replay_url(filename, *writer, naggle_ms);
+      replay_url(filename, *writer, naggle_ms, true, prepend_msgs);
    } else if (boost::filesystem::exists(filename)) {
       // loops forever
-      replay_file(filename, *writer, naggle_ms);
+      replay_file(filename, *writer, naggle_ms, false, prepend_msgs);
    }
    // try fallback
    if (boost::filesystem::exists(fallback)) {
-      replay_file(fallback, *writer, naggle_ms);
+      replay_file(fallback, *writer, naggle_ms, false, prepend_msgs);
    }
 }
+
 void print_recording(const string &filename) {
    ConsoleReplayWriter writer;
    replay_file(filename, writer, 0);

src/lib/md/relay/RelayRecorder.h

@@ -31,23 +31,34 @@ void record_everything(std::string path, std::string filename);
 void print_recording(const std::string &);
 void unprint_recording(const std::string &src, const std::string &dest);
 void trace_recording(const std::string &file, security_key key);
+
 void replay_recording(
-    const std::string &, int32_t delay_ms, const std::string fallback = "");
+    const std::string &,
+    int32_t delay_ms,
+    const std::string fallback = "",
+    const std::vector<co2::MDReply> &prepend_msgs = {});
+
 bool replay_url(
     const std::string &url,
     IRelayWriter &writer,
     int32_t naggle_ms,
-    bool loop = true);
+    bool loop = true,
+    const std::vector<co2::MDReply> &prepend_msgs = {});
+
 void replay_url_with_fallback(
     const std::string &url,
     IRelayWriter &writer,
     int32_t naggle_ms,
-    bool loop = true);
+    bool loop = true,
+    const std::vector<co2::MDReply> &prepend_msgs = {});
+
 void replay_file(
     const std::string &filename,
     IRelayWriter &writer,
     int32_t naggle_ms,
-    bool loop = false);
+    bool loop = false,
+    const std::vector<co2::MDReply> &prepend_msgs = {});
+
 void get_securities_from_replay(std::function<void(const api::Security &)>);
 
 } // namespace md

src/lib/md/service/MarketDataServiceServer.cpp

@@ -8,6 +8,7 @@
 
 #include <atomic>
 #include <set>
+#include <range/v3/view.hpp>
 
 namespace bts {
 namespace md {
@@ -61,6 +62,9 @@ void MarketDataService::get_products(
       prod->set_type(i.type());
       prod->set_symbol(i.symbol());
    }
+   auto syms_str = fmt::to_string(fmt::join(syms, ", "));
+   LOG(md, Info, "Gathering products: {}", syms_str);
+   
    reply(list);
 }
 
@@ -74,6 +78,11 @@ void MarketDataService::allow_securities(
     const bts::api::AuthToken &,
     const std::string &qualifier,
     const api::Response<bts::api::StreamAccessResponse> &reply) {
+   LOG(md, Info, "Allow securities {}", qualifier);
+   bool is_bitcoin = qualifier.find("BTC") != std::string::npos;
+   if (is_bitcoin) {
+      LOG(md, Info, "BTC securities auth...");
+   }
    // Non-empty qualifier is invalid and rejected
    reply(authorized(qualifier.empty()));
 }
@@ -82,6 +91,7 @@ void MarketDataService::allow_book(
     const bts::api::AuthToken &token,
     const std::string &qualifier,
     const api::Response<bts::api::StreamAccessResponse> &reply) {
+   LOG(md, Info, "Allow book {}", qualifier);
    api::user::TokenPayload payload;
    if (api::auth::decode_token(token, &payload)) {
       auto product = api::string_to_product(qualifier);
@@ -97,17 +107,27 @@ void MarketDataService::allow_trades(
     const bts::api::AuthToken &token,
     const std::string &qualifier,
     const api::Response<bts::api::StreamAccessResponse> &reply) {
+   bool is_bitcoin = qualifier.find("BTC") != std::string::npos;
+   bool decoded = false;
+   bool is_streaming = false;
    api::user::TokenPayload payload;
    if (api::auth::decode_token(token, &payload)) {
+      decoded = true;
       auto product = api::string_to_product(qualifier);
       if (product) {
+         is_streaming = true;
          // todo: start streaming
          reply(authorized(true));
       }
    }
+   if (is_bitcoin) {
+      LOG(md, Info,
+          "BTC trades auth. qualifier: {}; decoded: {}; is_streaming: {}",
+          qualifier, decoded, is_streaming);
+   }
    reply(authorized(false));
 }
 
 } // namespace md
 } // namespace bts
-#endif
+#endif

src/lib/os/os.cpp

@@ -1,6 +1,7 @@
 #include "os.h"
 #include "lib/config/config.h"
 #include "lib/utils/bts_array_size.h"
+#include "lib/utils/ThreadDescriptor.h"
 #include "lib/utils/bts_assert.h"
 #include "lib/utils/bts_debugger.h"
 #include "lib/utils/helper_macros.h"
@@ -926,6 +927,8 @@ BTS_EXPORT void set_thread_name(std::string const &n) {
 }
 
 BTS_EXPORT void set_thread_name(const char *name) {
+   threading::this_thread_descriptor.name = name;
+   threading::this_thread_descriptor.id = std::this_thread::get_id();
 #ifdef _WIN32
    if (!IsDebuggerPresent())
       return;

src/lib/utils/OSTREAM_FMT_IMPL.h

@@ -1,7 +1,6 @@
 #pragma once
 
-#define OSTREAM_FMT_IMPL(T)                                                    \
-   template <>                                                                 \
+#define OSTREAM_FMT_IMPL_BODY(T)                                               \
    struct fmt::formatter<T> {                                                  \
       constexpr auto parse(fmt::format_parse_context &ctx) {                   \
          return ctx.end();                                                     \
@@ -12,3 +11,11 @@
          return fmt::format_to(ctx.out(), "{}", t);                            \
       }                                                                        \
    }
+
+#define OSTREAM_FMT_IMPL_TEMPLATED(T, ...)                                     \
+   template <__VA_ARGS__>                                                      \
+   OSTREAM_FMT_IMPL_BODY(T)
+
+#define OSTREAM_FMT_IMPL(T)                                                    \
+   template <>                                                                 \
+   OSTREAM_FMT_IMPL_BODY(T)

src/lib/utils/SharedList.h

@@ -49,7 +49,8 @@ inline bool should_share(bool desire_shared) {
 }
 
 // Single writer, single thread!
-template <class THeader, class TEntry> class SharedList {
+template <class THeader, class TEntry>
+class SharedList {
 
  public:
    SharedList(
@@ -62,11 +63,21 @@ template <class THeader, class TEntry> class SharedList {
          _segments_mapped(0), _current_region(nullptr),
          _current_segment_header(nullptr), _header(nullptr),
          _shm(create_low_level_shared_memory(should_share(shared))) {
-      if (shared)
+      if (shared) {
+         LOG(md, Info, "SharedList {} in {}. BEGIN delete segments", name,
+             std::string(__FUNCTION__));
          delete_segments(_name);
+         LOG(md, Info, "SharedList {} in {}. END delete segments", name,
+             std::string(__FUNCTION__));
+      }
 
+      auto thread_id = std::this_thread::get_id();
+      LOG(md, Info, "SharedList {} in {}. BEGIN expand shm. thread: {}", name,
+          std::string(__FUNCTION__), thread_id);
       // Start with 1 segment
       expand(_current_region_size);
+      LOG(md, Info, "SharedList {} in {}. END expand shm. thread: {}", name,
+          std::string(__FUNCTION__), thread_id);
 
       // Create the header in the first segment
       _header = new (allocate(sizeof(THeader))) THeader();
@@ -146,14 +157,20 @@ template <class THeader, class TEntry> class SharedList {
 
  private:
    void expand(size_t size) {
+      auto thread_id = std::this_thread::get_id();
       // using namespace boost::interprocess;
       char segment_name[512];
       get_segment_name(
           segment_name, BTS_ARRAYSIZE(segment_name), _name, _segments_mapped);
 
-      LOG(md, Info, "Creating shared segment '{}'", std::string(segment_name));
+      LOG(md, Info, "Creating shared segment '{}'.  thread: {}",
+          std::string(segment_name), thread_id);
 
       _current_region = _shm->create(segment_name, size).address;
+      if (!_current_region) {
+         auto curr_seg = _shm->create(segment_name, size);
+         _current_region = curr_seg.address;
+      }
       bts_assert(_current_region);
       _current_region_size = size;
       _current_offset = 0;
@@ -201,7 +218,8 @@ template <class THeader, class TEntry> class SharedList {
 };
 
 // Thread safe to read
-template <class THeader, class TEntry> class SharedListReader {
+template <class THeader, class TEntry>
+class SharedListReader {
  public:
    SharedListReader(const std::string &name, bool shared = true)
        : _name(name), _header(nullptr), _mutex(),
@@ -219,7 +237,8 @@ template <class THeader, class TEntry> class SharedListReader {
       return *_header;
    }
 
-   template <class CB> void all_entries(CB callback) {
+   template <class CB>
+   void all_entries(CB callback) {
       init();
       if (!is_initialized())
          return;
@@ -303,19 +322,24 @@ template <class THeader, class TEntry> class SharedListReader {
       }
       // Find the first segment
       char segment_name[512];
+      std::string sname(segment_name);
+      bool is_test = sname.find("TEST") != std::string::npos;
+      if (is_test) {
+         // std::cout << "On test sec" << std::endl;
+      }
       SharedList<THeader, TEntry>::get_segment_name(
           segment_name, BTS_ARRAYSIZE(segment_name), _name, 0);
       auto segment = _shm->open(segment_name, !os::is_windows());
+      auto thread_id = std::this_thread::get_id();
       if (segment) {
-         LOG(md, Info, "Opening shared segment '{}' - SUCCEEDED",
-             std::string(segment_name));
+         LOG(md, Info, "Opening shared segment '{}' - SUCCEEDED.  thread: {}",
+             std::string(segment_name), thread_id);
          _header = (THeader *)(segment.address + sizeof(SegmentHeader));
          _segments.push_back(segment);
          // Map all segments
          check_segments();
       } else {
-         LOG_ONCE_MAYBE(
-             md, Info, "Opening shared segment '{}' - FAILED",
+         LOG_ONCE_MAYBE(md, Info, "Opening shared segment '{}' - FAILED",
              std::string(segment_name));
          return;
       }

src/lib/utils/SharedMemory.cpp

@@ -79,6 +79,11 @@ struct BoostLowLevelSharedMemoryImpl {
 
    Segment create(const string &name, size_t size) {
       try {
+         bool is_test = name.find("TEST") != std::string::npos;
+         if (is_test) {
+            std::cout << "Creating shared mem for TEST..." << std::endl;
+         }
+         LOG(md, Info, "Create shared memory segment: {}", name);
          shared_memory_object block(create_only, name.c_str(), read_write);
          block.truncate(size);
          add_region(move(block));
@@ -107,7 +112,7 @@ struct BoostLowLevelSharedMemoryImpl {
                  return p.first.get_name() == segment_name;
               }),
           _mappings.end());
-      LOG(md, Diag, "Remove shared memory segment: {}",
+      LOG(md, Info, "Remove shared memory segment: {}",
           std::string(segment_name));
       return boost::interprocess::shared_memory_object::remove(
           segment_name.c_str());
@@ -117,12 +122,12 @@ struct BoostLowLevelSharedMemoryImpl {
 
 BoostLowLevelSharedMemory::BoostLowLevelSharedMemory()
     : _me(std::make_unique<BoostLowLevelSharedMemoryImpl>()) {}
-Segment
-utils::BoostLowLevelSharedMemory::create(const string &name, size_t size) {
+Segment utils::BoostLowLevelSharedMemory::create(
+    const string &name, size_t size) {
    return _me->create(name, size);
 }
-const Segment
-BoostLowLevelSharedMemory::open(const std::string &name, bool read_only) {
+const Segment BoostLowLevelSharedMemory::open(
+    const std::string &name, bool read_only) {
    return _me->open(name, read_only);
 }
 bool utils::BoostLowLevelSharedMemory::release(const string &name) {
@@ -182,8 +187,8 @@ Segment PrivateSharedMemory::create(const string &name, size_t size) {
       _segments.push_back(make_pair(true, result));
    return (Segment)result;
 }
-const Segment
-PrivateSharedMemory::open(const std::string &name, bool read_only) {
+const Segment PrivateSharedMemory::open(
+    const std::string &name, bool read_only) {
    auto result = PrivateSharedMemoryImpl::instance().open(name, read_only);
    if (result.buffer)
       _segments.push_back(make_pair(false, result));
@@ -193,8 +198,8 @@ bool PrivateSharedMemory::release(const string &name) {
    return PrivateSharedMemoryImpl::instance().release(name);
 }
 
-std::unique_ptr<LowLevelSharedMemory>
-create_low_level_shared_memory(bool shared) {
+std::unique_ptr<LowLevelSharedMemory> create_low_level_shared_memory(
+    bool shared) {
    return unique_ptr<LowLevelSharedMemory>(
        shared ? static_cast<LowLevelSharedMemory *>(
                     new BoostLowLevelSharedMemory())

src/lib/utils/ThreadDescriptor.cpp

@@ -0,0 +1,13 @@
+#include "ThreadDescriptor.h"
+
+namespace bts::threading {
+
+thread_local ThreadDescriptor this_thread_descriptor{};
+
+std::ostream &operator<<(
+    std::ostream &out, const ThreadDescriptor &td) {
+   out << td.name << "(" << td.id << ")";
+   return out;
+}
+
+} // namespace bts::threading

src/lib/utils/ThreadDescriptor.h

@@ -0,0 +1,18 @@
+#pragma once
+#include <cstdint>
+#include <ostream>
+#include <thread>
+
+namespace bts::threading {
+
+class ThreadDescriptor {
+ public:
+   const char *name;
+   std::thread::id id;
+};
+
+extern thread_local ThreadDescriptor this_thread_descriptor;
+
+std::ostream &operator<<(std::ostream &out, const ThreadDescriptor &td);
+
+} // namespace bts::threading

src/server/sparkd/cmds/Run.cpp

@@ -4,6 +4,7 @@
 #include "lib/config/protos/ports.pb.h"
 #include "lib/config/protos/quickbus_cfg.pb.h"
 #include "lib/config/protos/toggles.pb.h"
+#include "lib/domain_model/Security.pb.h"
 #include "lib/logging/logging.h"
 #include "lib/md/MBOProcessor.h"
 #include "lib/md/MBOServiceServer.h"
@@ -70,14 +71,26 @@ static void execute() {
                           ? &mbo_processor.writer()
                           : nullptr;
    bts::md::TradeSummarySignalDispatcher tss_dispatcher;
+   // auto secs = std::make_shared<std::vector<domain::Security>>(
+   //     md::cmds::Run::get_random_secs());
+   auto secs = md::cmds::Run::get_random_secs();
+   secs.clear();
+
+   std::set<bts::md::UserSymbol> sec_set;
+   for (const auto &s : secs) {
+      bts::md::UserSymbol us(
+          s.security_common.symbol, bts::api::VENUE_GLBX,
+          bts::api::SECURITYTYPE_FUTURE);
+      sec_set.insert(us);
+   }
    md::cmds::Run::run(
        mbo_writer, &tss_dispatcher, md::cmds::Run::MDType::Configured, true,
-       true);
+       true, secs);
 
    api::synthetic::SyntheticSecurityService synthetic_service(sc);
    bts::md::MDClient mdclient(
        &utils::Dispatcher::background(), &synthetic_service,
-       &mbo_processor.reader());
+       &mbo_processor.reader(), &sec_set);
 
    bts::md::QueryServerPtr md_query_server(
        bts::md::run_query_server(sc, mdclient, stats));

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

@@ -1,17 +1,28 @@
 #include "Run.h"
 #include "Cleanup.h"
 #include "lib/config/protos/market_data.pb.h"
+#include "lib/api/domain/Currency.pb.h"
+#include "lib/api/domain/SecurityType.pb.h"
+#include "lib/api/domain/Venue.pb.h"
+#include "lib/domain/FutureSecurity.h"
+#include "lib/domain/OptionSecurity.h"
+#include "lib/domain/Price.h"
+#include "lib/domain/Security.h"
+#include "lib/domain/SecurityState.h"
+#include "lib/domain/Venue.h"
 #include "lib/logging/logging.h"
 #include "lib/md/UserSymbol.h"
 #include "lib/md/relay/RelayRecorder.h"
 #include "lib/os/os.h"
 #include "lib/utils/thread_manager.h"
 #include "server/sparkd/exchange_sim/ExchangeSim.h"
+#include "server/sparkd/md/MarketDataWriter.h"
 #include "server/sparkd/md/platforms/IMDPlatform.h"
 #include "server/sparkd/md/platforms/emumba/EmumbaMain.h"
 #include "server/sparkd/md/platforms/ice_onixs/ICEMain.h"
 #include "server/sparkd/md/platforms/mdp3_onixs/MDP3Main.h"
 #include <string>
+#include <range/v3/view.hpp>
 
 namespace bts::cmds::md::cmds {
 
@@ -54,18 +65,90 @@ static void run_replay() {
        replay_opts.fallback());
 }
 
-static void run_onixs(
+struct RunData::PImpl {
+   // TODO
+   bts::md::MDP3Main *mdp3_main;
+   bts::md::ICEMain *ice_main;
+};
+
+RunData::RunData() : RunData(nullptr, nullptr) {}
+
+RunData::RunData(bts::md::MDP3Main *mdp3_main, bts::md::ICEMain *ice_main)
+    : me_(std::make_unique<PImpl>()) {
+   me_->mdp3_main = mdp3_main;
+   me_->ice_main = ice_main;
+}
+
+RunData::~RunData() = default;
+
+bts::md::MDP3Main *RunData::mdp3_main() {
+   if (!me_) {
+      return nullptr;
+   }
+   return me_->mdp3_main;
+}
+
+bts::md::ICEMain *RunData::ice_main() {
+   if (!me_) {
+      return nullptr;
+   }
+
+   return me_->ice_main;
+}
+
+std::vector<domain::Security> Run::get_random_secs() {
+   std::vector<domain::Security> secs;
+   domain::Security &s = secs.emplace_back();
+   s.security_common.venue = api::domain::Venue::GLBX;
+   s.security_common.symbol = "TEST";
+   s.security_common.asset = "TEST";
+   s.security_common.currency = api::domain::Currency::USD;
+   s.security_common.desc = "TESTH2 100 C";
+   s.security_common.group = "TEST";
+   s.security_common.id.value() = 8675309;
+   s.security_common.cash_per_point = 1.0;
+   s.security_common.implied_matching = false;
+   s.security_common.type = api::domain::SecurityType::OPTION;
+   s.security_common.min_price_increment = domain::Price::from_double(0.25);
+   s.security_common.cme_security_type = "OPT";
+   s.security = domain::OptionSecurity();
+   auto &opt = std::get<domain::OptionSecurity>(s.security);
+   // 03/18/2022 begins 1647561600 nanos from epoch
+   opt.expiration = core::TimeStamp(1647561600);
+   opt.strike = domain::Price::from_double(100.0);
+   opt.call_put = domain::CallPut::Call;
+
+   domain::Security sf = s;
+   sf.security_common.desc = "TESTH2";
+   sf.security_common.cme_security_type = "FUT";
+   sf.security = domain::FutureSecurity();
+   sf.security_common.id.value() += 1;
+   auto &fut = std::get<domain::FutureSecurity>(sf.security);
+   fut.expiration = opt.expiration;
+   secs.push_back(sf);
+   return secs;
+}
+
+static RunData run_onixs(
     bts::md::MBOWriter *mbo_writer,
-    bts::md::TradeSummarySignalDispatcher *tss_dispatcher) {
+    bts::md::TradeSummarySignalDispatcher *tss_dispatcher,
+    const std::vector<domain::Security> /*append_secs*/ = {}) {
    LOG(md, Info, "Run::run_onixs t_id:{} t_name", os::get_thread_id(),
        get_thread_name());
+   RunData rv;
    std::vector<std::shared_ptr<bts::md::IMDPlatform>> platforms;
    // get all glbx symbols
    std::vector<bts::md::UserSymbol> cme_symbols;
    get_system_md_symbols(
        api::VENUE_GLBX, api::SECURITYTYPE_UNSET, &cme_symbols);
+   bts::md::MDP3Main *main = nullptr;
    if (cme_symbols.size() > 0) {
-      auto *main = new bts::md::MDP3Main();
+      main = new bts::md::MDP3Main();
+      LOG(md, Info, "MDP3Main subscribing to symbols: {}",
+          fmt::to_string(fmt::join(
+              ranges::views::transform(
+                  cme_symbols, [](const auto &s) { return s.to_string(); }),
+              ", ")));
       main->start(cme_symbols, mbo_writer, tss_dispatcher);
       platforms.emplace_back(main);
    }
@@ -88,6 +171,24 @@ static void run_onixs(
    heartbeat_timer.start();
 
    disp.run();
+   // bts::md::MarketDataWriter md_writer(main->index(), main->producer());
+   // std::vector<domain::Security> secs = get_random_secs();
+   // if (append_secs != nullptr) {
+   //    std::vector<const domain::Security *> secs_out;
+   //    secs_out.resize(append_secs->size());
+   //    for (const auto &sec : *append_secs) {
+   //       secs_out.push_back(&sec);
+   //    }
+   //    md_writer.on_securities(secs_out);
+   //    for (const auto &sec : *append_secs) {
+   //       md_writer.on_security_state(domain::SecurityState{
+   //           domain::SecurityState::State::OPEN, sec.security_common.id});
+   //       domain::Book bk = domain::Book::empty();
+   //       bk.set_security_id(sec.security_common.id);
+   //       md_writer.on_book(bk);
+   //    }
+   // }
+   return rv;
 }
 
 static void run_emumba() {
@@ -102,7 +203,8 @@ void Run::run(
     bts::md::TradeSummarySignalDispatcher *tss_dispatcher,
     Run::MDType type,
     bool cleanup,
-    bool run_on_md_thread) {
+    bool run_on_md_thread,
+    const std::vector<domain::Security> append_secs) {
    md::cmds::Cleanup::execute();
    LOG(md, Info, "Run::run cleanup:{} run_on_md_thread:{} t_id:{} t_name:{}",
        cleanup, run_on_md_thread, os::get_thread_id(), get_thread_name());
@@ -112,7 +214,7 @@ void Run::run(
       return;
    }
 
-   auto impl = [&mbo_writer, &tss_dispatcher, &type, &cleanup]() {
+   auto impl = [&mbo_writer, &tss_dispatcher, &type, &cleanup, &append_secs]() {
       LOG(md, Info, "Run::impl lambda t_id:{} t_name", os::get_thread_id(),
           get_thread_name());
       if (cleanup)
@@ -129,7 +231,7 @@ void Run::run(
          run_replay();
          break;
       case Run::MDType::Onixs:
-         run_onixs(mbo_writer, tss_dispatcher);
+         run_onixs(mbo_writer, tss_dispatcher, append_secs);
          break;
       default:
          bts_assert_msg(false, "Unexpected md type");
@@ -148,9 +250,13 @@ void Run::run(
          threading::make_thread("Emumba", []() { run_emumba(); }).detach();
       }
       if (bts::config::get<bts::md::MarketDataConfig>().use_sim()) {
-         threading::make_thread("ExchangeSimGLBX", [&mbo_writer]() {
-            sim::start_exchange_sim(api::Venue::VENUE_SIM, mbo_writer);
-         }).detach();
+         threading::make_thread(
+             "ExchangeSimGLBX",
+             [&mbo_writer, append_secs]() {
+                sim::start_exchange_sim(
+                    api::Venue::VENUE_SIM, mbo_writer, append_secs);
+             })
+             .detach();
       }
       boot_sleep();
    } else {

src/server/sparkd/cmds/md/cmds/Run.h

@@ -1,20 +1,39 @@
 #pragma once
 #include "lib/md/MBOProcessor.h"
+#include "lib/domain/Security.h"
 #include "server/sparkd/md/platforms/mdp3_onixs/TradeSummarySignalDispatcher.h"
 #include <CLI11/CLI11.hpp>
 
+namespace bts::md {
+class MDP3Main;
+class ICEMain;
+} // namespace bts::md
+
 namespace bts::cmds::md::cmds {
 
+struct RunData {
+   RunData();
+   RunData(bts::md::MDP3Main *mdp3_main, bts::md::ICEMain *ice_main);
+   RunData(const RunData &) = delete;
+   RunData(RunData &&) noexcept = default;
+   RunData &operator=(const RunData &) = delete;
+   RunData &operator=(RunData &&) noexcept = default;
+   ~RunData();
+
+   bts::md::MDP3Main *mdp3_main();
+   bts::md::ICEMain *ice_main();
+
+ private:
+   struct PImpl;
+   std::unique_ptr<PImpl> me_;
+};
+
 /**
  * Start market data feed, and store up-to-date snapshots of mbo, and mbp in
  * shared memory.
  */
 struct Run : public CLI::App {
-   enum class MDType {
-      Configured,
-      Replay,
-      Onixs
-   };
+   enum class MDType { Configured, Replay, Onixs };
 
    Run();
 
@@ -26,7 +45,10 @@ struct Run : public CLI::App {
        bts::md::TradeSummarySignalDispatcher *,
        MDType type,
        bool cleanup,
-       bool run_on_md_thread);
+       bool run_on_md_thread,
+       const std::vector<domain::Security> append_secs = {});
+   
+   static std::vector<domain::Security> get_random_secs();
 };
 
 } // namespace bts::cmds::md::cmds

src/server/sparkd/exchange_sim/ExchangeSim.cpp

@@ -1,35 +1,37 @@
 #include "ExchangeSimApplication.h"
-
-#include <boost/algorithm/string.hpp>
-#include <boost/algorithm/string/replace.hpp>
-#include <quickfix/FileStore.h>
-#include <quickfix/SessionSettings.h>
-#include <quickfix/SocketAcceptor.h>
-
-#include <fstream>
-#include <iostream>
-#include <string>
-
+#include "MarketDataPlayer.h"
 #include "lib/config/config.h"
 #include "lib/config/protos/exchange_sim_cfg.pb.h"
 #include "lib/config/protos/ports.pb.h"
-
+#include "lib/core/ProtoPrinter.h"
+#include "lib/domain/FutureSecurity.h"
+#include "lib/domain/Security.h"
+#include "lib/domain/SecurityID.h"
 #include "lib/domain_model/Security.pb.h"
 #include "lib/logging/logging.h"
 #include "lib/os/os.h"
-
 #include "lib/domain_model/Security.pb.h"
 #include "lib/md/IntPrice.h"
 #include "lib/md/MBOProcessor.h"
+#include "lib/md/protos/co2.pb.h"
 #include "lib/md/relay/RelayClient.h"
 #include "lib/md/relay/RelayRecorder.h"
+#include "lib/os/os.h"
 #include "lib/utils/bts_chrono.h"
+#include <boost/algorithm/string.hpp>
+#include <boost/algorithm/string/replace.hpp>
+#include <fstream>
+#include <iostream>
 #include <quickfix/FileLog.h>
-
+#include <quickfix/FileStore.h>
+#include <quickfix/SessionSettings.h>
+#include <quickfix/SocketAcceptor.h>
 #include <set>
+#include <string>
+#include <variant>
+
+namespace bts::sim {
 
-namespace bts {
-namespace sim {
 auto cfg = "[DEFAULT]\n"
            "ConnectionType=acceptor\n"
            "SocketAcceptPort=$EXCHANGE_SIM_FIX_PORT\n"
@@ -47,93 +49,78 @@ auto cfg = "[DEFAULT]\n"
            "ValidateUserDefinedFields=N\n"
            "ValidateFieldsOutOfOrder=N\n";
 
-md::PriceSize coerce_precision(const co2::PriceSizeMessage &p) {
-   md::IntPrice raw = {p.mantissa(), (int8_t)p.exponent()};
-   auto price = md::IntPrice_from_double<ex_sim::precision>(raw.price());
-   return {price, p.size()};
+static bts::api::Security to_api_sec(const bts::domain::Security &sec) {
+   bts::api::Security sec_out;
+   sec_out.set_id(static_cast<std::uint32_t>(sec.security_common.id.value()));
+   sec_out.set_exchange(bts::api::VENUE_GLBX);
+   *sec_out.mutable_glbx()->mutable_security_desc() = sec.security_common.desc;
+   *sec_out.mutable_symbol() = sec.security_common.symbol;
+   *sec_out.mutable_contract_spec() = bts::api::ContractSpecification();
+   auto &spec = *sec_out.mutable_contract_spec();
+   spec.set_cash_per_point(sec.security_common.cash_per_point);
+   spec.set_high_tick(sec.security_common.min_price_increment_ammount);
+   spec.set_low_tick(sec.security_common.min_price_increment_ammount);
+   spec.set_main_fraction(100);
+   if (std::holds_alternative<bts::domain::FutureSecurity>(sec.security)) {
+      // const auto &fut = std::get<domain::FutureSecurity>(sec.security);
+      *sec_out.mutable_future() = bts::api::FutureContract();
+      auto &fut_out = *sec_out.mutable_future();
+      fut_out.set_expiration_day(18);
+      fut_out.set_expiration_year(2022);
+      fut_out.set_expiration_month(3);
+   } else if (std::holds_alternative<bts::domain::OptionSecurity>(
+                  sec.security)) {
+      *sec_out.mutable_option() = bts::api::OptionContract();
+      auto &opt_out = *sec_out.mutable_future();
+      opt_out.set_expiration_day(18);
+      opt_out.set_expiration_year(2022);
+      opt_out.set_expiration_month(3);
+   }
+   return sec_out;
 }
 
-struct MDPlayer : md::IRelayWriter {
-   MDPlayer(
-       IMDPlayerObserver &obs,
-       std::map<uint32_t, md::Security> &known_securities)
-       : obs(obs), known_securities(known_securities),
-         forward_implied(config::get<config::ExchangeSim>().forward_implied()) {
+static co2::ChannelUpdate to_channel_update(const bts::api::Security &api_sec) {
+   co2::ChannelUpdate cup;
+   *cup.mutable_security() = api_sec;
+   *cup.mutable_direct() = co2::BookMessage();
+   co2::BookMessage &dbook = *cup.mutable_direct();
+   auto &bid = *dbook.mutable_bids()->Add();
+   bid.set_mantissa(99999);
+   bid.set_exponent(8);
+   bid.set_size(3);
+   auto &ask = *dbook.mutable_asks()->Add();
+   ask.set_mantissa(100000);
+   ask.set_exponent(8);
+   ask.set_size(5);
+   *cup.mutable_implied() = dbook;
+   cup.set_key(api_sec.id());
+
+   auto &misc = *cup.mutable_misc();
+   auto &settlement = *misc.mutable_settlement();
+   settlement.set_exponent(8);
+   settlement.set_mantissa(100001);
+   // misc.set_settlement_date();
+   return cup;
+   // std::stringstream ss;
+   // ss << core::ProtoPrinterFactory::create(&api_sec);
+   // LOG(md, Info, "Appending sec finally:\n{}", ss.str());
+   // player.on_update(cup);
+}
+
+static co2::MDReply to_md_reply(
+    const std::vector<bts::api::Security> &api_secs) {
+   co2::MDReply reply;
+   for (const auto &api_sec : api_secs) {
+      auto &channel = *reply.add_channels();
+      *channel.add_updates() = to_channel_update(api_sec);
    }
+   return reply;
+}
 
-   void write(const co2::MDReply &data, bool *) override {
-      for (auto const &ch : data.channels())
-         for (auto const &cr : ch.updates()) {
-            auto it = known_securities.find(cr.key());
-
-            api::Security const *sec_ptr(nullptr);
-            if (it == known_securities.end()) {
-               bts_assert(cr.has_security());
-               sec_ptr = &cr.security();
-               known_securities[cr.key()] = *sec_ptr;
-               obs.on_new_security(*sec_ptr);
-            } else
-               sec_ptr = &it->second;
-
-            auto const &book = cr.direct();
-            md::PriceSize bids[md::COCONUT_MAX_DEPTH],
-                asks[md::COCONUT_MAX_DEPTH];
-
-            size_t i = 0;
-            for (auto &b : book.bids())
-               bids[i++] = md::PriceSize(
-                   {b.mantissa(), (int8_t)b.exponent()}, b.size());
-
-            size_t n = 0;
-            for (auto &a : book.asks())
-               asks[n++] = md::PriceSize(
-                   {a.mantissa(), (int8_t)a.exponent()}, a.size());
-
-            std::array<md::PriceSize, md::COCONUT_MAX_IMPLIED_DEPTH>
-                implied_bids, implied_asks;
-            bool has_implied = false;
-            {
-               size_t x = 0;
-               for (const auto &b : cr.implied().bids()) {
-                  implied_bids[x] = coerce_precision(b);
-                  x += 1;
-                  has_implied = true;
-               }
-            }
-            {
-               size_t x = 0;
-               for (const auto &b : cr.implied().asks()) {
-                  implied_asks[x] = coerce_precision(b);
-                  x += 1;
-                  has_implied = true;
-               }
-            }
-
-            if (has_implied && forward_implied) {
-               obs.on_depth(*sec_ptr, bids, asks, &implied_bids, &implied_asks);
-            } else {
-               obs.on_depth(*sec_ptr, bids, asks, nullptr, nullptr);
-            }
-
-            if (cr.has_misc()) {
-               const auto &s = cr.misc();
-               obs.on_high(*sec_ptr, s.session_high());
-               obs.on_low(*sec_ptr, s.session_low());
-               obs.on_open(*sec_ptr, s.open_price());
-               obs.on_close(*sec_ptr, s.close_price());
-               if (cr.misc().has_settlement())
-                  obs.on_settlement(
-                      *sec_ptr, {s.settlement().mantissa(),
-                                 (int8_t)s.settlement().exponent()});
-            }
-         }
-   }
-   IMDPlayerObserver &obs;
-   std::map<uint32_t, md::Security> &known_securities;
-   bool forward_implied;
-};
-
-int start_exchange_sim(bts::api::Venue venue, md::MBOWriter *mbo) {
+int start_exchange_sim(
+    bts::api::Venue venue,
+    md::MBOWriter *mbo,
+    const std::vector<bts::domain::Security> append_secs) {
    try {
       std::string t_name;
       os::get_thread_name(&t_name);
@@ -161,6 +148,8 @@ int start_exchange_sim(bts::api::Venue venue, md::MBOWriter *mbo) {
       std::istringstream str{str_settings};
       FIX::SessionSettings settings(str);
 
+      LOG(md, Info, "Append secs was passed to exchange sim: {}",
+          !append_secs.empty());
       ex_sim::ExchangeSimApplication application(venue, mbo);
       FIX::FileStoreFactory storeFactory(settings);
       FIX::FileLogFactory logFactory(settings);
@@ -171,21 +160,36 @@ int start_exchange_sim(bts::api::Venue venue, md::MBOWriter *mbo) {
 
       std::map<uint32_t, md::Security> known_securities;
       const auto &conf = config::get<config::ExchangeSim>();
+      std::vector<co2::MDReply> prepend_msgs;
+      if (!append_secs.empty()) {
+         LOG(md, Info, "Appending secs to exchange sim...");
+         std::vector<bts::api::Security> append_api_secs;
+         append_api_secs.reserve(append_secs.size());
+         for (const auto &sec : append_secs) {
+            append_api_secs.push_back(to_api_sec(sec));
+         }
+         prepend_msgs.push_back(to_md_reply(append_api_secs));
+      }
       while (true) {
-         MDPlayer player(application, known_securities);
+         MarketDataPlayer player(application, known_securities);
          switch (conf.source_case()) {
          case config::ExchangeSim::SOURCE_NOT_SET:
          case config::ExchangeSim::kReplayUrl:
+            LOG(md, Info, "Replaying url w/ fallback: {}", conf.replay_url());
             md::replay_url_with_fallback(
                 conf.replay_url(), player,
-                config::get<md::RelayOptions>().naggle_ms());
+                config::get<md::RelayOptions>().naggle_ms(), true,
+                prepend_msgs);
             break;
          case config::ExchangeSim::kReplayFile:
+            LOG(md, Info, "Replaying file: {}", conf.replay_file());
             md::replay_file(
                 conf.replay_file(), player,
-                config::get<md::RelayOptions>().naggle_ms(), true);
+                config::get<md::RelayOptions>().naggle_ms(), true,
+                prepend_msgs);
             break;
          case config::ExchangeSim::kLocalRelay: {
+            LOG(md, Info, "Local relay");
             md::local_relay(player);
          } break;
          }
@@ -197,5 +201,5 @@ int start_exchange_sim(bts::api::Venue venue, md::MBOWriter *mbo) {
       return 1;
    }
 }
-} // namespace sim
-} // namespace bts
+
+} // namespace bts::sim

src/server/sparkd/exchange_sim/ExchangeSim.h

@@ -1,7 +1,9 @@
 #pragma once
 
 #include "lib/domain_model/Security.pb.h"
+#include "lib/domain/Security.h"
 #include <memory>
+#include <vector>
 
 namespace bts {
 namespace md {
@@ -9,7 +11,10 @@ class MBOWriter;
 }
 namespace sim {
 
-int start_exchange_sim(bts::api::Venue, md::MBOWriter * = nullptr);
+int start_exchange_sim(
+    bts::api::Venue,
+    md::MBOWriter * = nullptr,
+    const std::vector<bts::domain::Security> append_secs = {});
 
 }
 } // namespace bts

src/server/sparkd/exchange_sim/ExchangeSimApplication.cpp

@@ -1,4 +1,5 @@
 #include "ExchangeSimApplication.h"
+#include "lib/core/ProtoPrinter.h"
 #include "lib/md/SecurityDB.h"
 
 #include "IDGenerator.h"
@@ -79,8 +80,8 @@ FIX::OrdType convert(api::orderman::NativeOrderType type) {
    }
 }
 
-FIX::ManualOrderIndicator
-convert(const api::orderman::ManualOrderIndicator &moi) {
+FIX::ManualOrderIndicator convert(
+    const api::orderman::ManualOrderIndicator &moi) {
    switch (moi) {
    case api::orderman::MANUAL_ORDER_INDICATOR_YES:
       return FIX::ManualOrderIndicator(true);
@@ -89,8 +90,8 @@ convert(const api::orderman::ManualOrderIndicator &moi) {
    }
 }
 
-api::orderman::ManualOrderIndicator
-convert(const FIX::ManualOrderIndicator &moi) {
+api::orderman::ManualOrderIndicator convert(
+    const FIX::ManualOrderIndicator &moi) {
    if (moi.getValue())
       return api::orderman::MANUAL_ORDER_INDICATOR_YES;
    else
@@ -159,17 +160,19 @@ void ExchangeSimApplication::onCreate(const FIX::SessionID &) {}
 void ExchangeSimApplication::onLogon(const FIX::SessionID &) {}
 void ExchangeSimApplication::onLogout(const FIX::SessionID &) {}
 void ExchangeSimApplication::toAdmin(FIX::Message &, const FIX::SessionID &) {}
-void ExchangeSimApplication::toApp(
-    FIX::Message &, const FIX::SessionID &) EXCEPT(FIX::DoNotSend) {}
-void ExchangeSimApplication::
-    fromAdmin(const FIX::Message &, const FIX::SessionID &) EXCEPT(
+void ExchangeSimApplication::toApp(FIX::Message &, const FIX::SessionID &)
+    EXCEPT(FIX::DoNotSend) {}
+void ExchangeSimApplication::fromAdmin(
+    const FIX::Message &, const FIX::SessionID &)
+    EXCEPT(
         FIX::FieldNotFound,
         FIX::IncorrectDataFormat,
         FIX::IncorrectTagValue,
         FIX::RejectLogon) {}
 
-void ExchangeSimApplication::
-    fromApp(const FIX::Message &message, const FIX::SessionID &sessionID) EXCEPT(
+void ExchangeSimApplication::fromApp(
+    const FIX::Message &message, const FIX::SessionID &sessionID)
+    EXCEPT(
         FIX::FieldNotFound,
         FIX::IncorrectDataFormat,
         FIX::IncorrectTagValue,
@@ -607,7 +610,12 @@ void ExchangeSimApplication::on_new_security(api::Security const &s) {
          negate_id(&s2);
          md::convert_security(s2, fs, true);
          md::MDProducerHandle handle = me->producer.add_security(fs);
+         // auto sec_display_str =
+         // bts::core::ProtoPrinterFactory::to_string(&s);
+         std::string sec_display_str(s.symbol());
          auto sec_desc = securities::descriptor(s2);
+         LOG(md, Info, "on_new_security. symbol: {}; descriptor: {}",
+             sec_display_str, sec_desc);
          me->md_producer_handles.insert(make_pair(sec_desc, handle));
          me->md_consumer_recs.insert(make_pair(
              securities::descriptor(s),
@@ -665,7 +673,7 @@ void ExchangeSimApplication::on_close(
 }
 
 void ExchangeSimApplication::on_depth(
-    md::Security const &sec,
+    bts::api::Security const &sec,
     md::PriceSize bids[md::COCONUT_MAX_DEPTH],
     md::PriceSize asks[md::COCONUT_MAX_DEPTH],
     std::array<md::PriceSize, md::COCONUT_MAX_IMPLIED_DEPTH> *implied_bids,
@@ -679,9 +687,12 @@ void ExchangeSimApplication::on_depth(
    if (!me->do_merge_with_md)
       return;
 
-   auto it = me->md_consumer_recs.find(securities::descriptor(sec));
-   if (it == me->md_consumer_recs.end())
+   auto desc = securities::descriptor(sec);
+   auto it = me->md_consumer_recs.find(desc);
+   if (it == me->md_consumer_recs.end()) {
+      // LOG(md, Info, "Did not find desc {} in md_consumer_recs", desc);
       return;
+   }
 
    auto &matcher = me->matchers[get_sim_sec_desc(sec)];
    Book previous_md_book = matcher.get_book();

src/server/sparkd/exchange_sim/ExchangeSimApplication.h

@@ -124,7 +124,7 @@ class ExchangeSimApplication : public sim::IMDPlayerObserver,
    void on_close(md::Security const &sec, md::Price price) override;
    void on_settlement(md::Security const &, md::IntPrice) override;
    void on_depth(
-       md::Security const &s,
+       bts::api::Security const &s,
        md::PriceSize bids[md::COCONUT_MAX_DEPTH],
        md::PriceSize asks[md::COCONUT_MAX_DEPTH],
        std::array<md::PriceSize, md::COCONUT_MAX_IMPLIED_DEPTH> *implied_bids,

src/server/sparkd/exchange_sim/MarketDataPlayer.cpp

@@ -0,0 +1,93 @@
+#include "MarketDataPlayer.h"
+#include "ExchangeSimApplication.h"
+#include "lib/config/protos/ExchangeSimCfg.pb.h"
+
+namespace bts::sim {
+
+static md::PriceSize coerce_precision(const co2::PriceSizeMessage &p) {
+   md::IntPrice raw = {p.mantissa(), (int8_t)p.exponent()};
+   auto price = md::IntPrice_from_double<ex_sim::precision>(raw.price());
+   return {price, p.size()};
+}
+
+MarketDataPlayer::MarketDataPlayer(
+    IMDPlayerObserver &obs, std::map<uint32_t, md::Security> &known_securities)
+    : obs(obs), known_securities(known_securities),
+      forward_implied(config::get<config::ExchangeSim>().forward_implied()) {}
+
+const api::Security *MarketDataPlayer::get_security(
+    const co2::ChannelUpdate &cr) {
+   auto it = known_securities.find(cr.key());
+   if (it == known_securities.end()) {
+      bts_assert(cr.has_security());
+      known_securities[cr.key()] = cr.security();
+      obs.on_new_security(cr.security());
+      return &cr.security();
+   } else {
+      return &it->second;
+   }
+}
+
+void MarketDataPlayer::on_update(const bts::co2::ChannelUpdate &cr) {
+   const api::Security *sec_ptr = get_security(cr);
+   auto const &book = cr.direct();
+   md::PriceSize bids[md::COCONUT_MAX_DEPTH], asks[md::COCONUT_MAX_DEPTH];
+
+   size_t i = 0;
+   for (auto &b : book.bids()) {
+      bids[i++] = md::PriceSize({b.mantissa(), (int8_t)b.exponent()}, b.size());
+   }
+
+   size_t n = 0;
+   for (auto &a : book.asks()) {
+      asks[n++] = md::PriceSize({a.mantissa(), (int8_t)a.exponent()}, a.size());
+   }
+
+   std::array<md::PriceSize, md::COCONUT_MAX_IMPLIED_DEPTH> implied_bids,
+       implied_asks;
+   bool has_implied = false;
+   {
+      size_t x = 0;
+      for (const auto &b : cr.implied().bids()) {
+         implied_bids[x] = coerce_precision(b);
+         x += 1;
+         has_implied = true;
+      }
+   }
+   {
+      size_t x = 0;
+      for (const auto &b : cr.implied().asks()) {
+         implied_asks[x] = coerce_precision(b);
+         x += 1;
+         has_implied = true;
+      }
+   }
+
+   if (has_implied && forward_implied) {
+      obs.on_depth(*sec_ptr, bids, asks, &implied_bids, &implied_asks);
+   } else {
+      obs.on_depth(*sec_ptr, bids, asks, nullptr, nullptr);
+   }
+
+   if (cr.has_misc()) {
+      const auto &s = cr.misc();
+      obs.on_high(*sec_ptr, s.session_high());
+      obs.on_low(*sec_ptr, s.session_low());
+      obs.on_open(*sec_ptr, s.open_price());
+      obs.on_close(*sec_ptr, s.close_price());
+      if (cr.misc().has_settlement())
+         obs.on_settlement(
+             *sec_ptr,
+             {s.settlement().mantissa(), (int8_t)s.settlement().exponent()});
+   }
+}
+
+void MarketDataPlayer::write(const co2::MDReply &data, bool *) {
+   for (auto const &ch : data.channels()) {
+      for (auto const &cr : ch.updates()) {
+         on_update(cr);
+      }
+   }
+}
+
+} // namespace bts::sim

src/server/sparkd/exchange_sim/MarketDataPlayer.h

@@ -0,0 +1,23 @@
+#pragma once
+#include "IMDPlayerObserver.h"
+#include "lib/md/relay/RelayClient.h"
+
+namespace bts::sim {
+
+struct MarketDataPlayer : md::IRelayWriter {
+   MarketDataPlayer(
+       IMDPlayerObserver &obs,
+       std::map<uint32_t, md::Security> &known_securities);
+
+   const api::Security *get_security(const co2::ChannelUpdate &cr);
+
+   void on_update(const bts::co2::ChannelUpdate &cr);
+
+   void write(const co2::MDReply &data, bool *) override;
+
+   IMDPlayerObserver &obs;
+   std::map<uint32_t, md::Security> &known_securities;
+   bool forward_implied;
+};
+
+} // namespace bts::sim

src/server/sparkd/md/MarketDataWriter.cpp

@@ -0,0 +1,255 @@
+#include "MarketDataWriter.h"
+#include "lib/domain/AggressorSide.h"
+#include "lib/domain/Book.h"
+#include "lib/domain/FutureSecurity.h"
+#include "lib/domain/IndexSecurity.h"
+#include "lib/domain/OptionSecurity.h"
+#include "lib/domain/RFQ.h"
+#include "lib/domain/Security.h"
+#include "lib/domain/SecurityMarketDataStatistics.h"
+#include "lib/domain/SecurityState.h"
+#include "lib/domain/Side.h"
+#include "lib/domain/SpreadSecurity.h"
+#include "lib/domain/TradeRecord.h"
+#include "lib/domain/Venue.h"
+#include "lib/domain_model/Security.pb.h"
+#include "lib/logging/logging.h"
+#include "lib/md/IntPrice.h"
+#include "lib/md/MDProducer.h"
+#include "lib/md/MarketDataError.h"
+#include "lib/md/SecurityIndex.h"
+#include "lib/md/Shm/FixedSecurity.h"
+#include "lib/md/protos/co2.pb.h"
+#include "lib/utils/TimeStamp.h"
+#include "lib/utils/bts_assert.h"
+#include <array>
+#include <boost/date_time/posix_time/posix_time.hpp>
+#include <cstdint>
+#include <memory.h>
+#include <memory>
+#include <variant>
+
+namespace bts::md {
+
+using MdArray = std::array<PriceSize, COCONUT_MAX_DEPTH>;
+using GLBXSecurityIndex = ConcurrentSecurityIndex<std::uint32_t, SecurityData>;
+   //  ConcurrentSecurityIndex<std::int32_t, SecurityMarketData>;
+
+namespace helpers {
+
+using namespace bts::md;
+
+static void clear_books(MdArray &md_side) { md_side.fill(PriceSize{}); }
+
+static void copy_books(const domain::BookSide &from, MdArray &to, bool clear = true) {
+   if (clear) {
+      clear_books(to);
+   }
+   size_t i = 0;
+   for (const domain::BookLevel &l : from) {
+      if (i < COCONUT_MAX_DEPTH) {
+         to[i++] = PriceSize{// TODO
+                             IntPrice_from_double<9>(l.price.to_double()),
+                             l.direct_quantity.value()};
+      }
+   }
+}
+
+} // namespace helpers
+
+inline api::Venue convert_venue(domain::Venue venue_in) {
+   switch (venue_in) {
+      case domain::Venue::Glbx: return api::VENUE_GLBX;
+      default:
+         return api::VENUE_SIM;
+   }
+}
+
+inline api::Venue convert_venue(api::domain::Venue venue_in) {
+   switch (venue_in) {
+      case api::domain::Venue::GLBX: return api::VENUE_GLBX;
+      default:
+         return api::VENUE_SIM;
+   }
+}
+
+inline co2::TradingStatus convert_status(domain::SecurityState::State state) {
+   switch (state) {
+   case domain::SecurityState::State::OPEN:
+      return co2::TRADINGSTATUS_NORMAL;
+   case domain::SecurityState::State::PREOPEN:
+      return co2::TRADINGSTATUS_PRE_OPEN;
+   }
+   return co2::TRADINGSTATUS_UNKNOWN;
+}
+
+inline co2::AggressorSide convert_agg_side(domain::AggressorSide agg_side) {
+   switch (agg_side.value) {
+   case domain::AggressorSide::Buy:
+      return co2::AggressorSide::AGGRESSOR_SIDE_BUY;
+   case domain::AggressorSide::Sell:
+      return co2::AggressorSide::AGGRESSOR_SIDE_SELL;
+   case domain::AggressorSide::None:
+      return co2::AggressorSide::AGGRESSOR_SIDE_NONE;
+   }
+   return co2::AggressorSide::AGGRESSOR_SIDE_NONE;
+}
+
+static bts::md::IntPrice convert_to_intprice(domain::Price price) {
+   return IntPrice_from_double<9>(price.to_double());
+}
+
+inline SecurityMarketData *get_security(
+    GLBXSecurityIndex &sec_map, domain::SecurityID id) {
+   auto int_id = static_cast<int>(id.value());
+   auto *rv = sec_map.get_security(int_id);
+   bts_assert(rv != nullptr);
+   return rv;
+}
+
+struct MarketDataWriter::PImpl {
+   GLBXSecurityIndex &index;
+   MDProducer &md_producer;
+   MdArray bids;
+   MdArray asks;
+   explicit PImpl(GLBXSecurityIndex &index_a, MDProducer &md_producer_a)
+       : index(index_a), md_producer(md_producer_a) {}
+};
+
+MarketDataWriter::MarketDataWriter(
+    GLBXSecurityIndex &index, MDProducer &md_producer)
+    : me(std::make_unique<PImpl>(index, md_producer)) {}
+
+void MarketDataWriter::on_securities(
+    const std::vector<domain::Security const *> &complete_security_list) {
+   for (domain::Security const *s : complete_security_list) {
+      if (std::holds_alternative<domain::IndexSecurity>(s->security)) {
+         _add_index(*s);
+      } else if (std::holds_alternative<domain::FutureSecurity>(s->security)) {
+         _add_future(*s);
+      }
+   }
+}
+
+void MarketDataWriter::on_security_state(const domain::SecurityState &state) {
+   auto *sec_data = get_security(me->index, state.security_id);
+   if (sec_data == nullptr) {
+      LOG(md, Error,
+          "MarketDataWriter::on_security_statistics could not find id: {}",
+          state.security_id.value());
+      return;
+   }
+   auto co2_status = convert_status(state.state);
+   sec_data->handle.update_exchange_status(utils::TimeStamp::now(), co2_status);
+}
+
+void MarketDataWriter::on_security_statistics(
+    const domain::SecurityMarketDataStatistics &security_stats) {
+   auto *sec_data = get_security(me->index, security_stats.security_id);
+   if (sec_data == nullptr) {
+      LOG(md, Error,
+          "MarketDataWriter::on_security_statistics could not find id: {}",
+          security_stats.security_id.value());
+      return;
+   }
+   auto now = utils::TimeStamp::now();
+   sec_data->handle.update_session_low(now, security_stats.low.to_double());
+   sec_data->handle.update_session_high(now, security_stats.high.to_double());
+   bts::datetime::Date date_temp = {};
+   sec_data->handle.update_opening(
+       now, security_stats.open.to_double(), date_temp);
+   sec_data->handle.update_settlement(
+       now, IntPrice_from_double<8>(security_stats.settle.to_double()),
+       date_temp);
+}
+
+void MarketDataWriter::on_book(const domain::Book &book) {
+   auto *sec_data = get_security(me->index, book.security_id());
+   if (sec_data == nullptr) {
+      LOG(md, Error, "MarketDataWriter::on_book could not find id: {}",
+          book.security_id());
+      return;
+   }
+   helpers::copy_books(book[domain::Side::Buy], me->bids);
+   helpers::copy_books(book[domain::Side::Sell], me->asks);
+   sec_data->handle.update_books(
+       utils::TimeStamp::now(), me->bids, me->asks, {}, {});
+}
+
+void MarketDataWriter::on_trades(
+    const std::vector<domain::TradeRecord const *> &latest_trades) {
+   for (domain::TradeRecord const *t : latest_trades) {
+      auto *sec_data = get_security(me->index, t->sec_id);
+      auto now = utils::TimeStamp::now();
+      if (sec_data != nullptr) {
+         bts::datetime::LLDateTime lldt_temp = {};
+         sec_data->handle.add_trade(
+             now,
+             PriceSize(
+                 convert_to_intprice(t->trade_price),
+                 t->trade_quantity.value()),
+             lldt_temp, convert_agg_side(t->agg_side));
+      }
+   }
+}
+
+void MarketDataWriter::on_rfqs(
+    const std::vector<domain::RFQ const *> & /*unused*/) {
+   // TODO
+}
+
+void MarketDataWriter::on_error(const MarketDataError &error) {
+   LOG(md, Error, "MarketDataWriter::on_error error: {}", error);
+   // TODO: clear out books?
+}
+
+MarketDataWriter::~MarketDataWriter() = default;
+
+// private functions
+void MarketDataWriter::_add_index(const domain::Security &sec) {
+   FixedStockSecurity fsec = {};
+   _copy_security_common(sec, fsec);
+   _add_security(fsec);
+}
+
+void MarketDataWriter::_add_future(const domain::Security &sec) {
+   FixedFutureSecurity fsec = {};
+   _copy_security_common(sec, fsec);
+   _add_security(fsec);
+   const auto &future_sec = std::get<domain::FutureSecurity>(sec.security);
+   auto boost_ts =
+       boost::posix_time::from_time_t(future_sec.expiration.nanos());
+   fsec.expiration_year = boost_ts.date().year();
+   fsec.expiration_month = boost_ts.date().month().as_number();
+   fsec.expiration_day = boost_ts.date().day().as_number();
+}
+
+void MarketDataWriter::_copy_security_common(
+    const domain::Security &from, FixedSecurity &to) const {
+   to.numeric_id = from.security_common.id.value();
+   to.symbol(from.security_common.symbol);
+   to.security_desc(from.security_common.desc);
+   to.exchange = convert_venue(from.security_common.venue);
+   to.cash_per_point = from.security_common.cash_per_point;
+   to.low_tick = from.security_common.min_price_increment_ammount;
+   to.high_tick = from.security_common.min_price_increment_ammount;
+   // these are not filled in
+   // TODO
+   // fsec.display_factor = def.displayFactor().exponent() +
+   //                       floorlog10(def.displayFactor().mantissa());
+
+   // def.mainFraction(fsec.cme_main_fraction);
+   // def.subFraction(fsec.cme_sub_fraction);
+   // def.priceDisplayFormat(fsec.cme_price_display_format);
+}
+
+void MarketDataWriter::_add_security(const FixedSecurity &security) {
+   bts_assert(security.numeric_id != 0);
+   // Set up the shared memory for this security
+   auto handle = me->md_producer.add_security(security);
+   // Add it to the index so it can receive messages
+   me->index.add_security(security.numeric_id, handle);
+   // handle.update_exchange_status(utils::TimeStamp::now(), status);
+}
+
+} // namespace bts::md

src/server/sparkd/md/MarketDataWriter.h

@@ -0,0 +1,51 @@
+#pragma once
+#include "lib/md/IMarketDataListener.h"
+#include "lib/md/MDProducer.h"
+#include "lib/md/MDProducerHandle.h"
+#include "lib/md/SecurityIndex.h"
+#include "server/sparkd/md/platforms/mdp3_onixs/CMESecurityIndex.h"
+#include <cstdint>
+#include <memory>
+
+namespace bts::md {
+
+class MDProducer;
+class FixedSecurity;
+
+using SecurityMarketData = SecurityData;
+// struct SecurityMarketData {
+//    std::int32_t id;
+//    MDProducerHandle handle;
+// };
+
+// Listens to the emumba market data source and
+class MarketDataWriter : public IMarketDataListener {
+ public:
+   MarketDataWriter(
+       ConcurrentSecurityIndex<std::uint32_t, SecurityData> &index,
+       MDProducer &md_producer);
+   ~MarketDataWriter();
+
+   void on_securities(const std::vector<domain::Security const *>
+                          &complete_security_list) override;
+   void on_security_state(const domain::SecurityState &security_state) override;
+   void on_security_statistics(
+       const domain::SecurityMarketDataStatistics &security_stats) override;
+   void on_book(const domain::Book &book) override;
+   void on_trades(
+       const std::vector<domain::TradeRecord const *> &trades) override;
+   void on_rfqs(const std::vector<domain::RFQ const *> &rfqs) override;
+   void on_error(const MarketDataError &error) override;
+
+ private:
+   struct PImpl;
+   std::unique_ptr<PImpl> me;
+
+   void _add_index(const domain::Security &sec);
+   void _add_future(const domain::Security &sec);
+   void _copy_security_common(
+       const domain::Security &from, FixedSecurity &to) const;
+   void _add_security(const FixedSecurity &security);
+};
+
+} // namespace bts::md

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

@@ -21,7 +21,7 @@ MDP3Listener::MDP3Listener(
       _verbose_ticker(config::get<CMEOptions>().verbose_ticker()),
       _packet_arrived(), _latest_sequence(0),
       // _settlement_recorder(settles),
-      _symbol_filter_storage(), _symbol_filter(), _channels(channels),
+      _symbol_filter_storage(), _symbol_filter(), _channels(std::move(channels)),
       _concurrency_monitor(0), _book_updates(), _mbo(mbo),
       _trade_summary_dispatcher(trade_summary_dispatcher) {
    _book_updates.reserve(32);

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

@@ -1,7 +1,9 @@
 #include "MDP3Main.h"
 
 #include "lib/config/config.h"
+#include "lib/domain_model/Security.pb.h"
 #include "lib/md/MBOProcessor.h"
+#include "lib/md/MDProducerChannel.h"
 #include "lib/md/SecurityDB.h"
 #include "lib/md/cme/MDP3Utils.h"
 #include "lib/md/coconut2.h"
@@ -18,6 +20,7 @@
 #include <boost/filesystem.hpp>
 #include <memory>
 #include <optional>
+#include <range/v3/view.hpp>
 #include <set>
 
 using namespace std;
@@ -39,6 +42,10 @@ MDP3Main::MDP3Main()
 
 MDP3Main::~MDP3Main() {}
 
+CMESecurityIndex &MDP3Main::index() { return _index; }
+
+MDProducer &MDP3Main::producer() { return _producer; }
+
 const CMEOptions::CMEChannel channel_settings_for(const string &id) {
    const auto &config = config::get<CMEOptions>();
    for (const auto &ch : config.channel()) {
@@ -53,6 +60,7 @@ const CMEOptions::CMEChannel channel_settings_for(const string &id) {
    }
    return CMEOptions::CMEChannel();
 }
+
 bool feed_engine_exists(const string &name) {
    const auto &config = config::get<CMEOptions>();
    for (const auto &feed : config.feed_engine()) {
@@ -88,16 +96,16 @@ const onixs::FeedEngineSettings feed_engine_settings_from_config(
    return result;
 }
 
-size_t
-count_engine_uses(const vector<CMEChannel> &channels, const string &feed_name) {
+size_t count_engine_uses(
+    const vector<CMEChannel> &channels, const string &feed_name) {
    return count_if(channels.begin(), channels.end(), [&](const CMEChannel &ch) {
       const auto &settings = channel_settings_for(ch.id);
       return settings.has_feed_engine() && settings.feed_engine() == feed_name;
    });
 }
 
-map<string, onixs::FeedEngineSettings>
-create_feed_engine_settings(const std::vector<CMEChannel> &channels) {
+map<string, onixs::FeedEngineSettings> create_feed_engine_settings(
+    const std::vector<CMEChannel> &channels) {
    map<string, onixs::FeedEngineSettings> settings;
    const auto &config = config::get<CMEOptions>();
 
@@ -115,8 +123,8 @@ create_feed_engine_settings(const std::vector<CMEChannel> &channels) {
    return settings;
 }
 
-map<string, unique_ptr<onixs::FeedEngine>>
-create_feed_engines(const vector<CMEChannel> &channels) {
+map<string, unique_ptr<onixs::FeedEngine>> create_feed_engines(
+    const vector<CMEChannel> &channels) {
    map<string, unique_ptr<onixs::FeedEngine>> result;
    auto settings = create_feed_engine_settings(channels);
    // Loop through settings and instantiate feed engines
@@ -128,8 +136,8 @@ create_feed_engines(const vector<CMEChannel> &channels) {
    return result;
 }
 
-onixs::SessionSettings
-session_settings_for_channel(const string &channel_id, MBOWriter *mbo) {
+onixs::SessionSettings session_settings_for_channel(
+    const string &channel_id, MBOWriter *mbo) {
    const auto &options = channel_settings_for(channel_id);
    onixs::SessionSettings settings;
 
@@ -163,18 +171,37 @@ void MDP3Main::start(
    await_cme_config();
    auto license = onixs_create_temporary_license();
    initialize(license.path.string());
-   const auto &channels = parse_channels(symbols);
+   std::vector<const UserSymbol *> symbols_without_channels;
+   const auto &channels = parse_channels(symbols, &symbols_without_channels);
    _engines = create_feed_engines(channels);
+   if (!symbols_without_channels.empty()) {
+      LOG(md, Info, "The following configured symbols lack mdp3 channels:\n{}",
+          fmt::to_string(fmt::join(
+              ranges::transform_view(
+                  symbols_without_channels, [](const auto *s) { return *s; }),
+              ", ")));
+
+      // create_listener(const std::set<UserSymbol> &symbols, MBOWriter *,
+      // TradeSummarySignalDispatcher *) unique_ptr<onixs::Handler> phandler =
+      // prepare_handler("", mbo);
+      for (const auto *symbol : symbols_without_channels) {
+         get_channel(*symbol);
+         LOG(md, Info, "Initialized shared memory: {}", *symbol);
+         if (symbol->exchange() != api::VENUE_SIM) {
+            UserSymbol sim_symbol(
+                symbol->symbol(), api::VENUE_SIM, symbol->type());
+            get_channel(sim_symbol);
+            LOG(md, Info, "Initialized shared memory: {}", sim_symbol);
+         }
+      }
+   }
    for (const auto &channel : channels) {
       unique_ptr<onixs::Handler> phandler = prepare_handler(channel.id, mbo);
-      _listeners.push_back(prepare_listener(*phandler, channel, mbo, tssd));
+      _listeners.push_back(
+          prepare_listener(*phandler, channel.symbols, mbo, tssd));
       _handlers.insert(std::make_pair(channel.id, move(phandler)));
-      std::stringstream symstr;
-      for (const auto &sym : channel.symbols) {
-         symstr << sym.symbol() << "-" << SecurityType_Name(sym.type()) << " ";
-      }
       LOG(md, Info, "CME: Initialized channel {} ({})", channel.id,
-          symstr.str());
+          fmt::to_string(fmt::join(channel.symbols, " ")));
    }
 
    for (auto &pair : _handlers) {
@@ -201,8 +228,8 @@ void MDP3Main::initialize(const std::string &license_store) {
 
 // //
 // Initialize HandlerSettings and allocate the channel Handler
-unique_ptr<onixs::Handler>
-MDP3Main::prepare_handler(const string &channel_id, MBOWriter *mbo) {
+unique_ptr<onixs::Handler> MDP3Main::prepare_handler(
+    const string &channel_id, MBOWriter *mbo) {
    const auto &pub_options = config::get<PublisherOptions>();
    const auto &options = channel_settings_for(channel_id);
    LOG(md, Info, "Loaded channel options for channel {}:\n{}", channel_id,
@@ -354,24 +381,26 @@ MDP3Main::prepare_handler(const string &channel_id, MBOWriter *mbo) {
    }
 }
 
-//
-// Create a listener for this channel and register callbacks
-unique_ptr<MDP3Listener> MDP3Main::prepare_listener(
-    onixs::Handler &handler,
-    const CMEChannel &channel,
+MDProducerChannel MDP3Main::get_channel(const UserSymbol &symbol) {
+   return _producer.get_channel(
+       symbol.symbol().c_str(), symbol.type(), symbol.exchange());
+}
+
+unique_ptr<MDP3Listener> MDP3Main::create_listener_(
+    const std::set<UserSymbol> &symbols,
     MBOWriter *mbo,
     TradeSummarySignalDispatcher *trade_summary_dispatcher) {
    // We need to get a handler to the producer channels so we can produce
    // heartbeats
    vector<MDProducerChannel> producer_channels;
-   for (auto &symbol : channel.symbols) {
-      producer_channels.push_back(_producer.get_channel(
-          symbol.symbol().c_str(), symbol.type(), symbol.exchange()));
+   for (auto &symbol : symbols) {
+      producer_channels.push_back(get_channel(symbol));
    }
 
    unique_ptr<MDP3Listener> listener(new MDP3Listener(
-       _index, _producer, mbo, producer_channels, trade_summary_dispatcher));
-   for (auto &symbol : channel.symbols) {
+       _index, _producer, mbo, std::move(producer_channels),
+       trade_summary_dispatcher));
+   for (auto &symbol : symbols) {
       listener->add_filter_symbol(symbol.symbol());
    }
 
@@ -380,24 +409,40 @@ unique_ptr<MDP3Listener> MDP3Main::prepare_listener(
       auto &preload_symbols = config::get<md::SecurityPreload>();
       std::set<std::string> preload(
           preload_symbols.symbol().begin(), preload_symbols.symbol().end());
-      for (auto &symbol : channel.symbols) {
-         if (preload.find(symbol.symbol()) != preload.end()) {
-            std::set<UserSymbol> unit_set{symbol};
-            db::load_securities(unit_set, [&](const api::Security &security) {
-               with_security_like(security, [&](FixedSecurity &fs) {
-                  convert_security(security, fs, true);
-                  listener->add_security(
-                      fs, bts::co2::TradingStatus::TRADINGSTATUS_UNKNOWN);
-               });
-            });
+      for (const auto &symbol : symbols) {
+         if (preload.find(symbol.symbol()) == preload.end()) {
+            continue;
          }
+         std::set<UserSymbol> unit_set{symbol};
+         db::load_securities(unit_set, [&](const api::Security &security) {
+            with_security_like(security, [&](FixedSecurity &fs) {
+               convert_security(security, fs, true);
+               listener->add_security(
+                   fs, bts::co2::TradingStatus::TRADINGSTATUS_UNKNOWN);
+            });
+         });
       }
    }
 
-   handler.registerListener(*listener);
-   handler.registerMarketDataListener(*listener);
-   handler.registerSecurityListener(*listener);
+   return listener;
+}
 
+void MDP3Main::connect_listener_(
+    onixs::Handler &handler, MDP3Listener &listener) {
+   handler.registerListener(listener);
+   handler.registerMarketDataListener(listener);
+   handler.registerSecurityListener(listener);
+}
+
+//
+// Create a listener for this channel and register callbacks
+unique_ptr<MDP3Listener> MDP3Main::prepare_listener(
+    onixs::Handler &handler,
+    const std::set<UserSymbol> &symbols,
+    MBOWriter *mbo,
+    TradeSummarySignalDispatcher *trade_summary_dispatcher) {
+   auto listener = create_listener_(symbols, mbo, trade_summary_dispatcher);
+   connect_listener_(handler, *listener);
    return listener;
 }
 

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

@@ -2,6 +2,9 @@
 
 #include "CMESecurityIndex.h"
 #include "MDP3Listener.h"
+#include "lib/md/MDProducer.h"
+#include "lib/md/MDProducerChannel.h"
+#include "lib/md/UserSymbol.h"
 #include "lib/md/coconut2.h"
 #include "lib/utils/bts_no_copy.h"
 #include "onixs.h"
@@ -9,18 +12,16 @@
 
 #include <map>
 #include <memory>
+#include <set>
 #include <vector>
 
-namespace bts {
-namespace md {
+namespace bts::md {
 
 // void run_onixs_mdp3_server(UserSymbol* sym, size_t symcount);
 
 class MBOWriter;
 class TradeSummarySignalDispatcher;
 
-struct CMEChannel;
-
 /*
  * The MDP3 provider is very simple
  * Control starts in MDP3Main which gets a list of all the symbols we need and
@@ -34,17 +35,26 @@ class MDP3Main : public IMDPlatform {
 
    void start(
        std::vector<UserSymbol> &, MBOWriter *, TradeSummarySignalDispatcher *);
+
    void send_heartbeats() override;
 
    void initialize(const std::string &license_store);
-   std::unique_ptr<onixs::Handler>
-   prepare_handler(const std::string &channel, MBOWriter *);
+
+   std::unique_ptr<onixs::Handler> prepare_handler(
+       const std::string &channel, MBOWriter *);
+
    std::unique_ptr<MDP3Listener> prepare_listener(
        onixs::Handler &handler,
-       const CMEChannel &channel,
+       const std::set<UserSymbol> &symbols,
        MBOWriter *,
        TradeSummarySignalDispatcher *);
 
+   bts::md::MDProducerChannel get_channel(const UserSymbol &symbol);
+
+   CMESecurityIndex &index();
+
+   MDProducer &producer();
+
  private:
    BTS_NO_COPY(MDP3Main);
 
@@ -56,7 +66,15 @@ class MDP3Main : public IMDPlatform {
    std::map<std::string, std::unique_ptr<onixs::Handler>> _handlers;
    std::vector<std::unique_ptr<MDP3Listener>> _listeners;
    std::vector<std::unique_ptr<onixs::TcpRecoveryService>> _recovery;
+
+   void add_symbol_(const UserSymbol &symbol);
+   std::unique_ptr<MDP3Listener> create_listener_(
+       const std::set<UserSymbol> &symbols,
+       MBOWriter *,
+       TradeSummarySignalDispatcher *);
+
+   void connect_listener_(onixs::Handler &handler, MDP3Listener &listener);
+
 };
 
-} // namespace md
-} // namespace bts
+} // namespace bts::md

src/web/api/tsconfig.clang.release.json

@@ -0,0 +1,14 @@
+{
+  "extends": "./tsconfig.json",
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "../../../build/clang/release/web/api",
+    "removeComments": true,
+    "paths": {
+      "proto": ["../../../../build/clang/release/web/api/proto"]
+    }
+  },
+  "include": [
+    "./src/**/*.ts"
+  ]
+}