ADD: Public client test suite

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.21.1 - TBD
+
+### Enhancements
+- Renamed example directory to examples
+- Renamed test directory to tests
+
 ## 0.21.0 - 2024-07-30
 
 ### Enhancements

CMakeLists.txt

@@ -421,12 +421,12 @@ if(${PROJECT_NAME_UPPERCASE}_ENABLE_UNIT_TESTING)
   unset(CMAKE_CXX_CLANG_TIDY) # disable clang-tidy for tests
   enable_testing()
   message(STATUS "Build unit tests for the project.")
-  add_subdirectory(test)
+  add_subdirectory(tests)
 endif()
 
 if(${PROJECT_NAME_UPPERCASE}_ENABLE_EXAMPLES)
   unset(CMAKE_CXX_CPPCHECK) # disable cppcheck for examples
   unset(CMAKE_CXX_CLANG_TIDY) # disable clang-tidy for examples
   message(STATUS "Build examples for the project.")
-  add_subdirectory(example)
+  add_subdirectory(examples)
 endif()

example/live/CMakeLists.txt → examples/live/CMakeLists.txt

@@ -2,3 +2,4 @@ cmake_minimum_required(VERSION 3.14)
 
 add_example_target(live-readme readme.cpp)
 add_example_target(simple simple.cpp)
+add_example_target(live-smoke-test live_smoke_test.cpp)

examples/live/live_smoke_test.cpp

