ADD: Add BlockForStop to C++ client

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.10.0 - TBD
+
+#### Enhancements
+- Added `LiveThreaded::BlockForStop` to make it easier to wait for one or more records
+  before closing the session
+
 ## 0.9.1 - 2023-07-11
 
 #### Enhancements
@@ -44,18 +50,24 @@
 - Added initial support for live data with `LiveBlocking` and `LiveThreaded` clients
 - Added support for statistics schema
 - Added `SystemMsg` and `ErrorMsg` records for use in live data
-- Added `strike_price`, `strike_price_currency`, and `instrument_class` to `InstrumentDefMsg`
+- Added `strike_price`, `strike_price_currency`, and `instrument_class` to
+  `InstrumentDefMsg`
 - Added `FixedPx` helper class for formatting fixed prices
 - Added configurable log receiver `ILogReceiver`
-- Added `instrument_class`, `strike_price`, and `strike_price_currency` to definition schema
-- Added additional `condition` variants for `DatasetConditionDetail` (degraded, pending, missing)
-- Added additional member `last_modified_date` to `DatasetConditionDetail` Added `has_mixed_schema`, `has_mixed_stype_in`, and `ts_out` to `Metadata` to support live data
+- Added `instrument_class`, `strike_price`, and `strike_price_currency` to definition
+  schema
+- Added additional `condition` variants for `DatasetConditionDetail` (degraded, pending,
+  missing)
+- Added additional member `last_modified_date` to `DatasetConditionDetail`
+- Added `has_mixed_schema`, `has_mixed_stype_in`, and `ts_out` to `Metadata` to support
+  live data
 - Added optional `compression` parameter to `BatchSubmitJob`
 
 #### Breaking changes
 - Removed `related` and `related_security_id` from `InstrumentDefMsg`
 - Renamed `BatchJob.cost` to `cost_usd` and value now expressed as US dollars
-- Renamed `SType::ProductId` to `SType::InstrumentId` and `SType::Native` to `SType::RawSymbol`
+- Renamed `SType::ProductId` to `SType::InstrumentId` and `SType::Native` to
+  `SType::RawSymbol`
 - Renamed `RecordHeader::product_id` to `instrument_id`
 - Renamed `InstrumentDefMsg::symbol` to `raw_symbol`
 - Renamed `SymbolMapping::native_symbol` to `raw_symbol`

README.md

