VER: Release 0.9.0

.github/workflows/release.yaml

@@ -32,8 +32,8 @@ jobs:
       - name: Set output
         id: vars
         run: |
-          echo "::set-output name=tag_name::v$(scripts/get_version.sh)"
-          echo "::set-output name=release_name::$(scripts/get_version.sh)"
+          echo "name=tag_name::v$(scripts/get_version.sh)" >> $GITHUB_OUTPUT
+          echo "name=release_name::$(scripts/get_version.sh)" >> $GITHUB_OUTPUT
 
       # Create GitHub release
       - name: Create release

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## 0.9.0 - 2023-06-13
+- Added `Reconnect` methods to `LiveBlocking` and `LiveThreaded`
+- Added optional `exception_callback` argument to `LiveThreaded::Start` to improve
+  error handling options
+- Added batch download support data files (`condition.json` and `symbology.json`)
+- Added support for logging warnings from Historical API
+- Changed `use_ts_out` default to `false`
+- Fixed missing definition for `operator==` for `ImbalanceMsg`
+- Removed 10 minute minimum request time range restriction
+
 ## 0.8.0 - 2023-05-16
 - Changed `end` and `end_date` to optional to support new forward-fill behaviour
 - Renamed `booklevel` MBP field to `levels` for brevity and consistent naming

CMakeLists.txt

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

cmake/SourcesAndHeaders.cmake

