VER: Release 0.30.0

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## 0.30.0 - 2025-02-11
+
+### Enhancements
+- Added `Resubscribe()` methods to `LiveBlocking` and `LiveThreaded` to make it easier
+  to resume a live session after losing the connection to the live gateway
+- Added `Subscriptions()` getter methods to `LiveBlocking` and `LiveThreaded` for
+  getting all active subscriptions
+- Added `CommoditySpot` `InstrumentClass` variant
+
 ## 0.29.0 - 2025-02-04
 
 ### Enhancements

CMakeLists.txt

@@ -4,7 +4,7 @@ cmake_minimum_required(VERSION 3.14)
 # Project details
 #
 
-project("databento" VERSION 0.29.0 LANGUAGES CXX)
+project("databento" VERSION 0.30.0 LANGUAGES CXX)
 string(TOUPPER ${PROJECT_NAME} PROJECT_NAME_UPPERCASE)
 
 #

cmake/SourcesAndHeaders.cmake

@@ -23,6 +23,7 @@ set(headers
   include/databento/ireadable.hpp
   include/databento/live.hpp
   include/databento/live_blocking.hpp
+  include/databento/live_subscription.hpp
   include/databento/live_threaded.hpp
   include/databento/log.hpp
   include/databento/metadata.hpp

examples/historical/metadata.cpp

@@ -49,11 +49,11 @@ int main() {
 
   const auto all_unit_prices = client.MetadataListUnitPrices(kGlbxMdp3);
   std::cout << "Unit prices:\n";
-  for (const auto& mode_and_prices : all_unit_prices) {
-    const auto* mode_str = ToString(mode_and_prices.mode);
-    for (const auto& schema_and_price : mode_and_prices.unit_prices) {
-      std::cout << "- (" << mode_str << ", " << schema_and_price.first
-                << "): " << schema_and_price.second << '\n';
+  for (const auto& [mode, unit_prices] : all_unit_prices) {
+    const auto* mode_str = ToString(mode);
+    for (const auto [schema, price] : unit_prices) {
+      std::cout << "- (" << mode_str << ", " << schema << "): " << price
+                << '\n';
     }
   }
   std::cout << '\n';

examples/live/live_smoke_test.cpp

@@ -11,7 +11,6 @@
 #include <sstream>
 #include <stdexcept>
 #include <string>
-#include <unordered_map>
 
 using namespace databento;
 
@@ -28,7 +27,6 @@ std::vector<std::string> SplitSymbols(const std::string& symbols) {
 }
 
 std::pair<bool, UnixNanos> TryConvertToUnixNanos(const char* start) {
-  std::stringstream ss(start);
   std::size_t pos;
   const uint64_t result = std::stoul(start, &pos, 10);
   if (pos != std::strlen(start)) {
@@ -187,10 +185,10 @@ int main(int argc, char* argv[]) {
   if (use_snapshot) {
     client.SubscribeWithSnapshot(symbols, schema, stype);
   } else if (start) {
-    const auto converted = TryConvertToUnixNanos(start);
-    if (converted.first) {
-      start_from_epoch = converted.second.time_since_epoch().count() == 0;
-      client.Subscribe(symbols, schema, stype, converted.second);
+    const auto [success, start_nanos] = TryConvertToUnixNanos(start);
+    if (success) {
+      start_from_epoch = start_nanos.time_since_epoch().count() == 0;
+      client.Subscribe(symbols, schema, stype, start_nanos);
     } else {
       client.Subscribe(symbols, schema, stype, start);
     }

examples/live/simple.cpp

@@ -16,8 +16,8 @@ static std::sig_atomic_t volatile gSignal;
 
 int main() {
   databento::PitSymbolMap symbol_mappings;
-  std::unique_ptr<databento::ILogReceiver> log_receiver{
-      new databento::ConsoleLogReceiver{databento::LogLevel::Debug}};
+  auto log_receiver = std::make_unique<databento::ConsoleLogReceiver>(
+      databento::LogLevel::Debug);
 
   auto client = databento::LiveBuilder{}
                     .SetLogReceiver(log_receiver.get())

include/databento/detail/json_helpers.hpp

@@ -15,8 +15,7 @@ namespace httplib {
 using Params = std::multimap<std::string, std::string>;
 }
 
-namespace databento {
-namespace detail {
+namespace databento::detail {
 void SetIfNotEmpty(httplib::Params* params, const std::string& key,
                    const std::string& value);
 void SetIfNotEmpty(httplib::Params* params, const std::string& key,
@@ -95,5 +94,4 @@ date::year_month_day ParseAt(const std::string& endpoint,
                              const nlohmann::json& json,
                              const std::string& key);
 
-}  // namespace detail
-}  // namespace databento
+}  // namespace databento::detail

include/databento/detail/scoped_fd.hpp

@@ -4,8 +4,7 @@
 #include <winsock2.h>  // SOCKET
 #endif
 
-namespace databento {
-namespace detail {
+namespace databento::detail {
 #ifdef _WIN32
 using Socket = SOCKET;
 #else
@@ -35,5 +34,4 @@ class ScopedFd {
  private:
   Socket fd_{kUnset};
 };
-}  // namespace detail
-}  // namespace databento
+}  // namespace databento::detail

include/databento/detail/scoped_thread.hpp

@@ -3,8 +3,7 @@
 #include <thread>
 #include <utility>  // forward, move
 
-namespace databento {
-namespace detail {
+namespace databento::detail {
 // An RAII thread that joins if necessary on destruction, like std::jthread in
 // C++20.
 class ScopedThread {
@@ -37,5 +36,4 @@ class ScopedThread {
  private:
   std::thread thread_;
 };
-}  // namespace detail
-}  // namespace databento
+}  // namespace databento::detail

include/databento/detail/shared_channel.hpp

@@ -6,8 +6,7 @@
 
 #include "databento/ireadable.hpp"
 
-namespace databento {
-namespace detail {
+namespace databento::detail {
 // Copyable, thread-safe, unidirectional channel.
 class SharedChannel : public IReadable {
  public:
@@ -28,5 +27,4 @@ class SharedChannel : public IReadable {
 
   std::shared_ptr<Channel> channel_;
 };
-}  // namespace detail
-}  // namespace databento
+}  // namespace databento::detail

include/databento/detail/tcp_client.hpp

@@ -6,8 +6,7 @@
 
 #include "databento/detail/scoped_fd.hpp"  // ScopedFd
 
-namespace databento {
-namespace detail {
+namespace databento::detail {
 class TcpClient {
  public:
   enum class Status : std::uint8_t {
@@ -47,5 +46,4 @@ class TcpClient {
 
   ScopedFd socket_;
 };
-}  // namespace detail
-}  // namespace databento
+}  // namespace databento::detail

include/databento/detail/zstd_stream.hpp

@@ -11,8 +11,7 @@
 #include "databento/iwritable.hpp"
 #include "databento/log.hpp"
 
-namespace databento {
-namespace detail {
+namespace databento::detail {
 class ZstdDecodeStream : public IReadable {
  public:
   explicit ZstdDecodeStream(std::unique_ptr<IReadable> input);
@@ -54,5 +53,4 @@ class ZstdCompressStream : public IWritable {
   std::size_t in_size_;
   std::vector<std::uint8_t> out_buffer_;
 };
-}  // namespace detail
-}  // namespace databento
+}  // namespace databento::detail

include/databento/enums.hpp

@@ -191,6 +191,7 @@ enum InstrumentClass : char {
   FutureSpread = 'S',
   OptionSpread = 'T',
   FxSpot = 'X',
+  CommoditySpot = 'Y',
 };
 }  // namespace instrument_class
 using instrument_class::InstrumentClass;

include/databento/live_blocking.hpp

@@ -10,7 +10,8 @@
 #include "databento/datetime.hpp"           // UnixNanos
 #include "databento/dbn.hpp"                // Metadata
 #include "databento/detail/tcp_client.hpp"  // TcpClient
-#include "databento/enums.hpp"   // Schema, SType, VersionUpgradePolicy
+#include "databento/enums.hpp"  // Schema, SType, VersionUpgradePolicy
+#include "databento/live_subscription.hpp"
 #include "databento/record.hpp"  // Record, RecordHeader
 
 namespace databento {
@@ -44,6 +45,10 @@ class LiveBlocking {
   std::pair<bool, std::chrono::seconds> HeartbeatInterval() const {
     return {heartbeat_interval_.count() > 0, heartbeat_interval_};
   }
+  const std::vector<LiveSubscription>& Subscriptions() const {
+    return subscriptions_;
+  }
+  std::vector<LiveSubscription>& Subscriptions() { return subscriptions_; }
 
   /*
    * Methods
@@ -80,6 +85,9 @@ class LiveBlocking {
   void Stop();
   // Closes the current connection and attempts to reconnect to the gateway.
   void Reconnect();
+  // Resubscribes to all subscriptions, removing the original `start` time, if
+  // any. Usually performed after a `Reconnect()`.
+  void Resubscribe();
 
  private:
   std::string DetermineGateway() const;
@@ -105,6 +113,7 @@ class LiveBlocking {
   VersionUpgradePolicy upgrade_policy_;
   std::chrono::seconds heartbeat_interval_;
   detail::TcpClient client_;
+  std::vector<LiveSubscription> subscriptions_;
   // Must be 8-byte aligned for records
   alignas(RecordHeader) std::array<char, kMaxStrLen> read_buffer_{};
   std::size_t buffer_size_{};

include/databento/live_subscription.hpp

@@ -0,0 +1,21 @@
+#pragma once
+
+#include <string>
+#include <variant>
+#include <vector>
+
+#include "databento/datetime.hpp"  // UnixNanos
+#include "databento/enums.hpp"     // Schema, SType
+
+namespace databento {
+struct LiveSubscription {
+  struct Snapshot {};
+  struct NoStart {};
+  using Start = std::variant<Snapshot, UnixNanos, std::string, NoStart>;
+
+  std::vector<std::string> symbols;
+  Schema schema;
+  SType stype_in;
+  Start start;
+};
+}  // namespace databento

include/databento/live_threaded.hpp

@@ -10,6 +10,7 @@
 #include "databento/datetime.hpp"              // UnixNanos
 #include "databento/detail/scoped_thread.hpp"  // ScopedThread
 #include "databento/enums.hpp"                 // Schema, SType
+#include "databento/live_subscription.hpp"
 #include "databento/timeseries.hpp"  // MetadataCallback, RecordCallback
 
 namespace databento {
@@ -57,6 +58,8 @@ class LiveThreaded {
   // The the first member of the pair will be true, when the heartbeat interval
   // was overridden.
   std::pair<bool, std::chrono::seconds> HeartbeatInterval() const;
+  const std::vector<LiveSubscription>& Subscriptions() const;
+  std::vector<LiveSubscription>& Subscriptions();
 
   /*
    * Methods
@@ -86,6 +89,7 @@ class LiveThreaded {
              ExceptionCallback exception_callback);
   // Closes the current connection, and attempts to reconnect to the gateway.
   void Reconnect();
+  void Resubscribe();
   // Blocking wait with an optional timeout for the session to close when the
   // record_callback or the exception_callback return Stop.
   void BlockForStop();

include/databento/record.hpp

@@ -46,15 +46,10 @@ struct RecordHeader {
 };
 
 // Type trait helper for templated functions accepting DBN records.
-namespace detail {
-// std::void_t added in C++17
-template <typename... Ts>
-using void_t = void;
-}  // namespace detail
-template <typename, typename = detail::void_t<>>
+template <typename, typename = std::void_t<>>
 struct has_header : std::false_type {};
 template <typename T>
-struct has_header<T, detail::void_t<decltype(std::declval<T>().hd)>>
+struct has_header<T, std::void_t<decltype(std::declval<T>().hd)>>
     : std::is_same<decltype(std::declval<T>().hd), RecordHeader> {};
 
 class Record {

include/databento/v2.hpp

@@ -3,8 +3,7 @@
 #include "databento/constants.hpp"  // kSymbolCstrLen
 #include "databento/record.hpp"
 
-namespace databento {
-namespace v2 {
+namespace databento::v2 {
 static constexpr std::size_t kSymbolCstrLen = databento::kSymbolCstrLen;
 
 using MboMsg = databento::MboMsg;
@@ -28,5 +27,4 @@ using StatMsg = databento::StatMsg;
 using ErrorMsg = databento::ErrorMsg;
 using SymbolMappingMsg = databento::SymbolMappingMsg;
 using SystemMsg = databento::SystemMsg;
-}  // namespace v2
-}  // namespace databento
+}  // namespace databento::v2

include/databento/v3.hpp

@@ -8,8 +8,7 @@
 #include "databento/enums.hpp"  // InstrumentClass, MatchingAlgorithm, RType, SecurityUpdateAction, Side, UserDefinedInstrument
 #include "databento/record.hpp"  // RecordHeader
 
-namespace databento {
-namespace v3 {
+namespace databento::v3 {
 static constexpr std::size_t kSymbolCstrLen = databento::kSymbolCstrLen;
 
 using MboMsg = databento::MboMsg;
@@ -142,5 +141,4 @@ inline bool operator!=(const InstrumentDefMsg& lhs,
 std::string ToString(const InstrumentDefMsg& instr_def_msg);
 std::ostream& operator<<(std::ostream& stream,
                          const InstrumentDefMsg& instr_def_msg);
-}  // namespace v3
-}  // namespace databento
+}  // namespace databento::v3

pkg/PKGBUILD

@@ -1,7 +1,7 @@
 # Maintainer: Databento <support@databento.com>
 _pkgname=databento-cpp
 pkgname=databento-cpp-git
-pkgver=0.29.0
+pkgver=0.30.0
 pkgrel=1
 pkgdesc="Official C++ client for Databento"
 arch=('any')

src/dbn_decoder.cpp

@@ -65,12 +65,11 @@ date::year_month_day DecodeIso8601Date(std::uint32_t yyyymmdd_int) {
 DbnDecoder::DbnDecoder(ILogReceiver* log_receiver,
                        detail::SharedChannel channel)
     : DbnDecoder(log_receiver,
-                 std::unique_ptr<IReadable>{
-                     new detail::SharedChannel{std::move(channel)}}) {}
+                 std::make_unique<detail::SharedChannel>(std::move(channel))) {}
 
 DbnDecoder::DbnDecoder(ILogReceiver* log_receiver, InFileStream file_stream)
-    : DbnDecoder(log_receiver, std::unique_ptr<IReadable>{
-                                   new InFileStream{std::move(file_stream)}}) {}
+    : DbnDecoder(log_receiver,
+                 std::make_unique<InFileStream>(std::move(file_stream))) {}
 
 DbnDecoder::DbnDecoder(ILogReceiver* log_receiver,
                        std::unique_ptr<IReadable> input)
@@ -85,9 +84,8 @@ DbnDecoder::DbnDecoder(ILogReceiver* log_receiver,
       input_{std::move(input)} {
   read_buffer_.reserve(kBufferCapacity);
   if (DetectCompression()) {
-    input_ =
-        std::unique_ptr<detail::ZstdDecodeStream>(new detail::ZstdDecodeStream(
-            std::move(input_), std::move(read_buffer_)));
+    input_ = std::make_unique<detail::ZstdDecodeStream>(
+        std::move(input_), std::move(read_buffer_));
     // Reinitialize buffer and get it into the same state as uncompressed input
     read_buffer_ = std::vector<std::uint8_t>();
     read_buffer_.reserve(kBufferCapacity);
@@ -192,10 +190,10 @@ databento::Metadata DbnDecoder::DecodeMetadata() {
   // already read first 4 bytes detecting compression
   read_buffer_.resize(kMetadataPreludeSize);
   input_->ReadExact(&read_buffer_[4], 4);
-  const auto version_and_size = DbnDecoder::DecodeMetadataVersionAndSize(
+  const auto [version, size] = DbnDecoder::DecodeMetadataVersionAndSize(
       read_buffer_.data(), kMetadataPreludeSize);
-  version_ = version_and_size.first;
-  read_buffer_.resize(version_and_size.second);
+  version_ = version;
+  read_buffer_.resize(size);
   input_->ReadExact(read_buffer_.data(), read_buffer_.size());
   buffer_idx_ = read_buffer_.size();
   auto metadata = DbnDecoder::DecodeMetadataFields(version_, read_buffer_);