ADD: Add TryNextRecord and FillBuffer to C++

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.53.0 - TBD
+
+### Enhancements
+- Added `TryNextRecord` and `FillBuffer` to `LiveBlocking` for more fine-grained
+  control around I/O
+
 ## 0.52.0 - 2026-03-31
 
 ### Enhancements

include/databento/live_blocking.hpp

@@ -81,6 +81,26 @@ class LiveBlocking {
   //
   // This method should only be called after `Start`.
   const Record* NextRecord(std::chrono::milliseconds timeout);
+  // Returns the next record from the internal buffer without performing any
+  // I/O. Returns `nullptr` if no complete record is buffered. The returned
+  // pointer is valid until the next call to `TryNextRecord`, `NextRecord`,
+  // or `FillBuffer`.
+  //
+  // This method should only be called after `Start`.
+  const Record* TryNextRecord();
+  // Reads available data from the connection into the internal buffer using
+  // the heartbeat timeout. Returns the number of bytes read and the status.
+  // A `read_size` of 0 with `Status::Closed` indicates the connection was
+  // closed by the gateway.
+  //
+  // This method should only be called after `Start`.
+  IReadable::Result FillBuffer();
+  // Reads available data from the connection into the internal buffer.
+  // Returns the number of bytes read and the status. A `read_size` of 0 with
+  // `Status::Closed` indicates the connection was closed by the gateway.
+  //
+  // This method should only be called after `Start`.
+  IReadable::Result FillBuffer(std::chrono::milliseconds timeout);
   // Stops the session with the gateway. Once stopped, the session cannot be
   // restarted.
   void Stop();
@@ -117,7 +137,7 @@ class LiveBlocking {
   void IncrementSubCounter();
   void Subscribe(std::string_view sub_msg, const std::vector<std::string>& symbols,
                  bool use_snapshot);
-  IReadable::Result FillBuffer(std::chrono::milliseconds timeout);
+  const Record* ConsumeBufferedRecord();
   RecordHeader* BufferRecordHeader();
   std::chrono::milliseconds HeartbeatTimeout() const;
   void CheckHeartbeatTimeout() const;

src/live_blocking.cpp

@@ -196,9 +196,8 @@ const databento::Record& LiveBlocking::NextRecord() {
 }
 
 const databento::Record* LiveBlocking::NextRecord(std::chrono::milliseconds timeout) {
-  // need some unread_bytes
-  const auto unread_bytes = buffer_.ReadCapacity();
-  if (unread_bytes == 0) {
+  // need at least a header to read the record size
+  if (buffer_.ReadCapacity() < sizeof(RecordHeader)) {
     const auto read_res = FillBuffer(timeout);
     if (read_res.status == Status::Timeout) {
       CheckHeartbeatTimeout();
@@ -208,7 +207,7 @@ const databento::Record* LiveBlocking::NextRecord(std::chrono::milliseconds time
       throw LiveApiError{"Gateway closed the session"};
     }
   }
-  // check length
+  // wait for the full record
   while (buffer_.ReadCapacity() < BufferRecordHeader()->Size()) {
     const auto read_res = FillBuffer(timeout);
     if (read_res.status == Status::Timeout) {
@@ -219,12 +218,17 @@ const databento::Record* LiveBlocking::NextRecord(std::chrono::milliseconds time
       throw LiveApiError{"Gateway closed the session"};
     }
   }
-  current_record_ = Record{BufferRecordHeader()};
-  const auto bytes_to_consume = current_record_.Size();
-  buffer_.ConsumeNoShift(bytes_to_consume);
-  current_record_ = DbnDecoder::DecodeRecordCompat(
-      version_, upgrade_policy_, send_ts_out_, &compat_buffer_, current_record_);
-  return &current_record_;
+  return ConsumeBufferedRecord();
+}
+
+const databento::Record* LiveBlocking::TryNextRecord() {
+  if (buffer_.ReadCapacity() < sizeof(RecordHeader)) {
+    return nullptr;
+  }
+  if (buffer_.ReadCapacity() < BufferRecordHeader()->Size()) {
+    return nullptr;
+  }
+  return ConsumeBufferedRecord();
 }
 
 void LiveBlocking::Stop() { connection_.Close(); }
@@ -447,6 +451,10 @@ void LiveBlocking::IncrementSubCounter() {
   }
 }
 
+databento::IReadable::Result LiveBlocking::FillBuffer() {
+  return FillBuffer(HeartbeatTimeout());
+}
+
 databento::IReadable::Result LiveBlocking::FillBuffer(
     std::chrono::milliseconds timeout) {
   buffer_.Shift();
@@ -459,6 +467,14 @@ databento::IReadable::Result LiveBlocking::FillBuffer(
   return read_res;
 }
 
+const databento::Record* LiveBlocking::ConsumeBufferedRecord() {
+  current_record_ = Record{BufferRecordHeader()};
+  buffer_.ConsumeNoShift(current_record_.Size());
+  current_record_ = DbnDecoder::DecodeRecordCompat(
+      version_, upgrade_policy_, send_ts_out_, &compat_buffer_, current_record_);
+  return &current_record_;
+}
+
 databento::RecordHeader* LiveBlocking::BufferRecordHeader() {
   return reinterpret_cast<RecordHeader*>(buffer_.ReadBegin());
 }

tests/src/live_blocking_tests.cpp

@@ -15,6 +15,7 @@
 #include "databento/datetime.hpp"
 #include "databento/enums.hpp"  // Schema, SType
 #include "databento/exceptions.hpp"
+#include "databento/ireadable.hpp"
 #include "databento/live.hpp"
 #include "databento/live_blocking.hpp"
 #include "databento/live_subscription.hpp"
@@ -765,4 +766,149 @@ TEST_F(LiveBlockingTests, TestHeartbeatTimeoutOnNextRecordWithTimeout) {
   EXPECT_TRUE(got_timeout_exception) << "Expected heartbeat timeout exception";
 }
 
+TEST_F(LiveBlockingTests, TestTryNextRecordEmptyBuffer) {
+  constexpr auto kTsOut = false;
+  const mock::MockLsgServer mock_server{
+      dataset::kXnasItch, kTsOut, [](mock::MockLsgServer& self) {
+        self.Accept();
+        self.Authenticate();
+        std::this_thread::sleep_for(std::chrono::milliseconds{200});
+      }};
+
+  LiveBlocking target = builder_.SetDataset(dataset::kXnasItch)
+                            .SetSendTsOut(kTsOut)
+                            .SetAddress(kLocalhost, mock_server.Port())
+                            .BuildBlocking();
+  // Buffer is empty, no I/O should be performed
+  const auto* rec = target.TryNextRecord();
+  EXPECT_EQ(rec, nullptr);
+}
+
+TEST_F(LiveBlockingTests, TestTryNextRecordAfterFillBuffer) {
+  constexpr auto kTsOut = false;
+  constexpr OhlcvMsg kRec{DummyHeader<OhlcvMsg>(RType::Ohlcv1M), 1, 2, 3, 4, 5};
+  const mock::MockLsgServer mock_server{dataset::kXnasItch, kTsOut,
+                                        [kRec](mock::MockLsgServer& self) {
+                                          self.Accept();
+                                          self.Authenticate();
+                                          self.SendRecord(kRec);
+                                        }};
+
+  LiveBlocking target = builder_.SetDataset(dataset::kXnasItch)
+                            .SetSendTsOut(kTsOut)
+                            .SetAddress(kLocalhost, mock_server.Port())
+                            .BuildBlocking();
+  const auto fill_res = target.FillBuffer(std::chrono::milliseconds{1000});
+  ASSERT_EQ(fill_res.status, IReadable::Status::Ok);
+  ASSERT_GT(fill_res.read_size, 0);
+  const auto* rec = target.TryNextRecord();
+  ASSERT_NE(rec, nullptr);
+  ASSERT_TRUE(rec->Holds<OhlcvMsg>());
+  EXPECT_EQ(rec->Get<OhlcvMsg>(), kRec);
+  // Buffer drained
+  EXPECT_EQ(target.TryNextRecord(), nullptr);
+}
+
+TEST_F(LiveBlockingTests, TestFillBufferReturnsClosed) {
+  constexpr auto kTsOut = false;
+  const mock::MockLsgServer mock_server{dataset::kXnasItch, kTsOut,
+                                        [](mock::MockLsgServer& self) {
+                                          self.Accept();
+                                          self.Authenticate();
+                                          self.Close();
+                                        }};
+
+  LiveBlocking target = builder_.SetDataset(dataset::kXnasItch)
+                            .SetSendTsOut(kTsOut)
+                            .SetAddress(kLocalhost, mock_server.Port())
+                            .BuildBlocking();
+  const auto fill_res = target.FillBuffer(std::chrono::milliseconds{1000});
+  EXPECT_EQ(fill_res.status, IReadable::Status::Closed);
+  EXPECT_EQ(fill_res.read_size, 0);
+}
+
+TEST_F(LiveBlockingTests, TestTryNextRecordPollLoop) {
+  constexpr auto kTsOut = false;
+  constexpr auto kRecCount = 5;
+  constexpr OhlcvMsg kRec{DummyHeader<OhlcvMsg>(RType::Ohlcv1M), 1, 2, 3, 4, 5};
+  const mock::MockLsgServer mock_server{dataset::kXnasItch, kTsOut,
+                                        [kRec](mock::MockLsgServer& self) {
+                                          self.Accept();
+                                          self.Authenticate();
+                                          for (size_t i = 0; i < kRecCount; ++i) {
+                                            self.SendRecord(kRec);
+                                          }
+                                          self.Close();
+                                        }};
+
+  LiveBlocking target = builder_.SetDataset(dataset::kXnasItch)
+                            .SetSendTsOut(kTsOut)
+                            .SetAddress(kLocalhost, mock_server.Port())
+                            .BuildBlocking();
+  int record_count = 0;
+  while (true) {
+    while (const auto* rec = target.TryNextRecord()) {
+      ASSERT_TRUE(rec->Holds<OhlcvMsg>());
+      EXPECT_EQ(rec->Get<OhlcvMsg>(), kRec);
+      ++record_count;
+    }
+    const auto fill_res = target.FillBuffer(std::chrono::milliseconds{1000});
+    if (fill_res.status == IReadable::Status::Closed) {
+      break;
+    }
+  }
+  EXPECT_EQ(record_count, kRecCount);
+}
+
+TEST_F(LiveBlockingTests, TestTryNextRecordPartialRecord) {
+  constexpr auto kTsOut = false;
+  constexpr MboMsg kRec{DummyHeader<MboMsg>(RType::Mbo),
+                        1,
+                        2,
+                        3,
+                        {},
+                        4,
+                        Action::Add,
+                        Side::Bid,
+                        UnixNanos{},
+                        TimeDeltaNanos{},
+                        100};
+
+  bool send_remaining{};
+  std::mutex send_remaining_mutex;
+  std::condition_variable send_remaining_cv;
+  const mock::MockLsgServer mock_server{
+      dataset::kGlbxMdp3, kTsOut,
+      [kRec, &send_remaining, &send_remaining_mutex,
+       &send_remaining_cv](mock::MockLsgServer& self) {
+        self.Accept();
+        self.Authenticate();
+        self.SplitSendRecord(kRec, send_remaining, send_remaining_mutex,
+                             send_remaining_cv);
+      }};
+
+  LiveBlocking target = builder_.SetDataset(dataset::kGlbxMdp3)
+                            .SetSendTsOut(kTsOut)
+                            .SetAddress(kLocalhost, mock_server.Port())
+                            .BuildBlocking();
+  // Read partial record (just header)
+  auto fill_res = target.FillBuffer(std::chrono::milliseconds{1000});
+  ASSERT_EQ(fill_res.status, IReadable::Status::Ok);
+  // Record is incomplete
+  EXPECT_EQ(target.TryNextRecord(), nullptr);
+  // Signal server to send remaining bytes
+  {
+    const std::lock_guard<std::mutex> lock{send_remaining_mutex};
+    send_remaining = true;
+    send_remaining_cv.notify_one();
+  }
+  // Read the rest
+  fill_res = target.FillBuffer(std::chrono::milliseconds{1000});
+  ASSERT_EQ(fill_res.status, IReadable::Status::Ok);
+  const auto* rec = target.TryNextRecord();
+  ASSERT_NE(rec, nullptr);
+  ASSERT_TRUE(rec->Holds<MboMsg>());
+  EXPECT_EQ(rec->Get<MboMsg>(), kRec);
+}
+
 }  // namespace databento::tests