VER: Release 0.47.0

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Changelog
 
+## 0.47.0 - 2026-02-04
+
+### Enhancements
+- Added Zstd compression support to live clients which can be enabled with
+`LiveBuilder::SetCompression()`. It's disabled by default
+- Added `Compression()` getter to `LiveBlocking` and `LiveThreaded`
+- Upgraded default `httplib` version to 0.30.1
+
+### Breaking changes
+- Added an overload to the `ReadSome` method on `IReadable` for timeout support
+
 ## 0.46.1 - 2026-01-27
 
 ### Enhancements

CMakeLists.txt

@@ -6,7 +6,7 @@ cmake_minimum_required(VERSION 3.24..4.2)
 
 project(
   databento
-  VERSION 0.46.1
+  VERSION 0.47.0
   LANGUAGES CXX
   DESCRIPTION "Official Databento client library"
 )
@@ -128,10 +128,6 @@ endif()
 
 find_package(OpenSSL REQUIRED)
 find_package(zstd REQUIRED)
-if (APPLE)
-  find_library(CORE_FOUNDATION_LIB CoreFoundation REQUIRED)
-  find_library(CFNETWORK_LIB CFNetwork REQUIRED)
-endif()
 if(NOT TARGET zstd::libzstd)
   if(TARGET zstd::libzstd_shared)
     add_library(zstd::libzstd ALIAS zstd::libzstd_shared)
@@ -182,7 +178,7 @@ if(${PROJECT_NAME_UPPERCASE}_USE_EXTERNAL_HTTPLIB)
     find_package(httplib REQUIRED)
   endif()
 else()
