ADD: Add compression support to C++ live clients

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## 0.47.0 - TBD
+
+### 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`
+
+### Breaking changes
+- Added an overload to the `ReadSome` method on `IReadable` for timeout support
+
 ## 0.46.1 - 2026-01-27
 
 ### Enhancements

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_;

include/databento/ireadable.hpp

@@ -1,18 +1,37 @@
 #pragma once
 
+#include <chrono>   // milliseconds
 #include <cstddef>  // byte, size_t
+#include <cstdint>  // uint8_t
 
 namespace databento {
 // An abstract class for readable objects to allow for runtime polymorphism
 // around DBN decoding.
 class IReadable {
  public:
+  enum class Status : std::uint8_t {
+    Ok,       // Data read successfully
+    Timeout,  // Timeout reached before any data available
+    Closed,   // Stream is closed/EOF
+  };
+
+  struct Result {
+    std::size_t read_size;  // Number of bytes read
+    Status status;          // Status of the read operation
+  };
+
   virtual ~IReadable() = default;
 
   // Read exactly `length` bytes into `buffer`.
   virtual void ReadExact(std::byte* buffer, std::size_t length) = 0;
-  // Read at most `length` bytes. Returns the number of bytes read. Will only
-  // return 0 if the end of the stream is reached.
+  // Read at most `length` bytes. Returns the number of bytes read. Will only return 0
+  // if the end of the stream is reached.
   virtual std::size_t ReadSome(std::byte* buffer, std::size_t max_length) = 0;
+  // Read at most `max_length` bytes with timeout support. Returns Result with bytes
+  // read and status. Status will be Timeout if no data available within timeout period,
+  // Closed if stream is closed, or Ok if data was read. A timeout of 0 means wait
+  // indefinitely (same as the no-timeout overload).
+  virtual Result ReadSome(std::byte* buffer, std::size_t max_length,
+                          std::chrono::milliseconds timeout) = 0;
 };
 }  // namespace databento