MOD: Batch long subscriptions

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## 0.9.1 - TBD
+
+#### Enhancements
+- Added constants for dataset codes for Databento Equity Basic and OPRA Pillar
+
+#### Bug fixes
+- Batch live subscriptions to avoid hitting max message length
+
 ## 0.9.0 - 2023-06-13
 
 #### Enhancements

include/databento/constants.hpp

@@ -18,7 +18,13 @@ static constexpr auto kUndefOrderSize =
 // This is not necessarily a comprehensive list of available datasets. Please
 // use `Historical.MetadataListDatasets` to retrieve an up-to-date list.
 namespace dataset {
+// The dataset code for Databento Equity Basic.
+static constexpr auto kDbeqBasic = "DBEQ.BASIC";
+// The dataset code for CME Globex MDP 3.0.
 static constexpr auto kGlbxMdp3 = "GLBX.MDP3";
+// The dataset code for OPRA.PILLAR.
+static constexpr auto kOpraPillar = "OPRA.PILLAR";
+// The dataset code for Nasdaq TotalView ITCH.
 static constexpr auto kXnasItch = "XNAS.ITCH";
 }  // namespace dataset
 }  // namespace databento

include/databento/symbology.hpp

@@ -27,7 +27,12 @@ struct SymbologyResolution {
 // Converts a vector of symbols to a comma-delineated string for sending to
 // Databento's historical and live APIs.
 //
-// Throws InvalidArgumentError if symbols is empty.
+// Throws InvalidArgumentError if symbols is empty or the iterator range is
+// empty.
+std::string JoinSymbolStrings(
+    const std::string& method_name,
+    std::vector<std::string>::const_iterator symbols_begin,
+    std::vector<std::string>::const_iterator symbols_end);
 std::string JoinSymbolStrings(const std::string& method_name,
                               const std::vector<std::string>& symbols);
 std::string ToString(const StrMappingInterval& mapping_interval);

src/live_blocking.cpp

@@ -3,7 +3,8 @@
 #include <openssl/sha.h>  // SHA256, SHA256_DIGEST_LENGTH
 
 #include <algorithm>  // copy
-#include <cctype>
+#include <cctype>     // tolower
+#include <cstddef>    // ptrdiff_t
 #include <cstdlib>
 #include <ios>  //hex, setfill, setw
 #include <sstream>
@@ -53,16 +54,30 @@ 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)
-          << "|symbols="
-          << JoinSymbolStrings("LiveBlocking::Subscribe", symbols);
-  if (start.time_since_epoch().count()) {
-    sub_msg << "|start=" << start.time_since_epoch().count();
+  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"};
   }
-  sub_msg << '\n';
+  auto symbols_it = symbols.begin();
+  while (symbols_it != symbols.end()) {
+    const auto chunk_size =
+        std::min(kSymbolMaxChunkSize, std::distance(symbols_it, symbols.end()));
 
-  client_.WriteAll(sub_msg.str());
+    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());
+
+    symbols_it += chunk_size;
+  }
 }
 
 void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,

src/symbology.cpp

@@ -7,18 +7,28 @@
 #include "stream_op_helper.hpp"      // StreamOpBuilder
 
 namespace databento {
-std::string JoinSymbolStrings(const std::string& method_name,
-                              const std::vector<std::string>& symbols) {
-  if (symbols.empty()) {
+std::string JoinSymbolStrings(
+    const std::string& method_name,
+    std::vector<std::string>::const_iterator symbols_begin,
+    std::vector<std::string>::const_iterator symbols_end) {
+  if (symbols_begin == symbols_end) {
     throw InvalidArgumentError{method_name, "symbols", "Cannot be empty"};
   }
-  return std::accumulate(symbols.begin(), symbols.end(), std::string{},
+  return std::accumulate(symbols_begin, symbols_end, std::string{},
                          [](std::string acc, const std::string& sym) {
                            return acc.empty() ? sym
                                               : std::move(acc) + ',' + sym;
                          });
 }
 
+std::string JoinSymbolStrings(const std::string& method_name,
+                              const std::vector<std::string>& symbols) {
+  if (symbols.empty()) {
+    throw InvalidArgumentError{method_name, "symbols", "Cannot be empty"};
+  }
+  return JoinSymbolStrings(method_name, symbols.begin(), symbols.end());
+}
+
 std::string ToString(const StrMappingInterval& mapping_interval) {
   return MakeString(mapping_interval);
 }

test/src/live_blocking_tests.cpp

@@ -86,6 +86,33 @@ TEST_F(LiveBlockingTests, TestSubscribe) {
   target.Subscribe(kSymbols, kSchema, kSType);
 }
 
+TEST_F(LiveBlockingTests, TestSubscriptionChunking) {
+  constexpr auto kTsOut = false;
+  constexpr auto kDataset = dataset::kXnasItch;
+  constexpr auto kSymbol = "TEST";
+  constexpr auto kSymbolCount = 1000;
+  constexpr auto kSchema = Schema::Ohlcv1M;
+  constexpr auto kSType = SType::RawSymbol;
+
+  const mock::MockLsgServer mock_server{
+      kDataset, kTsOut, [](mock::MockLsgServer& self) {
+        self.Accept();
+        self.Authenticate();
+        std::size_t i{};
+        while (i < 1000) {
+          const auto chunk_size = std::min(128UL, kSymbolCount - i);
+          const std::vector<std::string> symbols_chunk(chunk_size, kSymbol);
+          self.Subscribe(symbols_chunk, kSchema, kSType);
+          i += chunk_size;
+        }
+      }};
+
+  LiveBlocking target{logger_.get(),      kKey,  kDataset, "127.0.0.1",
+                      mock_server.Port(), kTsOut};
+  const std::vector<std::string> kSymbols(kSymbolCount, kSymbol);
+  target.Subscribe(kSymbols, kSchema, kSType);
+}
+
 TEST_F(LiveBlockingTests, TestNextRecord) {
   constexpr auto kTsOut = false;
   constexpr auto kRecCount = 12;