-  set(httplib_version 0.28.0)
+  set(httplib_version 0.30.1)
   FetchContent_Declare(
     httplib
     URL https://github.com/yhirose/cpp-httplib/archive/refs/tags/v${httplib_version}.tar.gz
@@ -227,9 +223,6 @@ target_link_libraries(
     OpenSSL::SSL
     Threads::Threads
     zstd::libzstd
-    # macOS-specific libraries required by httplib
-    $<$<PLATFORM_ID:Darwin>:${CFNETWORK_LIB}>
-    $<$<PLATFORM_ID:Darwin>:${CORE_FRAMEWORK_LIB}>
 )
 
 target_compile_definitions(

cmake/SourcesAndHeaders.cmake

@@ -54,9 +54,11 @@ set(sources
   src/detail/dbn_buffer_decoder.cpp
   src/detail/http_client.cpp
   src/detail/json_helpers.cpp
+  src/detail/live_connection.cpp
   src/detail/scoped_fd.cpp
   src/detail/sha256_hasher.cpp
   src/detail/tcp_client.cpp
+  src/detail/tcp_readable.cpp
   src/detail/zstd_stream.cpp
   src/enums.cpp
   src/exceptions.cpp

examples/live/simple.cpp

@@ -25,12 +25,13 @@ int main() {
                     .SetSendTsOut(true)
                     .SetKeyFromEnv()
                     .SetDataset(db::Dataset::GlbxMdp3)
+                    .SetCompression(db::Compression::Zstd)
                     .BuildThreaded();
 
   // Set up signal handler for Ctrl+C
   std::signal(SIGINT, [](int signal) { gSignal = signal; });
 
-  std::vector<std::string> symbols{"ESZ5", "ESZ5 C6200", "ESZ5 P5500"};
+  std::vector<std::string> symbols{"ESZ6", "ESZ6 C8200", "ESZ6 P7500"};
   client.Subscribe(symbols, db::Schema::Definition, db::SType::RawSymbol);
   client.Subscribe(symbols, db::Schema::Mbo, db::SType::RawSymbol);
 

include/databento/detail/buffer.hpp

@@ -38,6 +38,9 @@ class Buffer : public IReadable, public IWritable {
   // Will throw if `length > ReadCapacity()`.
   void ReadExact(std::byte* buffer, std::size_t length) override;
   std::size_t ReadSome(std::byte* buffer, std::size_t max_length) override;
+  // timeout is ignored
+  IReadable::Result ReadSome(std::byte* buffer, std::size_t max_length,
+                             std::chrono::milliseconds timeout) override;
 
   std::byte* ReadBegin() { return read_pos_; }
   std::byte* ReadEnd() { return write_pos_; }

include/databento/detail/live_connection.hpp

@@ -0,0 +1,38 @@
+#pragma once
+
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <string_view>
+
+#include "databento/detail/tcp_client.hpp"
+#include "databento/detail/zstd_stream.hpp"
+#include "databento/enums.hpp"
+#include "databento/ireadable.hpp"
+#include "databento/iwritable.hpp"
+
+namespace databento::detail {
+// Manages the TCP connection to the live gateway with optionally compressed reads for
+// the DBN data.
+class LiveConnection : IWritable {
+ public:
+  LiveConnection(const std::string& gateway, std::uint16_t port);
+
+  void WriteAll(std::string_view str);
+  void WriteAll(const std::byte* buffer, std::size_t size);
+  void ReadExact(std::byte* buffer, std::size_t size);
+  IReadable::Result ReadSome(std::byte* buffer, std::size_t max_size);
+  IReadable::Result ReadSome(std::byte* buffer, std::size_t max_size,
+                             std::chrono::milliseconds timeout);
+  // Closes the socket.
+  void Close();
+  // Sets compression for subsequent reads.
+  void SetCompression(Compression compression);
+
+ private:
+  TcpClient client_;
+  std::optional<ZstdDecodeStream> zstd_stream_;
+};
+}  // namespace databento::detail

include/databento/detail/tcp_client.hpp

@@ -7,21 +7,11 @@
 #include <string_view>
 
 #include "databento/detail/scoped_fd.hpp"  // ScopedFd
+#include "databento/ireadable.hpp"
 
 namespace databento::detail {
 class TcpClient {
  public:
-  enum class Status : std::uint8_t {
-    Ok,
-    Timeout,
-    Closed,
-  };
-
-  struct Result {
-    std::size_t read_size;
-    Status status;
-  };
-
   struct RetryConf {
     std::uint32_t max_attempts{1};
     std::chrono::seconds max_wait{std::chrono::minutes{1}};
@@ -33,11 +23,11 @@ class TcpClient {
   void WriteAll(std::string_view str);
   void WriteAll(const std::byte* buffer, std::size_t size);
   void ReadExact(std::byte* buffer, std::size_t size);
-  Result ReadSome(std::byte* buffer, std::size_t max_size);
+  IReadable::Result ReadSome(std::byte* buffer, std::size_t max_size);
   // Passing a timeout of 0 will block until data is available of the socket is
   // closed, the same behavior as the Read overload without a timeout.
-  Result ReadSome(std::byte* buffer, std::size_t max_size,
-                  std::chrono::milliseconds timeout);
+  IReadable::Result ReadSome(std::byte* buffer, std::size_t max_size,
+                             std::chrono::milliseconds timeout);
   // Closes the socket.
   void Close();
 

include/databento/detail/tcp_readable.hpp

@@ -0,0 +1,24 @@
+#pragma once
+
+#include <chrono>
+#include <cstddef>
+
+#include "databento/detail/tcp_client.hpp"
+#include "databento/ireadable.hpp"
+
+namespace databento::detail {
+// Adapter wrapping TcpClient to implement IReadable interface and be passed
+// as a non-owned pointer to ZstdDecodeStream.
+class TcpReadable : public IReadable {
+ public:
+  explicit TcpReadable(TcpClient* client) : client_{client} {}
+
+  void ReadExact(std::byte* buffer, std::size_t length) override;
+  std::size_t ReadSome(std::byte* buffer, std::size_t max_length) override;
+  IReadable::Result ReadSome(std::byte* buffer, std::size_t max_length,
+                             std::chrono::milliseconds timeout) override;
+
+ private:
+  TcpClient* client_;
+};
+}  // namespace databento::detail

include/databento/detail/zstd_stream.hpp

@@ -2,6 +2,7 @@
 
 #include <zstd.h>
 
+#include <chrono>
 #include <cstddef>  // size_t
 #include <memory>   // unique_ptr
 #include <vector>
@@ -19,9 +20,12 @@ class ZstdDecodeStream : public IReadable {
 
   // Read exactly `length` bytes into `buffer`.
   void ReadExact(std::byte* buffer, std::size_t length) override;
-  // Read at most `length` bytes. Returns the number of bytes read. Will only
+  // Read at most `max_length` bytes. Returns the number of bytes read. Will only
   // return 0 if the end of the stream is reached.
   std::size_t ReadSome(std::byte* buffer, std::size_t max_length) override;
+  // Read at most `max_length` bytes with timeout support.
+  IReadable::Result ReadSome(std::byte* buffer, std::size_t max_length,
+                             std::chrono::milliseconds timeout) override;
 
   IReadable* Input() const { return input_.get(); }
 
@@ -44,6 +48,8 @@ class ZstdCompressStream : public IWritable {
   ~ZstdCompressStream() override;
 
   void WriteAll(const std::byte* buffer, std::size_t length) override;
+  // Flush any buffered data without ending the stream
+  void Flush();
 
  private:
   ILogReceiver* log_receiver_;

include/databento/file_stream.hpp

@@ -17,6 +17,9 @@ class InFileStream : public IReadable {
   // Read at most `length` bytes. Returns the number of bytes read. Will only
   // return 0 if the end of the stream is reached.
   std::size_t ReadSome(std::byte* buffer, std::size_t max_length) override;
+  // timeout is ignored
+  Result ReadSome(std::byte* buffer, std::size_t max_length,
+                  std::chrono::milliseconds timeout) override;
 
  private:
   std::ifstream stream_;