ADD: Add configurable buffer size for live clients

Author: threecgreen

CHANGELOG.md

@@ -3,9 +3,14 @@
 ## 0.38.0 - TBD
 
 ### Enhancements
+- Made the buffer size used by the live clients when reading from the TCP socket
+  configurable through the `LiveBuilder::SetBufferSize()` method
 - Added log level prefix to `ConsoleLogReceiver` output
 
 ### Breaking changes
+- Live client instances can only be created through the `LiveBuilder` class
+- Changed `HeartbeatInterval()` getters on `LiveBlocking` and `LiveThreaded` to return
+  an `std::optional`
 - Added new optional `ShouldLog` virtual method to `ILogReceiver` to
   filter the levels of log messages that will be sent to the receiver
 

include/databento/detail/buffer.hpp

@@ -11,7 +11,9 @@
 namespace databento::detail {
 class Buffer : public IReadable, public IWritable {
  public:
-  Buffer() : Buffer(64 * std::size_t{1 << 10}) {}
+  static constexpr std::size_t kDefaultBufSize = 64 * std::size_t{1 << 10};
+
+  Buffer() : Buffer(kDefaultBufSize) {}
   explicit Buffer(std::size_t init_capacity)
       : buf_{AlignedNew(init_capacity), AlignedDelete},
         end_{buf_.get() + init_capacity},

include/databento/live.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <chrono>
+#include <cstddef>
 #include <string>
 
 #include "databento/enums.hpp"  // VersionUpgradePolicy
@@ -9,22 +10,32 @@
 #include "databento/publishers.hpp"
 
 namespace databento {
+// Forward declarations
 class ILogReceiver;
 
 // A helper class for constructing a Live client, either an instance of
 // LiveBlocking or LiveThreaded.
 class LiveBuilder {
  public:
-  LiveBuilder() = default;
+  LiveBuilder();
+
+  /*
+   * Required settters
+   */
 
   // Sets `key_` based on the environment variable DATABENTO_API_KEY.
   //
   // NOTE: This is not thread-safe if `std::setenv` is used elsewhere in the
   // program.
   LiveBuilder& SetKeyFromEnv();
   LiveBuilder& SetKey(std::string key);
-  LiveBuilder& SetDataset(std::string dataset);
   LiveBuilder& SetDataset(Dataset dataset);
+  LiveBuilder& SetDataset(std::string dataset);
+
+  /*
+   * Optional settters
+   */
+
   // Whether to append the gateway send timestamp after each DBN message.
   LiveBuilder& SetSendTsOut(bool send_ts_out);
   // Set the version upgrade policy for when receiving DBN data from a prior
@@ -36,6 +47,13 @@ class LiveBuilder {
   LiveBuilder& SetHeartbeatInterval(std::chrono::seconds heartbeat_interval);
   // Overrides the gateway and port. This is an advanced method.
   LiveBuilder& SetAddress(std::string gateway, std::uint16_t port);
+  // Overrides the size of the buffer used for reading data from the TCP socket.
+  LiveBuilder& SetBufferSize(std::size_t size);
+
+  /*
+   * Build a live client instance
+   */
+
   // Attempts to construct an instance of a blocking live client or throws an
   // exception.
   LiveBlocking BuildBlocking();
@@ -51,8 +69,10 @@ class LiveBuilder {
   std::uint16_t port_{};
   std::string key_;
   std::string dataset_;
+
   bool send_ts_out_{false};
   VersionUpgradePolicy upgrade_policy_{VersionUpgradePolicy::UpgradeToV3};
-  std::chrono::seconds heartbeat_interval_{};
+  std::optional<std::chrono::seconds> heartbeat_interval_{};
+  std::size_t buffer_size_;
 };
 }  // namespace databento

include/databento/live_blocking.hpp

@@ -4,6 +4,7 @@
 #include <chrono>  // milliseconds
 #include <cstddef>
 #include <cstdint>
+#include <optional>
 #include <string>
 #include <string_view>
 #include <utility>  // pair
@@ -18,21 +19,17 @@
 #include "databento/record.hpp"  // Record, RecordHeader
 
 namespace databento {
+// Forward declaration
 class ILogReceiver;
+class LiveBuilder;
+class LiveThreaded;
 
 // A client for interfacing with Databento's real-time and intraday replay
 // market data API. This client provides a blocking API for getting the next
 // record. Unlike Historical, each instance of LiveBlocking is associated with a
 // particular dataset.
 class LiveBlocking {
  public:
-  LiveBlocking(ILogReceiver* log_receiver, std::string key, std::string dataset,
-               bool send_ts_out, VersionUpgradePolicy upgrade_policy,
-               std::chrono::seconds heartbeat_interval);
-  LiveBlocking(ILogReceiver* log_receiver, std::string key, std::string dataset,
-               std::string gateway, std::uint16_t port, bool send_ts_out,
-               VersionUpgradePolicy upgrade_policy,
-               std::chrono::seconds heartbeat_interval);
   /*
    * Getters
    */
@@ -45,8 +42,8 @@ class LiveBlocking {
   VersionUpgradePolicy UpgradePolicy() const { return upgrade_policy_; }
   // The the first member of the pair will be true, when the heartbeat interval
   // was overridden.
-  std::pair<bool, std::chrono::seconds> HeartbeatInterval() const {
-    return {heartbeat_interval_.count() > 0, heartbeat_interval_};
+  std::optional<std::chrono::seconds> HeartbeatInterval() const {
+    return heartbeat_interval_;
   }
   const std::vector<LiveSubscription>& Subscriptions() const {
     return subscriptions_;
@@ -93,6 +90,19 @@ class LiveBlocking {
   void Resubscribe();
 
  private:
+  friend LiveBuilder;
+  friend LiveThreaded;
+
+  LiveBlocking(ILogReceiver* log_receiver, std::string key, std::string dataset,
+               bool send_ts_out, VersionUpgradePolicy upgrade_policy,
+               std::optional<std::chrono::seconds> heartbeat_interval,
+               std::size_t buffer_size);
+  LiveBlocking(ILogReceiver* log_receiver, std::string key, std::string dataset,
+               std::string gateway, std::uint16_t port, bool send_ts_out,
+               VersionUpgradePolicy upgrade_policy,
+               std::optional<std::chrono::seconds> heartbeat_interval,
+               std::size_t buffer_size);
+
   std::string DetermineGateway() const;
   std::uint64_t Authenticate();
   std::string DecodeChallenge();
@@ -115,11 +125,11 @@ class LiveBlocking {
   bool send_ts_out_;
   std::uint8_t version_{};
   VersionUpgradePolicy upgrade_policy_;
-  std::chrono::seconds heartbeat_interval_;
+  std::optional<std::chrono::seconds> heartbeat_interval_;
   detail::TcpClient client_;
   std::uint32_t sub_counter_{};
   std::vector<LiveSubscription> subscriptions_;
-  detail::Buffer buffer_{};
+  detail::Buffer buffer_;
   // Must be 8-byte aligned for records
   alignas(RecordHeader) std::array<std::byte, kMaxRecordLen> compat_buffer_{};
   std::uint64_t session_id_;

include/databento/live_threaded.hpp

@@ -1,9 +1,12 @@
 #pragma once
 
 #include <chrono>
+#include <cstdint>
 #include <functional>  // function
 #include <memory>      // unique_ptr
+#include <optional>
 #include <string>
+#include <string_view>
 #include <utility>  // pair
 #include <vector>
 
@@ -14,15 +17,17 @@
 #include "databento/timeseries.hpp"  // MetadataCallback, RecordCallback
 
 namespace databento {
+// Forward declaration
 class ILogReceiver;
+class LiveBuilder;
 
 // A client for interfacing with Databento's real-time and intraday replay
 // market data API. This client provides a threaded event-driven API for
 // receiving the next record. Unlike Historical, each instance of LiveThreaded
 // is associated with a particular dataset.
 class LiveThreaded {
  public:
-  enum class ExceptionAction {
+  enum class ExceptionAction : std::uint8_t {
     // Start a new session. Return this instead of calling `Start`, which would
     // cause a deadlock.
     Restart,
@@ -32,13 +37,6 @@ class LiveThreaded {
   using ExceptionCallback =
       std::function<ExceptionAction(const std::exception&)>;
 
-  LiveThreaded(ILogReceiver* log_receiver, std::string key, std::string dataset,
-               bool send_ts_out, VersionUpgradePolicy upgrade_policy,
-               std::chrono::seconds heartbeat_interval);
-  LiveThreaded(ILogReceiver* log_receiver, std::string key, std::string dataset,
-               std::string gateway, std::uint16_t port, bool send_ts_out,
-               VersionUpgradePolicy upgrade_policy,
-               std::chrono::seconds heartbeat_interval);
   LiveThreaded(const LiveThreaded&) = delete;
   LiveThreaded& operator=(const LiveThreaded&) = delete;
   LiveThreaded(LiveThreaded&& other) noexcept;
@@ -57,7 +55,7 @@ class LiveThreaded {
   VersionUpgradePolicy UpgradePolicy() const;
   // The the first member of the pair will be true, when the heartbeat interval
   // was overridden.
-  std::pair<bool, std::chrono::seconds> HeartbeatInterval() const;
+  std::optional<std::chrono::seconds> HeartbeatInterval() const;
   const std::vector<LiveSubscription>& Subscriptions() const;
   std::vector<LiveSubscription>& Subscriptions();
 
@@ -96,15 +94,27 @@ class LiveThreaded {
   KeepGoing BlockForStop(std::chrono::milliseconds timeout);
 
  private:
+  friend LiveBuilder;
+
   struct Impl;
 
   static void ProcessingThread(Impl* impl, MetadataCallback&& metadata_callback,
                                RecordCallback&& record_callback,
                                ExceptionCallback&& exception_callback);
   static ExceptionAction ExceptionHandler(
       Impl* impl, const ExceptionCallback& exception_callback,
-      const std::exception& exc, const char* pretty_function_name,
-      const char* message);
+      const std::exception& exc, std::string_view pretty_function_name,
+      std::string_view message);
+
+  LiveThreaded(ILogReceiver* log_receiver, std::string key, std::string dataset,
+               bool send_ts_out, VersionUpgradePolicy upgrade_policy,
+               std::optional<std::chrono::seconds> heartbeat_interval,
+               std::size_t buffer_size);
+  LiveThreaded(ILogReceiver* log_receiver, std::string key, std::string dataset,
+               std::string gateway, std::uint16_t port, bool send_ts_out,
+               VersionUpgradePolicy upgrade_policy,
+               std::optional<std::chrono::seconds> heartbeat_interval,
+               std::size_t buffer_size);
 
   // unique_ptr to be movable
   std::unique_ptr<Impl> impl_;

live.hpp

@@ -0,0 +1,78 @@
+#pragma once
+
+#include <chrono>
+#include <cstddef>
+#include <string>
+
+#include "databento/enums.hpp"  // VersionUpgradePolicy
+#include "databento/live_blocking.hpp"
+#include "databento/live_threaded.hpp"
+#include "databento/publishers.hpp"
+
+namespace databento {
+// Forward declarations
+class ILogReceiver;
+
+// A helper class for constructing a Live client, either an instance of
+// LiveBlocking or LiveThreaded.
+class LiveBuilder {
+ public:
+  LiveBuilder();
+
+  /*
+   * Required settters
+   */
+
+  // Sets `key_` based on the environment variable DATABENTO_API_KEY.
+  //
+  // NOTE: This is not thread-safe if `std::setenv` is used elsewhere in the
+  // program.
+  LiveBuilder& SetKeyFromEnv();
+  LiveBuilder& SetKey(std::string key);
+  LiveBuilder& SetDataset(Dataset dataset);
+  LiveBuilder& SetDataset(std::string dataset);
+
+  /*
+   * Optional settters
+   */
+
+  // Whether to append the gateway send timestamp after each DBN message.
+  LiveBuilder& SetSendTsOut(bool send_ts_out);
+  // Set the version upgrade policy for when receiving DBN data from a prior
+  // version. Defaults to upgrading to DBNv2 (if not already).
+  LiveBuilder& SetUpgradePolicy(VersionUpgradePolicy upgrade_policy);
+  // Sets the receiver of the logs to be used by the client.
+  LiveBuilder& SetLogReceiver(ILogReceiver* log_receiver);
+  // Overrides the heartbeat interval.
+  LiveBuilder& SetHeartbeatInterval(std::chrono::seconds heartbeat_interval);
+  // Overrides the gateway and port. This is an advanced method.
+  LiveBuilder& SetAddress(std::string gateway, std::uint16_t port);
+  // Overrides the size of the buffer used for reading data from the TCP socket.
+  LiveBuilder& SetBufferSize(std::size_t size);
+
+  /*
+   * Build a live client instance
+   */
+
+  // Attempts to construct an instance of a blocking live client or throws an
+  // exception.
+  LiveBlocking BuildBlocking();
+  // Attempts to construct an instance of a threaded live client or throws an
+  // exception.
+  LiveThreaded BuildThreaded();
+
+ private:
+  void Validate();
+
+  ILogReceiver* log_receiver_{};
+  std::string gateway_{};
+  std::uint16_t port_{};
+  std::string key_;
+  std::string dataset_;
+
+  bool send_ts_out_{false};
+  VersionUpgradePolicy upgrade_policy_{VersionUpgradePolicy::UpgradeToV3};
+  std::optional<std::chrono::seconds> heartbeat_interval_{};
+  std::size_t buffer_size_;
+};
+}  // namespace databento