@@ -22,6 +22,7 @@ set(headers
   include/databento/with_ts_out.hpp
   include/databento/detail/file_stream.hpp
   include/databento/detail/http_client.hpp
+  include/databento/detail/json_helpers.hpp
   include/databento/detail/scoped_fd.hpp
   include/databento/detail/scoped_thread.hpp
   include/databento/detail/shared_channel.hpp
@@ -49,6 +50,7 @@ set(sources
   src/symbology.cpp
   src/detail/file_stream.cpp
   src/detail/http_client.cpp
+  src/detail/json_helpers.cpp
   src/detail/scoped_fd.cpp
   src/detail/shared_channel.cpp
   src/detail/tcp_client.cpp

cmake/StandardSettings.cmake

@@ -11,17 +11,6 @@ option(${PROJECT_NAME_UPPERCASE}_USE_EXTERNAL_GTEST "Use an external google test
 #
 
 option(${PROJECT_NAME_UPPERCASE}_WARNINGS_AS_ERRORS "Treat compiler warnings as errors." ${IS_MAIN_PROJECT})
-option(${PROJECT_NAME_UPPERCASE}_FORCE_COLOR_OUTPUT "Always produce ANSI-colored output" OFF)
-
-if(${PROJECT_NAME_UPPERCASE}_FORCE_COLOR_OUTPUT)
-  if("${CMAKE_CXX_COMPILER_ID}" STREQUAL "GNU")
-    add_compile_options(-fdiagnostics-color=always)
-  elseif("${CMAKE_CXX_COMPILER_ID}" STREQUAL "Clang")
-    add_compile_options(-fcolor-diagnostics)
-  else()
-    message(WARNING "Couldn't force color output with unsupported compiler: ${CMAKE_CXX_COMPILER_ID}")
-  endif()
-endif()
 
 #
 # Unit testing

include/databento/detail/http_client.hpp

@@ -12,12 +12,14 @@
 #include <string>
 
 namespace databento {
+class ILogReceiver;
 namespace detail {
 class HttpClient {
  public:
-  HttpClient(const std::string& key, const std::string& gateway);
-  HttpClient(const std::string& key, const std::string& gateway,
-             std::uint16_t port);
+  HttpClient(ILogReceiver* log_receiver, const std::string& key,
+             const std::string& gateway);
+  HttpClient(ILogReceiver* log_receiver, const std::string& key,
+             const std::string& gateway, std::uint16_t port);
 
   nlohmann::json GetJson(const std::string& path,
                          const httplib::Params& params);
@@ -27,12 +29,14 @@ class HttpClient {
                     const httplib::ContentReceiver& callback);
 
  private:
-  static nlohmann::json CheckAndParseResponse(const std::string& path,
-                                              httplib::Result&& res);
+  nlohmann::json CheckAndParseResponse(const std::string& path,
+                                       httplib::Result&& res) const;
+  void CheckWarnings(const httplib::Response& response) const;
   static bool IsErrorStatus(int status_code);
 
   static const httplib::Headers kHeaders;
 
+  ILogReceiver* log_receiver_;
   httplib::Client client_;
 };
 }  // namespace detail

include/databento/detail/json_helpers.hpp

@@ -0,0 +1,91 @@
+#pragma once
+
+#include <nlohmann/json.hpp>
+
+#include <map>  // multimap
+#include <string>
+#include <vector>
+
+#include "databento/datetime.hpp"    // UnixNanos
+#include "databento/enums.hpp"       // FromString
+#include "databento/exceptions.hpp"  // JsonResponseError
+
+namespace httplib {
+using Params = std::multimap<std::string, std::string>;
+}
+
+namespace databento {
+namespace detail {
+void SetIfNotEmpty(httplib::Params* params, const std::string& key,
+                   const std::string& value);
+void SetIfNotEmpty(httplib::Params* params, const std::string& key,
+                   const std::vector<databento::JobState>& states);
+
+template <typename T>
+void SetIfPositive(httplib::Params* params, const std::string& key,
+                   const T value) {
+  if (value > 0) {
+    params->emplace(key, std::to_string(value));
+  }
+}
+
+template <>
+inline void SetIfPositive<databento::UnixNanos>(
+    httplib::Params* params, const std::string& key,
+    const databento::UnixNanos value) {
+  if (value.time_since_epoch().count()) {
+    params->emplace(key, databento::ToString(value));
+  }
+}
+
+const nlohmann::json& CheckedAt(const std::string& endpoint,
+                                const nlohmann::json& json,
+                                const std::string& key);
+
+template <typename T>
+T FromCheckedAtString(const std::string& endpoint, const nlohmann::json& json,
+                      const std::string& key) {
+  const auto& val_json = CheckedAt(endpoint, json, key);
+  if (!val_json.is_string()) {
+    throw JsonResponseError::TypeMismatch(endpoint, key + " string", val_json);
+  }
+  return databento::FromString<T>(val_json);
+}
+
+template <typename T>
+T FromCheckedAtStringOrNull(const std::string& endpoint,
+                            const nlohmann::json& json, const std::string& key,
+                            T null_value) {
+  const auto& val_json = CheckedAt(endpoint, json, key);
+  if (val_json.is_null()) {
+    return null_value;
+  }
+  if (val_json.is_string()) {
+    return databento::FromString<T>(val_json);
+  }
+  throw JsonResponseError::TypeMismatch(endpoint, key + " null or string",
+                                        val_json);
+}
+
+template <typename T>
+T ParseAt(const std::string& endpoint, const nlohmann::json& json,
+          const std::string& key);
+template <>
+bool ParseAt(const std::string& endpoint, const nlohmann::json& json,
+             const std::string& key);
+template <>
+std::string ParseAt(const std::string& endpoint, const nlohmann::json& json,
+                    const std::string& key);
+template <>
+std::size_t ParseAt(const std::string& endpoint, const nlohmann::json& json,
+                    const std::string& key);
+template <>
+double ParseAt(const std::string& endpoint, const nlohmann::json& json,
+               const std::string& key);
+template <>
+std::vector<std::string> ParseAt(const std::string& endpoint,
+                                 const nlohmann::json& json,
+                                 const std::string& key);
+
+}  // namespace detail
+}  // namespace databento

include/databento/detail/scoped_thread.hpp

@@ -30,6 +30,7 @@ class ScopedThread {
     }
   }
 
+  std::thread::id Id() const { return thread_.get_id(); }
   bool Joinable() const { return thread_.joinable(); }
   void Join() { return thread_.join(); }
 

include/databento/detail/tcp_client.hpp

@@ -21,7 +21,14 @@ class TcpClient {
     Status status;
   };
 
+  struct RetryConf {
+    std::uint32_t max_attempts{1};
+    std::chrono::seconds max_wait{std::chrono::minutes{1}};
+  };
+
   TcpClient(const std::string& gateway, std::uint16_t port);
+  TcpClient(const std::string& gateway, std::uint16_t port,
+            RetryConf retry_conf);
 
   void WriteAll(const std::string& str);
   void WriteAll(const char* buffer, std::size_t size);
@@ -35,7 +42,8 @@ class TcpClient {
   void Close();
 
  private:
-  static ScopedFd InitSocket(const std::string& gateway, std::uint16_t port);
+  static ScopedFd InitSocket(const std::string& gateway, std::uint16_t port,
+                             RetryConf retry_conf);
 
   ScopedFd socket_;
 };

include/databento/enums.hpp

@@ -51,8 +51,6 @@ enum class Compression : std::uint8_t {
 enum class SType : std::uint8_t {
   InstrumentId = 0,
   RawSymbol = 1,
-  // Deprecated in 0.7.0. Separated into Parent and Continuous.
-  SmartDeprecated = 2,
   Continuous = 3,
   Parent = 4,
 };

include/databento/historical.hpp

@@ -16,12 +16,16 @@
 #include "databento/timeseries.hpp"  // KeepGoing, MetadataCallback, RecordCallback
 
 namespace databento {
+class ILogReceiver;
+
 // A client for interfacing with Databento's historical market data API.
 class Historical {
  public:
-  Historical(std::string key, HistoricalGateway gateway);
+  Historical(ILogReceiver* log_receiver, std::string key,
+             HistoricalGateway gateway);
   // Primarily for unit tests
-  Historical(std::string key, std::string gateway, std::uint16_t port);
+  Historical(ILogReceiver* log_receiver, std::string key, std::string gateway,
+             std::uint16_t port);
 
   /*
    * Getters
@@ -248,6 +252,7 @@ class Historical {
   DbnFileStore TimeseriesGetRangeToFile(const HttplibParams& params,
                                         const std::string& file_path);
 
+  ILogReceiver* log_receiver_;
   const std::string key_;
   const std::string gateway_;
   detail::HttpClient client_;
@@ -265,11 +270,14 @@ class HistoricalBuilder {
   HistoricalBuilder& SetKeyFromEnv();
   HistoricalBuilder& SetKey(std::string key);
   HistoricalBuilder& SetGateway(HistoricalGateway gateway);
+  // Sets the receiver of the logs to be used by the client.
+  HistoricalBuilder& SetLogReceiver(ILogReceiver* log_receiver);
   // Attempts to construct an instance of Historical or throws an exception if
   // no key has been set.
   Historical Build();
 
  private:
+  ILogReceiver* log_receiver_{};
   std::string key_;
   HistoricalGateway gateway_{HistoricalGateway::Bo1};
 };

include/databento/live.hpp

@@ -38,6 +38,6 @@ class LiveBuilder {
   ILogReceiver* log_receiver_{};
   std::string key_;
   std::string dataset_;
-  bool send_ts_out_{true};
+  bool send_ts_out_{false};
 };
 }  // namespace databento

include/databento/live_blocking.hpp

@@ -33,6 +33,7 @@ class LiveBlocking {
   const std::string& Key() const { return key_; }
   const std::string& Dataset() const { return dataset_; }
   const std::string& Gateway() const { return gateway_; }
+  std::uint16_t Port() const { return port_; }
 
   /*
    * Methods
@@ -65,6 +66,8 @@ class LiveBlocking {
   // Stops the session with the gateway. Once stopped, the session cannot be
   // restarted.
   void Stop();
+  // Closes the current connection and attempts to reconnect to the gateway.
+  void Reconnect();
 
  private:
   std::string DetermineGateway() const;
@@ -82,6 +85,7 @@ class LiveBlocking {
   std::string key_;
   std::string dataset_;
   std::string gateway_;
+  std::uint16_t port_;
   bool send_ts_out_;
   detail::TcpClient client_;
   std::array<char, kMaxStrLen> buffer_{};

include/databento/live_threaded.hpp

@@ -21,6 +21,16 @@ class ILogReceiver;
 // is associated with a particular dataset.
 class LiveThreaded {
  public:
+  enum class ExceptionAction {
+    // Start a new session. Return this instead of calling `Start`, which would
+    // cause a deadlock.
+    Restart,
+    // Close the connection and stop the callback thread.
+    Stop,
+  };
+  using ExceptionCallback =
+      std::function<ExceptionAction(const std::exception&)>;
+
   LiveThreaded(ILogReceiver* log_receiver, std::string key, std::string dataset,
                bool send_ts_out);
   LiveThreaded(ILogReceiver* log_receiver, std::string key, std::string dataset,
@@ -36,7 +46,9 @@ class LiveThreaded {
    */
 
   const std::string& Key() const;
+  const std::string& Dataset() const;
   const std::string& Gateway() const;
+  std::uint16_t Port() const;
 
   /*
    * Methods
@@ -60,12 +72,21 @@ class LiveThreaded {
   void Start(RecordCallback record_callback);
   void Start(MetadataCallback metadata_callback,
              RecordCallback record_callback);
+  void Start(MetadataCallback metadata_callback, RecordCallback record_callback,
+             ExceptionCallback exception_callback);
+  // Closes the current connection, and attempts to reconnect to the gateway.
+  void Reconnect();
 
  private:
   struct Impl;
 
   static void ProcessingThread(Impl* impl, MetadataCallback&& metadata_callback,
-                               RecordCallback&& record_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);
 
   // unique_ptr to be movable
   std::unique_ptr<Impl> impl_;

pkg/PKGBUILD

@@ -1,7 +1,7 @@
 # Maintainer: Databento <[email protected]>
 _pkgname=databento-cpp
 pkgname=databento-cpp-git
-pkgver=0.8.0
+pkgver=0.9.0
 pkgrel=1
 pkgdesc="Official C++ client for Databento"
 arch=('any')