ADD: Add subscription ID to clients

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.33.0 - TBD
+
+##### Enhancements
+- Added `id` field to `LiveSubscription` requests, which will be used for improved error
+  messages
+
 ## 0.32.1 - 2025-04-07
 
 ### Bug fixes

include/databento/live_blocking.hpp

@@ -96,6 +96,7 @@ class LiveBlocking {
   std::string GenerateCramReply(const std::string& challenge_key);
   std::string EncodeAuthReq(const std::string& auth);
   std::uint64_t DecodeAuthResp();
+  void IncrementSubCounter();
   void Subscribe(const std::string& sub_msg,
                  const std::vector<std::string>& symbols, bool use_snapshot);
   detail::TcpClient::Result FillBuffer(std::chrono::milliseconds timeout);
@@ -113,6 +114,7 @@ class LiveBlocking {
   VersionUpgradePolicy upgrade_policy_;
   std::chrono::seconds heartbeat_interval_;
   detail::TcpClient client_;
+  std::uint32_t sub_counter_{};
   std::vector<LiveSubscription> subscriptions_;
   // Must be 8-byte aligned for records
   alignas(RecordHeader) std::array<char, kMaxStrLen> read_buffer_{};

include/databento/live_subscription.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <cstdint>
 #include <string>
 #include <variant>
 #include <vector>
@@ -17,5 +18,6 @@ struct LiveSubscription {
   Schema schema;
   SType stype_in;
   Start start;
+  std::uint32_t id{};
 };
 }  // namespace databento

src/live_blocking.cpp

@@ -8,6 +8,7 @@
 #include <cstddef>  // ptrdiff_t
 #include <cstdlib>
 #include <ios>  // hex, setfill, setw
+#include <limits>
 #include <sstream>
 #include <variant>
 
@@ -66,42 +67,46 @@ 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) {
+  IncrementSubCounter();
   std::ostringstream sub_msg;
   sub_msg << "schema=" << ToString(schema) << "|stype_in=" << ToString(stype_in)
-          << "|start=" << start.time_since_epoch().count();
+          << "|start=" << start.time_since_epoch().count()
+          << "|id=" << std::to_string(sub_counter_);
   Subscribe(sub_msg.str(), symbols, false);
   subscriptions_.emplace_back(
-      LiveSubscription{symbols, schema, stype_in, start});
+      LiveSubscription{symbols, schema, stype_in, start, sub_counter_});
 }
 
 void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
                              Schema schema, SType stype_in,
                              const std::string& start) {
+  IncrementSubCounter();
   std::ostringstream sub_msg;
-  sub_msg << "schema=" << ToString(schema)
-          << "|stype_in=" << ToString(stype_in);
+  sub_msg << "schema=" << ToString(schema) << "|stype_in=" << ToString(stype_in)
+          << "|id=" << std::to_string(sub_counter_);
   if (!start.empty()) {
     sub_msg << "|start=" << start;
   }
   Subscribe(sub_msg.str(), symbols, false);
   if (start.empty()) {
-    subscriptions_.emplace_back(LiveSubscription{symbols, schema, stype_in,
-                                                 LiveSubscription::NoStart{}});
+    subscriptions_.emplace_back(LiveSubscription{
+        symbols, schema, stype_in, LiveSubscription::NoStart{}, sub_counter_});
   } else {
     subscriptions_.emplace_back(
-        LiveSubscription{symbols, schema, stype_in, start});
+        LiveSubscription{symbols, schema, stype_in, start, sub_counter_});
   }
 }
 
 void LiveBlocking::SubscribeWithSnapshot(
     const std::vector<std::string>& symbols, Schema schema, SType stype_in) {
+  IncrementSubCounter();
   std::ostringstream sub_msg;
-  sub_msg << "schema=" << ToString(schema)
-          << "|stype_in=" << ToString(stype_in);
+  sub_msg << "schema=" << ToString(schema) << "|stype_in=" << ToString(stype_in)
+          << "|id=" << std::to_string(sub_counter_);
 
   Subscribe(sub_msg.str(), symbols, true);
-  subscriptions_.emplace_back(LiveSubscription{symbols, schema, stype_in,
-                                               LiveSubscription::Snapshot{}});
+  subscriptions_.emplace_back(LiveSubscription{
+      symbols, schema, stype_in, LiveSubscription::Snapshot{}, sub_counter_});
 }
 
 void LiveBlocking::Subscribe(const std::string& sub_msg,
@@ -182,6 +187,7 @@ void LiveBlocking::Stop() { client_.Close(); }
 void LiveBlocking::Reconnect() {
   log_receiver_->Receive(LogLevel::Info, "Reconnecting");
   client_ = detail::TcpClient{gateway_, port_};
+  sub_counter_ = 0;
   session_id_ = this->Authenticate();
 }
 
@@ -191,9 +197,11 @@ void LiveBlocking::Resubscribe() {
         std::holds_alternative<std::string>(subscription.start)) {
       subscription.start = LiveSubscription::NoStart{};
     }
+    sub_counter_ = std::max(sub_counter_, subscription.id);
     std::ostringstream sub_msg;
     sub_msg << "schema=" << ToString(subscription.schema)
-            << "|stype_in=" << ToString(subscription.stype_in);
+            << "|stype_in=" << ToString(subscription.stype_in)
+            << "|id=" << std::to_string(sub_counter_);
     Subscribe(
         sub_msg.str(), subscription.symbols,
         std::holds_alternative<LiveSubscription::Snapshot>(subscription.start));
@@ -377,6 +385,16 @@ std::uint64_t LiveBlocking::DecodeAuthResp() {
   return session_id;
 }
 
+void LiveBlocking::IncrementSubCounter() {
+  if (sub_counter_ == std::numeric_limits<uint32_t>::max()) {
+    log_receiver_->Receive(
+        LogLevel::Warning,
+        "[LiveBlocking::Subscribe] Exhausted all subscription IDs");
+  } else {
+    ++sub_counter_;
+  }
+}
+
 databento::detail::TcpClient::Result LiveBlocking::FillBuffer(
     std::chrono::milliseconds timeout) {
   // Shift data forward

tests/src/mock_lsg_server.cpp

@@ -136,6 +136,7 @@ void MockLsgServer::SubscribeWithSnapshot(
   EXPECT_NE(received.find(std::string{"stype_in="} + ToString(stype)),
             std::string::npos);
   EXPECT_EQ(received.find("start="), std::string::npos);
+  EXPECT_NE(received.find("id="), std::string::npos);
   EXPECT_NE(received.find("snapshot=1"), std::string::npos);
 }
 
@@ -156,6 +157,7 @@ void MockLsgServer::Subscribe(const std::vector<std::string>& symbols,
   } else {
     EXPECT_NE(received.find(std::string{"start="} + start), std::string::npos);
   }
+  EXPECT_NE(received.find("id="), std::string::npos);
   EXPECT_NE(received.find("snapshot=0"), std::string::npos);
 }