FIX: Fix symbol chunking with string start param

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.18.1 - TBD
+
+### Bug fixes
+- Added missing symbol chunking for live `Subscribe` overloads with `const std::string&`
+  `start` parameter
+
 ## 0.18.0 - 2024-05-14
 
 ### Breaking changes

include/databento/live_blocking.hpp

@@ -78,6 +78,8 @@ class LiveBlocking {
   std::string GenerateCramReply(const std::string& challenge_key);
   std::string EncodeAuthReq(const std::string& auth);
   std::uint64_t DecodeAuthResp();
+  void Subscribe(const std::string& sub_msg,
+                 const std::vector<std::string>& symbols);
   detail::TcpClient::Result FillBuffer(std::chrono::milliseconds timeout);
   RecordHeader* BufferRecordHeader();
 

src/live_blocking.cpp

@@ -59,8 +59,32 @@ void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
 
 void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
                              Schema schema, SType stype_in, UnixNanos start) {
+  std::ostringstream sub_msg;
+  sub_msg << "schema=" << ToString(schema)
+          << "|stype_in=" << ToString(stype_in);
+  if (start.time_since_epoch().count()) {
+    sub_msg << "|start=" << start.time_since_epoch().count();
+  }
+  Subscribe(sub_msg.str(), symbols);
+}
+
+void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
+                             Schema schema, SType stype_in,
+                             const std::string& start) {
+  std::ostringstream sub_msg;
+  sub_msg << "schema=" << ToString(schema)
+          << "|stype_in=" << ToString(stype_in);
+  if (!start.empty()) {
+    sub_msg << "|start=" << start;
+  }
+  Subscribe(sub_msg.str(), symbols);
+}
+
+void LiveBlocking::Subscribe(const std::string& sub_msg,
+                             const std::vector<std::string>& symbols) {
   static constexpr auto kMethodName = "Live::Subscribe";
   constexpr std::ptrdiff_t kSymbolMaxChunkSize = 128;
+
   if (symbols.empty()) {
     throw InvalidArgumentError{kMethodName, "symbols",
                                "must contain at least one symbol"};
@@ -70,36 +94,17 @@ void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
     const auto chunk_size =
         std::min(kSymbolMaxChunkSize, std::distance(symbols_it, symbols.end()));
 
-    std::ostringstream sub_msg;
-    sub_msg << "schema=" << ToString(schema)
-            << "|stype_in=" << ToString(stype_in) << "|symbols="
-            << JoinSymbolStrings(kMethodName, symbols_it,
-                                 symbols_it + chunk_size);
-    if (start.time_since_epoch().count()) {
-      sub_msg << "|start=" << start.time_since_epoch().count();
-    }
-    sub_msg << '\n';
-    client_.WriteAll(sub_msg.str());
+    std::ostringstream chunked_sub_msg;
+    chunked_sub_msg << sub_msg << "|symbols="
+                    << JoinSymbolStrings(kMethodName, symbols_it,
+                                         symbols_it + chunk_size)
+                    << '\n';
+    client_.WriteAll(chunked_sub_msg.str());
 
     symbols_it += chunk_size;
   }
 }
 
-void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
-                             Schema schema, SType stype_in,
-                             const std::string& start) {
-  std::ostringstream sub_msg;
-  sub_msg << "schema=" << ToString(schema) << "|stype_in=" << ToString(stype_in)
-          << "|symbols="
-          << JoinSymbolStrings("LiveBlocking::Subscribe", symbols);
-  if (!start.empty()) {
-    sub_msg << "|start=" << start;
-  }
-  sub_msg << '\n';
-
-  client_.WriteAll(sub_msg.str());
-}
-
 databento::Metadata LiveBlocking::Start() {
   constexpr auto kMetadataPreludeSize = 8;
   client_.WriteAll("start_session\n");

test/include/mock/mock_lsg_server.hpp

@@ -35,6 +35,8 @@ class MockLsgServer {
   void Authenticate();
   void Subscribe(const std::vector<std::string>& symbols, Schema schema,
                  SType stype);
+  void Subscribe(const std::vector<std::string>& symbols, Schema schema,
+                 SType stype, const std::string& start);
   void Start();
   std::size_t Send(const std::string& msg);
   ::ssize_t UncheckedSend(const std::string& msg);

test/src/live_blocking_tests.cpp

@@ -92,7 +92,7 @@ TEST_F(LiveBlockingTests, TestSubscribe) {
   target.Subscribe(kSymbols, kSchema, kSType);
 }
 
-TEST_F(LiveBlockingTests, TestSubscriptionChunking) {
+TEST_F(LiveBlockingTests, TestSubscriptionChunkingUnixNanos) {
   constexpr auto kTsOut = false;
   constexpr auto kDataset = dataset::kXnasItch;
   const auto kSymbol = "TEST";
@@ -126,6 +126,42 @@ TEST_F(LiveBlockingTests, TestSubscriptionChunking) {
   target.Subscribe(kSymbols, kSchema, kSType);
 }
 
+TEST_F(LiveBlockingTests, TestSubscriptionChunkingStringStart) {
+  constexpr auto kTsOut = false;
+  constexpr auto kDataset = dataset::kXnasItch;
+  const auto kSymbol = "TEST";
+  const std::size_t kSymbolCount = 1000;
+  const auto kSchema = Schema::Ohlcv1M;
+  const auto kSType = SType::RawSymbol;
+  const auto kStart = "2020-01-01T00:00:00";
+
+  const mock::MockLsgServer mock_server{
+      kDataset, kTsOut,
+      [kSymbol, kSymbolCount, kSchema, kSType,
+       kStart](mock::MockLsgServer& self) {
+        self.Accept();
+        self.Authenticate();
+        std::size_t i{};
+        while (i < 1000) {
+          const auto chunk_size =
+              std::min(static_cast<std::size_t>(128), kSymbolCount - i);
+          const std::vector<std::string> symbols_chunk(chunk_size, kSymbol);
+          self.Subscribe(symbols_chunk, kSchema, kSType, kStart);
+          i += chunk_size;
+        }
+      }};
+
+  LiveBlocking target{logger_.get(),
+                      kKey,
+                      kDataset,
+                      kLocalhost,
+                      mock_server.Port(),
+                      kTsOut,
+                      VersionUpgradePolicy{}};
+  const std::vector<std::string> kSymbols(kSymbolCount, kSymbol);
+  target.Subscribe(kSymbols, kSchema, kSType, kStart);
+}
+
 TEST_F(LiveBlockingTests, TestNextRecord) {
   constexpr auto kTsOut = false;
   const auto kRecCount = 12;

test/src/mock_lsg_server.cpp

@@ -92,6 +92,12 @@ void MockLsgServer::Authenticate() {
 
 void MockLsgServer::Subscribe(const std::vector<std::string>& symbols,
                               Schema schema, SType stype) {
+  Subscribe(symbols, schema, stype, {});
+}
+
+void MockLsgServer::Subscribe(const std::vector<std::string>& symbols,
+                              Schema schema, SType stype,
+                              const std::string& start) {
   const auto received = Receive();
   EXPECT_NE(
       received.find("symbols=" +
@@ -101,6 +107,11 @@ void MockLsgServer::Subscribe(const std::vector<std::string>& symbols,
             std::string::npos);
   EXPECT_NE(received.find(std::string{"stype_in="} + ToString(stype)),
             std::string::npos);
+  if (start.empty()) {
+    EXPECT_EQ(received.find(std::string{"start="}), std::string::npos);
+  } else {
+    EXPECT_NE(received.find(std::string{"start="} + start), std::string::npos);
+  }
 }
 
 void MockLsgServer::Start() {