@@ -0,0 +1,210 @@
+#include <cassert>
+#include <chrono>
+#include <cstdlib>
+#include <cstring>
+#include <databento/constants.hpp>
+#include <databento/datetime.hpp>
+#include <databento/enums.hpp>
+#include <databento/live.hpp>
+#include <databento/symbology.hpp>
+#include <iostream>
+#include <sstream>
+#include <stdexcept>
+#include <string>
+#include <unordered_map>
+
+using namespace databento;
+
+std::vector<std::string> SplitSymbols(const std::string& symbols) {
+  std::vector<std::string> result;
+  std::stringstream ss(symbols);
+
+  std::string buffer;
+  while (std::getline(ss, buffer, ',')) {
+    result.push_back(buffer);
+  }
+
+  return result;
+}
+
+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)) {
+    return std::make_pair(false, UnixNanos{});
+  }
+
+  return std::make_pair(true, UnixNanos{std::chrono::nanoseconds(result)});
+}
+
+void ProcessRecords(LiveBlocking& client, Schema schema,
+                    bool start_from_epoch) {
+  client.Start();
+
+  std::cout << "Starting client...\n";
+
+  // For start != 0 we stop at SymbolMappingMsg so that the tests can be run
+  // outside trading hours
+  auto expected_rtype = Record::RTypeFromSchema(schema);
+  if (!start_from_epoch) {
+    expected_rtype = databento::rtype::SymbolMapping;
+  }
+
+  constexpr auto timeout = std::chrono::seconds{30};
+
+  while (auto record = client.NextRecord(timeout)) {
+    if (record->RType() == expected_rtype) {
+      std::cout << "Received expected record type " << expected_rtype << '\n';
+      break;
+    } else if (auto* msg = record->GetIf<ErrorMsg>()) {
+      std::stringstream ss;
+      ss << "Received error " << msg->Err() << '\n';
+      std::cerr << ss.str();
+      throw std::runtime_error(ss.str());
+    }
+  }
+
+  std::cout << "Finished client\n";
+}
+
+void ProcessSnapshotRecords(LiveBlocking& client, Schema schema) {
+  client.Start();
+
+  std::cout << "Starting client...\n";
+
+  const auto expected_rtype = Record::RTypeFromSchema(schema);
+
+  constexpr auto timeout = std::chrono::seconds{30};
+
+  auto received_snapshot_record = false;
+
+  while (auto record = client.NextRecord(timeout)) {
+    if (auto* mbo_msg = record->GetIf<MboMsg>()) {
+      if (mbo_msg->flags.IsSnapshot()) {
+        received_snapshot_record = true;
+      } else {
+        std::cout << "Received expected record type " << expected_rtype << '\n';
+        break;
+      }
+    } else if (auto* error_msg = record->GetIf<ErrorMsg>()) {
+      std::stringstream ss;
+      ss << "Received error " << error_msg->Err() << '\n';
+      throw std::runtime_error(ss.str());
+    }
+  }
+
+  std::cout << "Finished client\n";
+
+  if (!received_snapshot_record) {
+    throw std::runtime_error("Did not receive snapshot record");
+  }
+}
+
+class ArgParser {
+ public:
+  struct Arg {
+    const std::string name;
+    const std::string arg;
+    const char* value = nullptr;
+  };
+
+  void Add(Arg arg) { args.emplace_back(arg); }
+
+  void Parse(int argc, char* argv[]) {
+    for (auto i = 1; i < argc;) {
+      const auto cur_arg = argv[i];
+      auto it = std::find_if(
+          args.begin(), args.end(),
+          [&cur_arg](const auto& arg) { return cur_arg == arg.arg; });
+      if (it != args.end()) {
+        it->value = argv[i + 1];
+      }
+      i += 2;
+    }
+  }
+
+  const char* Get(const std::string& arg_name) const {
+    auto it = std::find_if(
+        args.begin(), args.end(),
+        [&arg_name](const auto& arg) { return arg_name == arg.name; });
+
+    if (it == args.end()) {
+      return nullptr;
+    }
+
+    return it->value;
+  }
+
+ private:
+  std::vector<Arg> args;
+};
+
+ArgParser ParseArgs(int argc, char* argv[]
+
+) {
+  ArgParser parser;
+  parser.Add(ArgParser::Arg{"gateway", "--gateway"});
+  parser.Add(ArgParser::Arg{"port", "--port", "13000"});
+  parser.Add(ArgParser::Arg{"api_key_env_var", "--api-key-env-var",
+                            "DATABENTO_API_KEY"});
+  parser.Add(ArgParser::Arg{"dataset", "--dataset"});
+  parser.Add(ArgParser::Arg{"schema", "--schema"});
+  parser.Add(ArgParser::Arg{"stype", "--stype"});
+  parser.Add(ArgParser::Arg{"symbols", "--symbols"});
+  parser.Add(ArgParser::Arg{"start", "--start"});
+  parser.Add(ArgParser::Arg{"use_snapshot", "--use-snapshot", "0"});
+
+  parser.Parse(argc, argv);
+
+  return parser;
+}
+
+int main(int argc, char* argv[]) {
+  const auto parser = ParseArgs(argc, argv);
+
+  const auto gateway = parser.Get("gateway");
+  const auto port = std::atoi(parser.Get("port"));
+  const auto api_key_env_var = parser.Get("api_key_env_var");
+  const auto dataset = FromString<Dataset>(parser.Get("dataset"));
+  const auto schema = FromString<Schema>(parser.Get("schema"));
+  const auto stype = FromString<SType>(parser.Get("stype"));
+  const auto symbols = SplitSymbols(parser.Get("symbols"));
+  const auto start = parser.Get("start");
+  const auto use_snapshot = std::atoi(parser.Get("use_snapshot"));
+
+  const auto api_key = std::getenv(api_key_env_var);
+  assert(api_key);
+
+  auto client = LiveBuilder{}
+                    .SetAddress(gateway, static_cast<uint16_t>(port))
+                    .SetKey(std::string{api_key})
+                    .SetDataset(dataset)
+                    .BuildBlocking();
+
+  bool start_from_epoch = false;
+
+  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);
+    } else {
+      client.Subscribe(symbols, schema, stype, start);
+    }
+  } else {
+    client.Subscribe(symbols, schema, stype);
+  }
+
+  if (use_snapshot) {
+    ProcessSnapshotRecords(client, schema);
+  } else {
+    ProcessRecords(client, schema, start_from_epoch);
+  }
+
+  std::cout << "Finished client\n";
+
+  return 0;
+}