SP-142 Terrible meat hack

src/lib/config/defaults/MarketData.conf

@@ -37,6 +37,12 @@ CMEOptions {
             # channels_per_thread: 1
             spin_time: 0
         }
+        channel {
+            id: "316"
+            feed_engine: "mdp3feed"
+            use_tcp_replay: false
+            sunday_pre_open: true
+        }
         channel {
             id: "*"
             feed_engine: "mdp3feed"

src/lib/config/protos/MarketData.proto

@@ -126,6 +126,8 @@ message CMEOptions
       // optional BookType book_type = 34 [default = BOOKTYPE_CONSOLIDATED];
       optional bool validate_mbo = 35 [default = false];
 
+      optional bool sunday_pre_open = 36 [ default = true ];
+
    }
    optional string onixs_license = 1 [default = "{default}"];
    repeated CMEChannel channel = 3;

src/lib/utils/datetime.cpp

@@ -350,6 +350,14 @@ bool is_local_time_sunday() {
    auto now = local_now();
    return now.date().day_of_week() == boost::date_time::Sunday;
 }
+bool is_local_time_near_sunday_sunday() {
+   auto now = local_now();
+   auto day = now.date().day_of_week();
+   auto hour = now.time_of_day().hours();
+   return 
+      day == boost::date_time::Sunday
+      || ( day == boost::date_time::Monday && hour < 6);
+}
 bool is_local_time_between(int hh_start, int mm_start, int hh_end, int mm_end) {
    using namespace boost::posix_time;
    auto now = local_now();

src/lib/utils/datetime.h

@@ -86,6 +86,7 @@ extern "C" BTS_EXPORT LLDateTime ll_from_parts(int year, int month, int day, int
 BTS_EXPORT LLDateTime ll_from_date(const Date& date);
 
 bool is_local_time_sunday();
+bool is_local_time_near_sunday_sunday();
 bool is_local_time_between(int hh_start, int mm_start, int hh_end, int mm_end);
 
 #ifndef NO_CPP0X

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

@@ -25,10 +25,11 @@ using std::size_t;
 namespace bts {
 namespace md {
 
-MDP3Main::MDP3Main() :
+MDP3Main::MDP3Main(MBOWriter* mbo) :
    _index(),
    _recorder ({ utils::Dispatcher::current(), [=](ExchangeKey k) { return SecurityKey(_index.get_key(k.value())); } }),
-   _producer(PLATFORM_MDP3ONIXS, _recorder ),
+   _producer(PLATFORM_MDP3ONIXS, _recorder),
+   _mbo(mbo),
    // _settlement_recorder(PLATFORM_MDP3ONIXS),
    _engines(),
    _handlers(),
@@ -117,8 +118,7 @@ map<string, unique_ptr<onixs::FeedEngine>> create_feed_engines(const vector<CMEC
    return result;
 }
 
-onixs::SessionSettings session_settings_for_channel(const string& channel_id, MBOWriter* mbo) {
-   const auto& options = channel_settings_for(channel_id);
+onixs::SessionSettings session_settings_for_channel(const CMEOptions::CMEChannel& options, MBOWriter* mbo) {
    onixs::SessionSettings settings;
 
    if (mbo) {
@@ -139,38 +139,63 @@ onixs::SessionSettings session_settings_for_channel(const string& channel_id, MB
       settings.joinRecovery(onixs::JoinRecoveryOptions::Disabled);
       settings.gapAndErrorHandling(onixs::GapAndErrorHandlingOptions::ContinueProcessing);
    }
+
+   // if (options.sunday_pre_open() && datetime::is_local_time_sunday()) {
+   //    LOG(md,Info,"Setting pre-open flag for channel {}", channel_id);
+   //    settings.joinRecovery(onixs::JoinRecoveryOptions::Disabled);
+   // }
+
    return settings;
 }
 
-void MDP3Main::start(std::vector<UserSymbol>& symbols, MBOWriter* mbo) {
+void MDP3Main::start(std::vector<UserSymbol>& symbols) {
    await_cme_config();
    initialize();
    const auto& channels = parse_channels(symbols);
    _engines = create_feed_engines(channels);
    for (const auto& channel : channels) {
-      unique_ptr<onixs::Handler> phandler = prepare_handler(channel.id, mbo);
-      _listeners.push_back(prepare_listener(*phandler, channel, mbo));
-      _handlers.insert(std::make_pair(channel.id, move(phandler)));
+      unique_ptr<onixs::Handler> phandler = prepare_handler(channel.id);
+      _listeners.push_back(prepare_listener(*phandler, channel));
+      _handlers.insert(std::make_pair(channel.id, ChannelHandler{ false, 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());
    }
 
-   for (auto& pair : _handlers) {
-      onixs::SessionSettings settings = session_settings_for_channel(pair.first, mbo);
-      pair.second->start(settings);
-   }
+   start_handlers();
    LOG(md,Info, "** CME MDP3 Publisher started **");
 }
 
+bool should_start(const CMEOptions::CMEChannel& options) {
+   if (options.sunday_pre_open() &&
+       datetime::is_local_time_near_sunday_sunday() ) {
+      return false;
+   }
+   return true;
+}
+
+void MDP3Main::start_handlers() {
+   size_t count = 0;
+   for (auto& pair : _handlers) {
+      const auto& options = channel_settings_for(pair.first);
+      if (!pair.second.started && should_start(options)) {
+         onixs::SessionSettings settings = session_settings_for_channel(options, _mbo);
+         pair.second.handler->start(settings);
+         pair.second.started = true;
+         LOG(md,Info, "Started handler {}", pair.first);
+      }
+      if(pair.second.started) ++count;
+   }
+   if (count != _handlers.size()) {
+      // If some handlers were not started try again later
+      utils::Dispatcher::current().post([this]() { start_handlers(); }, /*1 hour*/ 60*60*1000);
+   }
+}
+
 void MDP3Main::send_heartbeats() {
-   for (const auto& listener: _listeners) 
+   for (const auto& listener: _listeners) {
       listener->heartbeat();   
-      
-   // utils::DispatcherTimer settlement_timer(disp, 100, [&]() {
-   //    _settlement_recorder.do_work();
-   // });
-   // settlement_timer.start();      
+   }
 }
 
 void MDP3Main::initialize() {
@@ -197,7 +222,7 @@ void MDP3Main::initialize() {
 
 // //
 // 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) {
    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 ,options.DebugString());
@@ -210,7 +235,7 @@ unique_ptr<onixs::Handler> MDP3Main::prepare_handler(const string& channel_id, M
    settings.bookManagement().directBooks().maintain(true);
    settings.bookManagement().impliedBooks().maintain(true);
    settings.bookManagement().consolidatedBooks().maintain(false);
-   if (options.validate_mbo() && mbo != nullptr) {
+   if (options.validate_mbo() && _mbo != nullptr) {
       settings.bookManagement().mboBooks().maintain(true);
    }
 
@@ -340,14 +365,14 @@ unique_ptr<onixs::Handler> MDP3Main::prepare_handler(const string& channel_id, M
 
 //
 // Create a listener for this channel and register callbacks
-unique_ptr<MDP3Listener> MDP3Main::prepare_listener(onixs::Handler& handler, const CMEChannel& channel, MBOWriter* mbo) {
+unique_ptr<MDP3Listener> MDP3Main::prepare_listener(onixs::Handler& handler, const CMEChannel& channel) {
    // 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()));
   }
 
-   unique_ptr<MDP3Listener> listener(new MDP3Listener(_index, _producer, mbo, producer_channels));
+   unique_ptr<MDP3Listener> listener(new MDP3Listener(_index, _producer, _mbo, producer_channels));
    for (auto& symbol : channel.symbols) {
       listener->add_filter_symbol(symbol.symbol());
    }

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

@@ -26,18 +26,26 @@ struct CMEChannel;
  * There is a single listener for all callbacks which maps the exchange id to the MDProducerHandle and writes 
  * the data to it.
  */
+
+struct ChannelHandler {
+   bool started;
+   std::unique_ptr<onixs::Handler> handler;
+};
+
 class MDP3Main : public IMDPlatform
 {
 public:
-   MDP3Main();
+   MDP3Main(MBOWriter*);
    ~MDP3Main();
 
-   void start(std::vector<UserSymbol>&, MBOWriter*);
+   void start(std::vector<UserSymbol>&);
    void send_heartbeats() override;
    
    void initialize();
-   std::unique_ptr<onixs::Handler> prepare_handler(const std::string& channel, MBOWriter*);
-   std::unique_ptr<MDP3Listener> prepare_listener(onixs::Handler& handler, const CMEChannel& channel, MBOWriter*);
+   std::unique_ptr<onixs::Handler> prepare_handler(const std::string& channel);
+   std::unique_ptr<MDP3Listener> prepare_listener(onixs::Handler& handler, const CMEChannel& channel);
+
+   void start_handlers();
 
 private:
    BTS_NO_COPY(MDP3Main);
@@ -45,9 +53,10 @@ private:
    CMESecurityIndex _index;
    SecurityIDRecorder _recorder;
    MDProducer _producer;
+   MBOWriter* _mbo;
    // co2::SettlementRecorder _settlement_recorder;
    std::map<std::string, std::unique_ptr<onixs::FeedEngine>> _engines;
-   std::map<std::string, std::unique_ptr<onixs::Handler>> _handlers;
+   std::map<std::string, ChannelHandler> _handlers;
    std::vector<std::unique_ptr<MDP3Listener>> _listeners;
    std::vector<std::unique_ptr<onixs::TcpRecoveryService>> _recovery;
 };

src/server/sparkd/md/run_md.cpp

@@ -76,9 +76,9 @@ void run_onixs(MBOWriter* mbo) {
    get_system_md_symbols(api::VENUE_GLBX, api::SECURITYTYPE_UNSET, &cme_symbols);
    if(cme_symbols.size() > 0)
    {
-      MDP3Main* main = new MDP3Main();
+      MDP3Main* main = new MDP3Main(mbo);
       std::shared_ptr<IMDPlatform> ptr(main);
-      main->start(cme_symbols, mbo);
+      main->start(cme_symbols);
       platforms.push_back(ptr);
    }