@@ -86,10 +86,8 @@ using namespace databento;
 int main() {
   std::unordered_map<std::uint32_t, std::string> symbol_mappings;
 
-  auto client = LiveBuilder{}
-                    .SetKey("$YOUR_API_KEY")
-                    .SetDataset("GLBX.MDP3")
-                    .BuildThreaded();
+  auto client =
+      LiveBuilder{}.SetKeyFromEnv().SetDataset("GLBX.MDP3").BuildThreaded();
 
   auto handler = [&symbol_mappings](const Record& rec) {
     if (rec.Holds<TradeMsg>()) {
@@ -99,8 +97,7 @@ int main() {
                 << '\n';
     } else if (rec.Holds<SymbolMappingMsg>()) {
       auto mapping = rec.Get<SymbolMappingMsg>();
-      symbol_mappings[mapping.hd.instrument_id] =
-          mapping.stype_out_symbol.data();
+      symbol_mappings[mapping.hd.instrument_id] = mapping.STypeOutSymbol();
     }
     return KeepGoing::Continue;
   };

example/live/readme.cpp

@@ -13,10 +13,8 @@ using namespace databento;
 int main() {
   std::unordered_map<std::uint32_t, std::string> symbol_mappings;
 
-  auto client = LiveBuilder{}
-                    .SetKey("$YOUR_API_KEY")
-                    .SetDataset("GLBX.MDP3")
-                    .BuildThreaded();
+  auto client =
+      LiveBuilder{}.SetKeyFromEnv().SetDataset("GLBX.MDP3").BuildThreaded();
 
   auto handler = [&symbol_mappings](const Record& rec) {
     if (rec.Holds<TradeMsg>()) {
@@ -26,8 +24,7 @@ int main() {
                 << '\n';
     } else if (rec.Holds<SymbolMappingMsg>()) {
       auto mapping = rec.Get<SymbolMappingMsg>();
-      symbol_mappings[mapping.hd.instrument_id] =
-          mapping.stype_out_symbol.data();
+      symbol_mappings[mapping.hd.instrument_id] = mapping.STypeOutSymbol();
     }
     return KeepGoing::Continue;
   };

example/live/simple.cpp

@@ -40,7 +40,7 @@ int main() {
   };
   auto record_callback = [&symbol_mappings](const databento::Record& rec) {
     using databento::RType;
-    switch (rec.Header().rtype) {
+    switch (rec.RType()) {
       case RType::Mbo: {
         auto ohlcv = rec.Get<databento::WithTsOut<databento::MboMsg>>();
         std::cout << "Received tick for "
@@ -58,7 +58,7 @@ int main() {
         auto mapping = rec.Get<databento::SymbolMappingMsg>();
         std::cout << "Received symbol mapping: " << mapping << '\n';
         symbol_mappings.emplace(mapping.hd.instrument_id,
-                                mapping.stype_in_symbol.data());
+                                mapping.STypeInSymbol());
         break;
       }
       case RType::System: {
@@ -75,7 +75,7 @@ int main() {
       }
       default: {
         std::cerr << "Received unknown record with rtype " << std::hex
-                  << static_cast<std::uint16_t>(rec.Header().rtype) << '\n';
+                  << static_cast<std::uint16_t>(rec.RType()) << '\n';
       }
     }
     return databento::KeepGoing::Continue;

include/databento/live_threaded.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <chrono>
 #include <functional>  // function
 #include <memory>      // unique_ptr
 #include <string>
@@ -76,6 +77,10 @@ class LiveThreaded {
              ExceptionCallback exception_callback);
   // Closes the current connection, and attempts to reconnect to the gateway.
   void Reconnect();
+  // Blocking wait with an optional timeout for the session to close when the
+  // record_callback or the exception_callback return Stop.
+  void BlockForStop();
+  KeepGoing BlockForStop(std::chrono::milliseconds timeout);
 
  private:
   struct Impl;

src/live_threaded.cpp

@@ -2,7 +2,9 @@
 
 #include <atomic>
 #include <chrono>  // milliseconds
+#include <condition_variable>
 #include <exception>
+#include <mutex>
 #include <sstream>
 #include <thread>
 #include <utility>  // forward, move, swap
@@ -19,8 +21,18 @@ struct LiveThreaded::Impl {
       : log_receiver{log_receiver},
         blocking{log_receiver, std::forward<A>(args)...} {}
 
+  void NotifyOfStop() {
+    const std::lock_guard<std::mutex> lock{last_cb_ret_mutex};
+    last_cb_ret = KeepGoing::Stop;
+    last_cb_ret_cv.notify_all();
+  }
+
   ILogReceiver* log_receiver;
+  // Set to false when destructor is called
   std::atomic<bool> keep_going{true};
+  KeepGoing last_cb_ret{KeepGoing::Continue};
+  std::mutex last_cb_ret_mutex;
+  std::condition_variable last_cb_ret_cv;
   LiveBlocking blocking;
 };
 
@@ -111,6 +123,27 @@ void LiveThreaded::Start(MetadataCallback metadata_callback,
 
 void LiveThreaded::Reconnect() { impl_->blocking.Reconnect(); }
 
+void LiveThreaded::BlockForStop() {
+  std::unique_lock<std::mutex> lock{impl_->last_cb_ret_mutex};
+  auto* impl = impl_.get();
+  // wait for stop
+  impl_->last_cb_ret_cv.wait(
+      lock, [impl] { return impl->last_cb_ret == KeepGoing::Stop; });
+}
+
+databento::KeepGoing LiveThreaded::BlockForStop(
+    std::chrono::milliseconds timeout) {
+  std::unique_lock<std::mutex> lock{impl_->last_cb_ret_mutex};
+  auto* impl = impl_.get();
+  // wait for stop
+  if (impl_->last_cb_ret_cv.wait_for(lock, timeout, [impl] {
+        return impl->last_cb_ret == KeepGoing::Stop;
+      })) {
+    return KeepGoing::Stop;
+  }
+  return KeepGoing::Continue;
+}
+
 void LiveThreaded::ProcessingThread(Impl* impl,
                                     MetadataCallback&& metadata_callback,
                                     RecordCallback&& record_callback,
@@ -146,6 +179,7 @@ void LiveThreaded::ProcessingThread(Impl* impl,
         if (rec) {
           if (record_cb(*rec) == KeepGoing::Stop) {
             impl->blocking.Stop();
+            impl->NotifyOfStop();
             return;
           }
         }  // else timeout
@@ -155,6 +189,7 @@ void LiveThreaded::ProcessingThread(Impl* impl,
             ExceptionAction::Restart) {
           break;  // break out of NextRecord loop, to restart Start loop
         } else {
+          impl->NotifyOfStop();
           return;
         }
       }

test/include/mock/mock_lsg_server.hpp

@@ -28,7 +28,7 @@ class MockLsgServer {
   void Authenticate();
   void Subscribe(const std::vector<std::string>& symbols, Schema schema,
                  SType stype);
-  void Start(Schema schema);
+  void Start();
   std::size_t Send(const std::string& msg);
   ::ssize_t UncheckedSend(const std::string& msg);
   template <typename Rec>

test/src/dbn_decoder_tests.cpp

@@ -707,8 +707,8 @@ TEST_P(DbnDecoderSchemaTests, TestDecodeDefinition) {
   const auto& ch_def1 = ch_record1->Get<InstrumentDefMsg>();
   const auto& f_def1 = f_record1->Get<InstrumentDefMsg>();
   EXPECT_EQ(ch_def1, f_def1);
-  EXPECT_STREQ(ch_def1.exchange.data(), "XNAS");
-  EXPECT_STREQ(ch_def1.raw_symbol.data(), "MSFT");
+  EXPECT_STREQ(ch_def1.Exchange(), "XNAS");
+  EXPECT_STREQ(ch_def1.RawSymbol(), "MSFT");
   EXPECT_EQ(ch_def1.security_update_action, SecurityUpdateAction::Add);
   EXPECT_EQ(ch_def1.min_lot_size_round_lot, 100);
   EXPECT_EQ(ch_def1.instrument_class, InstrumentClass::Stock);
@@ -723,8 +723,8 @@ TEST_P(DbnDecoderSchemaTests, TestDecodeDefinition) {
   const auto& ch_def2 = ch_record2->Get<InstrumentDefMsg>();
   const auto& f_def2 = f_record2->Get<InstrumentDefMsg>();
   EXPECT_EQ(ch_def2, f_def2);
-  EXPECT_STREQ(ch_def2.exchange.data(), "XNAS");
-  EXPECT_STREQ(ch_def2.raw_symbol.data(), "MSFT");
+  EXPECT_STREQ(ch_def2.Exchange(), "XNAS");
+  EXPECT_STREQ(ch_def2.RawSymbol(), "MSFT");
   EXPECT_EQ(ch_def2.security_update_action, SecurityUpdateAction::Add);
   EXPECT_EQ(ch_def2.min_lot_size_round_lot, 100);
   EXPECT_EQ(ch_def2.instrument_class, InstrumentClass::Stock);

test/src/live_blocking_tests.cpp

@@ -31,6 +31,7 @@ class LiveBlockingTests : public testing::Test {
   }
 
   static constexpr auto kKey = "32-character-with-lots-of-filler";
+  static constexpr auto kLocalhost = "127.0.0.1";
 
   std::unique_ptr<ILogReceiver> logger_{new NullLogReceiver};
 };
@@ -44,26 +45,24 @@ TEST_F(LiveBlockingTests, TestAuthentication) {
                                         }};
 
   const LiveBlocking target{logger_.get(),      kKey,
-                            dataset::kXnasItch, "127.0.0.1",
+                            dataset::kXnasItch, kLocalhost,
                             mock_server.Port(), kTsOut};
 }
 
 TEST_F(LiveBlockingTests, TestStart) {
   constexpr auto kTsOut = false;
-  constexpr auto kSchema = Schema::Mbo;
   const mock::MockLsgServer mock_server{dataset::kGlbxMdp3, kTsOut,
                                         [](mock::MockLsgServer& self) {
                                           self.Accept();
                                           self.Authenticate();
-                                          self.Start(kSchema);
+                                          self.Start();
                                         }};
 
-  LiveBlocking target{logger_.get(),      kKey,
-                      dataset::kGlbxMdp3, "127.0.0.1",
+  LiveBlocking target{logger_.get(),      kKey,  dataset::kGlbxMdp3, kLocalhost,
                       mock_server.Port(), kTsOut};
   const auto metadata = target.Start();
   EXPECT_EQ(metadata.version, 1);
-  EXPECT_EQ(metadata.schema, kSchema);
+  EXPECT_TRUE(metadata.has_mixed_schema);
   EXPECT_EQ(metadata.dataset, dataset::kGlbxMdp3);
 }
 
@@ -81,7 +80,7 @@ TEST_F(LiveBlockingTests, TestSubscribe) {
         self.Subscribe(kSymbols, kSchema, kSType);
       }};
 
-  LiveBlocking target{logger_.get(),      kKey,  kDataset, "127.0.0.1",
+  LiveBlocking target{logger_.get(),      kKey,  kDataset, kLocalhost,
                       mock_server.Port(), kTsOut};
   target.Subscribe(kSymbols, kSchema, kSType);
 }
@@ -107,7 +106,7 @@ TEST_F(LiveBlockingTests, TestSubscriptionChunking) {
         }
       }};
 
-  LiveBlocking target{logger_.get(),      kKey,  kDataset, "127.0.0.1",
+  LiveBlocking target{logger_.get(),      kKey,  kDataset, kLocalhost,
                       mock_server.Port(), kTsOut};
   const std::vector<std::string> kSymbols(kSymbolCount, kSymbol);
   target.Subscribe(kSymbols, kSchema, kSType);
@@ -126,8 +125,7 @@ TEST_F(LiveBlockingTests, TestNextRecord) {
         }
       }};
 
-  LiveBlocking target{logger_.get(),      kKey,
-                      dataset::kXnasItch, "127.0.0.1",
+  LiveBlocking target{logger_.get(),      kKey,  dataset::kXnasItch, kLocalhost,
                       mock_server.Port(), kTsOut};
   for (size_t i = 0; i < kRecCount; ++i) {
     const auto rec = target.NextRecord();
@@ -177,8 +175,7 @@ TEST_F(LiveBlockingTests, TestNextRecordTimeout) {
         self.SendRecord(kRec);
       }};
 
-  LiveBlocking target{logger_.get(),      kKey,
-                      dataset::kXnasItch, "127.0.0.1",
+  LiveBlocking target{logger_.get(),      kKey,  dataset::kXnasItch, kLocalhost,
                       mock_server.Port(), kTsOut};
   {
     // wait for server to send first record to avoid flaky timeouts
@@ -232,8 +229,7 @@ TEST_F(LiveBlockingTests, TestNextRecordPartialRead) {
                              send_remaining_cv);
       }};
 
-  LiveBlocking target{logger_.get(),      kKey,
-                      dataset::kGlbxMdp3, "127.0.0.1",
+  LiveBlocking target{logger_.get(),      kKey,  dataset::kGlbxMdp3, kLocalhost,
                       mock_server.Port(), kTsOut};
   auto rec = target.NextRecord();
   ASSERT_TRUE(rec.Holds<MboMsg>());
@@ -276,8 +272,7 @@ TEST_F(LiveBlockingTests, TestNextRecordWithTsOut) {
         }
       }};
 
-  LiveBlocking target{logger_.get(),      kKey,
-                      dataset::kXnasItch, "127.0.0.1",
+  LiveBlocking target{logger_.get(),      kKey,  dataset::kXnasItch, kLocalhost,
                       mock_server.Port(), kTsOut};
   for (size_t i = 0; i < kRecCount; ++i) {
     const auto rec = target.NextRecord();
@@ -323,7 +318,7 @@ TEST_F(LiveBlockingTests, TestStop) {
   };                              // namespace databento
 
   LiveBlocking target{logger_.get(),       kKey,
-                      dataset::kXnasItch,  "127.0.0.1",
+                      dataset::kXnasItch,  kLocalhost,
                       mock_server->Port(), kTsOut};
   ASSERT_EQ(target.NextRecord().Get<WithTsOut<TradeMsg>>(), kRec);
   target.Stop();
@@ -335,8 +330,8 @@ TEST_F(LiveBlockingTests, TestStop) {
 
 TEST_F(LiveBlockingTests, TestConnectWhenGatewayNotUp) {
   constexpr auto kTsOut = true;
-  ASSERT_THROW(LiveBlocking(logger_.get(), kKey, dataset::kXnasItch,
-                            "127.0.0.1", 80, kTsOut),
+  ASSERT_THROW(LiveBlocking(logger_.get(), kKey, dataset::kXnasItch, kLocalhost,
+                            80, kTsOut),
                databento::TcpError);
 }
 
@@ -380,11 +375,11 @@ TEST_F(LiveBlockingTests, TestReconnect) {
         self.Accept();
         self.Authenticate();
         self.Subscribe(kAllSymbols, Schema::Trades, SType::RawSymbol);
-        self.Start(Schema::Trades);
+        self.Start();
         self.SendRecord(kRec);
       }}};
   LiveBlocking target{logger_.get(),       kKey,
-                      dataset::kXnasItch,  "127.0.0.1",
+                      dataset::kXnasItch,  kLocalhost,
                       mock_server->Port(), kTsOut};
   // Tell server to close connection
   {
@@ -401,7 +396,7 @@ TEST_F(LiveBlockingTests, TestReconnect) {
   target.Reconnect();
   target.Subscribe(kAllSymbols, Schema::Trades, SType::RawSymbol);
   const auto metadata = target.Start();
-  ASSERT_EQ(metadata.schema, Schema::Trades);
+  EXPECT_TRUE(metadata.has_mixed_schema);
   const auto rec = target.NextRecord();
   ASSERT_TRUE(rec.Holds<TradeMsg>());
   ASSERT_EQ(rec.Get<TradeMsg>(), kRec);