ADD: Add pull-based TimeseriesGetRange overloads

Author: threecgreen

CHANGELOG.md

@@ -9,6 +9,12 @@
 
 ### Enhancements
 - Added support for `progress` field in `BatchJob` response
+- Added blocking (pull-based) `TimeseriesGetRange` overloads that return a `DbnStore`
+  with `GetMetadata()` and `NextRecord()` methods, unifying the consumption pattern
+  across historical, live, and file sources
+- Generalized `DbnFileStore` into `DbnStore` which accepts any `IReadable` source.
+  `DbnFileStore` is retained as a type alias for backwards compatibility
+- Upgraded default `httplib` version to 0.37.2
 
 ## 0.50.0 - 2026-03-03
 

CMakeLists.txt

@@ -178,7 +178,7 @@ if(${PROJECT_NAME_UPPERCASE}_USE_EXTERNAL_HTTPLIB)
     find_package(httplib REQUIRED)
   endif()
 else()
-  set(httplib_version 0.30.1)
+  set(httplib_version 0.37.2)
   FetchContent_Declare(
     httplib
     URL https://github.com/yhirose/cpp-httplib/archive/refs/tags/v${httplib_version}.tar.gz

README.md

@@ -138,19 +138,15 @@ namespace db = databento;
 
 int main() {
   auto client = db::Historical::Builder().SetKey("$YOUR_API_KEY").Build();
-  db::TsSymbolMap symbol_map;
-  auto decode_symbols = [&symbol_map](const db::Metadata& metadata) {
-    symbol_map = metadata.CreateSymbolMap();
-  };
-  auto print_trades = [&symbol_map](const db::Record& record) {
-    const auto& trade_msg = record.Get<db::TradeMsg>();
-    std::cout << "Received trade for " << symbol_map.At(trade_msg) << ": " << trade_msg
-              << '\n';
-    return db::KeepGoing::Continue;
-  };
-  client.TimeseriesGetRange("GLBX.MDP3", {"2022-06-10T14:30", "2022-06-10T14:40"},
-                            {"ESM2", "NQZ2"}, db::Schema::Trades, db::SType::RawSymbol,
-                            db::SType::InstrumentId, {}, decode_symbols, print_trades);
+  auto store = client.TimeseriesGetRange(
+      "GLBX.MDP3", {"2022-06-10T14:30", "2022-06-10T14:40"},
+      {"ESM2", "NQZ2"}, db::Schema::Trades);
+  auto symbol_map = store.GetMetadata().CreateSymbolMap();
+  while (const auto* record = store.NextRecord()) {
+    const auto& trade_msg = record->Get<db::TradeMsg>();
+    std::cout << "Received trade for " << symbol_map.At(trade_msg) << ": "
+              << trade_msg << '\n';
+  }
 }
 ```
 

cmake/SourcesAndHeaders.cmake

@@ -7,6 +7,7 @@ set(headers
   include/databento/dbn_decoder.hpp
   include/databento/dbn_encoder.hpp
   include/databento/dbn_file_store.hpp
+  include/databento/dbn_store.hpp
   include/databento/detail/buffer.hpp
   include/databento/detail/dbn_buffer_decoder.hpp
   include/databento/detail/http_client.hpp
@@ -39,7 +40,8 @@ set(headers
   include/databento/v2.hpp
   include/databento/v3.hpp
   include/databento/with_ts_out.hpp
-  src/stream_op_helper.hpp
+  src/detail/http_stream_reader.hpp
+  src/detail/stream_op_helper.hpp
 )
 
 set(sources
@@ -49,10 +51,11 @@ set(sources
   src/dbn_constants.hpp
   src/dbn_decoder.cpp
   src/dbn_encoder.cpp
-  src/dbn_file_store.cpp
+  src/dbn_store.cpp
   src/detail/buffer.cpp
   src/detail/dbn_buffer_decoder.cpp
   src/detail/http_client.cpp
+  src/detail/http_stream_reader.cpp
   src/detail/json_helpers.cpp
   src/detail/live_connection.cpp
   src/detail/scoped_fd.cpp

examples/historical/readme.cpp

@@ -9,17 +9,13 @@ namespace db = databento;
 
 int main() {
   auto client = db::Historical::Builder().SetKey("$YOUR_API_KEY").Build();
-  db::TsSymbolMap symbol_map;
-  auto decode_symbols = [&symbol_map](const db::Metadata& metadata) {
-    symbol_map = metadata.CreateSymbolMap();
-  };
-  auto print_trades = [&symbol_map](const db::Record& record) {
-    const auto& trade_msg = record.Get<db::TradeMsg>();
+  auto store =
+      client.TimeseriesGetRange("GLBX.MDP3", {"2022-06-10T14:30", "2022-06-10T14:40"},
+                                {"ESM2", "NQZ2"}, db::Schema::Trades);
+  auto symbol_map = store.GetMetadata().CreateSymbolMap();
+  while (const auto* record = store.NextRecord()) {
+    const auto& trade_msg = record->Get<db::TradeMsg>();
     std::cout << "Received trade for " << symbol_map.At(trade_msg) << ": " << trade_msg
               << '\n';
-    return db::KeepGoing::Continue;
-  };
-  client.TimeseriesGetRange("GLBX.MDP3", {"2022-06-10T14:30", "2022-06-10T14:40"},
-                            {"ESM2", "NQZ2"}, db::Schema::Trades, db::SType::RawSymbol,
-                            db::SType::InstrumentId, {}, decode_symbols, print_trades);
+  }
 }

examples/historical/timeseries_get_range.cpp

@@ -9,15 +9,14 @@ namespace db = databento;
 int main() {
   auto client = db::Historical::Builder().SetKeyFromEnv().Build();
   const auto limit = 1000;
-  client.TimeseriesGetRange(
+  auto store = client.TimeseriesGetRange(
       db::dataset::kGlbxMdp3, db::DateTimeRange<std::string>{"2022-10-03"}, {"ESZ2"},
-      db::Schema::Trades, db::SType::RawSymbol, db::SType::InstrumentId, limit,
-      [](db::Metadata&& metadata) { std::cout << metadata << '\n'; },
-      [](const db::Record& record) {
-        const auto& trade_msg = record.Get<db::TradeMsg>();
-        std::cout << trade_msg << '\n';
-        return db::KeepGoing::Continue;
-      });
+      db::Schema::Trades, db::SType::RawSymbol, db::SType::InstrumentId, limit);
+  std::cout << store.GetMetadata() << '\n';
+  while (const auto* record = store.NextRecord()) {
+    const auto& trade_msg = record->Get<db::TradeMsg>();
+    std::cout << trade_msg << '\n';
+  }
 
   return 0;
 }

examples/historical/timeseries_get_range_to_file.cpp

@@ -1,6 +1,6 @@
 #include "databento/constants.hpp"
 #include "databento/datetime.hpp"
-#include "databento/dbn_file_store.hpp"
+#include "databento/dbn_store.hpp"
 #include "databento/enums.hpp"
 #include "databento/historical.hpp"
 #include "databento/record.hpp"
@@ -10,14 +10,13 @@ namespace db = databento;
 int main() {
   auto client = db::Historical::Builder().SetKeyFromEnv().Build();
   const auto limit = 1000;
-  db::DbnFileStore dbn_file_store = client.TimeseriesGetRangeToFile(
+  db::DbnStore dbn_store = client.TimeseriesGetRangeToFile(
       db::dataset::kGlbxMdp3, {"2022-10-03T00:00", "2022-10-04T00:00"}, {"ESZ2"},
       db::Schema::Ohlcv1M, db::SType::RawSymbol, db::SType::InstrumentId, limit,
       "ESZ2-ohlcv1m-20201003-20201004.dbn.zst");
-  dbn_file_store.Replay([](const db::Record record) {
-    const auto& ohlcv_bar = record.Get<db::OhlcvMsg>();
+  while (const auto* record = dbn_store.NextRecord()) {
+    const auto& ohlcv_bar = record->Get<db::OhlcvMsg>();
     std::cout << ohlcv_bar << '\n';
-    return db::KeepGoing::Continue;
-  });
+  }
   return 0;
 }

include/databento/dbn_file_store.hpp

@@ -1,40 +1,7 @@
 #pragma once
 
-#include <filesystem>  // path
-
-#include "databento/dbn.hpp"          // DecodeMetadata
-#include "databento/dbn_decoder.hpp"  // DbnDecoder
-#include "databento/enums.hpp"        // VersionUpgradePolicy
-#include "databento/log.hpp"
-#include "databento/record.hpp"
-#include "databento/timeseries.hpp"  // MetadataCallback, RecordCallback
+#include "databento/dbn_store.hpp"
 
 namespace databento {
-// A reader for DBN files. This class provides both a callback API similar to
-// TimeseriesGetRange in historical data and LiveThreaded for live data as well
-// as a blocking API similar to that of LiveBlocking. Only one API should be
-// used on a given instance.
-class DbnFileStore {
- public:
-  explicit DbnFileStore(const std::filesystem::path& file_path);
-  DbnFileStore(ILogReceiver* log_receiver, const std::filesystem::path& file_path,
-               VersionUpgradePolicy upgrade_policy);
-
-  // Callback API: calling Replay consumes the input.
-  void Replay(const MetadataCallback& metadata_callback,
-              const RecordCallback& record_callback);
-  void Replay(const RecordCallback& record_callback);
-
-  // Blocking API
-  const Metadata& GetMetadata();
-  // Returns the next record or `nullptr` if there are no remaining records.
-  const Record* NextRecord();
-
- private:
-  void MaybeDecodeMetadata();
-
-  DbnDecoder decoder_;
-  Metadata metadata_{};
-  bool has_decoded_metadata_{false};
-};
-}  // namespace databento
+using DbnFileStore = DbnStore;
+}

include/databento/dbn_store.hpp

@@ -0,0 +1,43 @@
+#pragma once
+
+#include <filesystem>  // path
+#include <memory>      // unique_ptr
+
+#include "databento/dbn.hpp"          // DecodeMetadata
+#include "databento/dbn_decoder.hpp"  // DbnDecoder
+#include "databento/enums.hpp"        // VersionUpgradePolicy
+#include "databento/ireadable.hpp"
+#include "databento/log.hpp"
+#include "databento/record.hpp"
+#include "databento/timeseries.hpp"  // MetadataCallback, RecordCallback
+
+namespace databento {
+// A reader for DBN data from files or streams. This class provides both a callback API
+// similar to `LiveThreaded` for live data as well as a blocking API similar to that of
+// `LiveBlocking`. Only one API should be used on a given instance.
+class DbnStore {
+ public:
+  explicit DbnStore(const std::filesystem::path& file_path);
+  DbnStore(ILogReceiver* log_receiver, const std::filesystem::path& file_path,
+           VersionUpgradePolicy upgrade_policy);
+  DbnStore(ILogReceiver* log_receiver, std::unique_ptr<IReadable> input,
+           VersionUpgradePolicy upgrade_policy);
+
+  // Callback API: calling Replay consumes the input.
+  void Replay(const MetadataCallback& metadata_callback,
+              const RecordCallback& record_callback);
+  void Replay(const RecordCallback& record_callback);
+
+  // Blocking API
+  const databento::Metadata& GetMetadata();
+  // Returns the next record or `nullptr` if there are no remaining records.
+  const Record* NextRecord();
+
+ private:
+  void MaybeDecodeMetadata();
+
+  DbnDecoder decoder_;
+  databento::Metadata metadata_{};
+  bool has_decoded_metadata_{false};
+};
+}  // namespace databento

include/databento/detail/http_client.hpp

@@ -9,10 +9,12 @@
 #include <nlohmann/json.hpp>
 
 #include <cstdint>
+#include <memory>  // unique_ptr
 #include <string>
 
 namespace databento {
 class ILogReceiver;
+class IReadable;
 namespace detail {
 class HttpClient {
  public:
@@ -27,6 +29,8 @@ class HttpClient {
                     const httplib::ContentReceiver& callback);
   void PostRawStream(const std::string& path, const httplib::Params& form_params,
                      const httplib::ContentReceiver& callback);
+  std::unique_ptr<IReadable> OpenPostStream(const std::string& path,
+                                            const httplib::Params& form_params);
 
  private:
   static bool IsErrorStatus(int status_code);