interop_fixture.cpp

#include <chrono>
#include <condition_variable>
#include <cctype>
#include <cstdlib>
#include <exception>
#include <iostream>
#include <mutex>
#include <optional>
#include <stdexcept>
#include <string>
#include <string_view>
#include <thread>
#include <vector>

#include <nlohmann/json.hpp>

#include "kinopio/kinopio.hpp"

namespace {

using Clock = std::chrono::steady_clock;

class UsageError final : public std::runtime_error {
public:
  using std::runtime_error::runtime_error;
};

std::vector<std::string> splitServers(const std::string& value) {
  std::vector<std::string> servers;
  std::size_t start = 0;

  while (start <= value.size()) {
    const auto comma = value.find(',', start);
    const auto length = comma == std::string::npos ? value.size() - start : comma - start;
    const auto token = value.substr(start, length);
    if (!token.empty()) {
      servers.push_back(token);
    }
    if (comma == std::string::npos) {
      break;
    }
    start = comma + 1;
  }

  return servers;
}

std::vector<std::string> resolveServers() {
  if (const char* value = std::getenv("KINOPIO_NATS_URLS")) {
    const auto servers = splitServers(value);
    if (!servers.empty()) {
      return servers;
    }
  }
  if (const char* value = std::getenv("KINOPIO_NATS_URL")) {
    return {value};
  }
  return {"nats://demo.nats.io:4222"};
}

std::optional<kinopio::ServerSelectionMode> resolveServerSelectionMode() {
  const char* value = std::getenv("KINOPIO_SERVER_SELECTION_MODE");
  if (value == nullptr) {
    return std::nullopt;
  }

  const std::string mode = value;
  if (mode.empty() || mode == "default") {
    return std::nullopt;
  }
  if (mode == "ordered") {
    return kinopio::ServerSelectionMode::ordered;
  }
  if (mode == "random") {
    return kinopio::ServerSelectionMode::random;
  }
  if (mode == "latency") {
    return kinopio::ServerSelectionMode::latency;
  }

  throw UsageError("Unknown KINOPIO_SERVER_SELECTION_MODE: " + mode);
}

std::chrono::milliseconds resolveTimeout() {
  if (const char* value = std::getenv("KINOPIO_INTEROP_TIMEOUT_MS")) {
    return std::chrono::milliseconds{std::stoll(value)};
  }
  return std::chrono::milliseconds{10000};
}

kinopio::KinopioOptions makeOptions() {
  kinopio::KinopioOptions options;
  options.servers = resolveServers();
  options.serverSelectionMode = resolveServerSelectionMode();
  options.timeout = std::chrono::milliseconds{3000};
  options.reconnectTimeout = std::chrono::milliseconds{3000};
  options.waitOnFirstConnect = true;
  options.autoRetry = false;
  return options;
}

nlohmann::json payloadToJson(const kinopio::Payload& payload) {
  return std::visit(
      [](const auto& value) -> nlohmann::json {
        using ValueType = std::decay_t<decltype(value)>;
        if constexpr (std::is_same_v<ValueType, std::nullptr_t>) {
          return nlohmann::json{{"kind", "null"}};
        } else if constexpr (std::is_same_v<ValueType, nlohmann::json>) {
          return nlohmann::json{{"kind", "json"}, {"value", value}};
        } else if constexpr (std::is_same_v<ValueType, std::string>) {
          return nlohmann::json{{"kind", "string"}, {"value", value}};
        } else {
          static constexpr char kHexDigits[] = "0123456789abcdef";
          std::string hex;
          hex.reserve(value.size() * 2);
          for (const auto byte : value) {
            hex.push_back(kHexDigits[(byte >> 4U) & 0x0FU]);
            hex.push_back(kHexDigits[byte & 0x0FU]);
          }
          return nlohmann::json{{"kind", "bytes"}, {"hex", hex}, {"size", value.size()}};
        }
      },
      static_cast<const kinopio::Payload::Base&>(payload));
}

nlohmann::json optionalPayloadToJson(const std::optional<kinopio::Payload>& payload) {
  if (!payload.has_value()) {
    return nullptr;
  }
  return payloadToJson(*payload);
}

kinopio::ByteBuffer parseHex(std::string_view hexText) {
  if (hexText.size() % 2U != 0U) {
    throw UsageError("Hex payload must contain an even number of characters");
  }

  auto hexValue = [](char value) -> std::uint8_t {
    if (value >= '0' && value <= '9') {
      return static_cast<std::uint8_t>(value - '0');
    }
    if (value >= 'a' && value <= 'f') {
      return static_cast<std::uint8_t>(10 + (value - 'a'));
    }
    if (value >= 'A' && value <= 'F') {
      return static_cast<std::uint8_t>(10 + (value - 'A'));
    }
    throw UsageError("Invalid hex payload character");
  };

  kinopio::ByteBuffer bytes;
  bytes.reserve(hexText.size() / 2U);
  for (std::size_t index = 0; index < hexText.size(); index += 2U) {
    const auto high = hexValue(hexText[index]);
    const auto low = hexValue(hexText[index + 1U]);
    bytes.push_back(static_cast<std::uint8_t>((high << 4U) | low));
  }
  return bytes;
}

int parsePayloadArg(int index, int argc, char** argv, kinopio::Payload& payload) {
  if (index >= argc) {
    throw UsageError("Missing payload kind");
  }

  const std::string kind = argv[index++];
  if (kind == "empty") {
    payload = nullptr;
    return index;
  }

  if (index >= argc) {
    throw UsageError("Missing payload value for kind: " + kind);
  }

  const std::string value = argv[index++];
  if (kind == "json") {
    payload = nlohmann::json::parse(value);
    return index;
  }
  if (kind == "string") {
    payload = value;
    return index;
  }
  if (kind == "bytes-hex") {
    payload = parseHex(value);
    return index;
  }

  throw UsageError("Unsupported payload kind: " + kind);
}

void printJson(const nlohmann::json& json) {
  std::cout << json.dump() << std::endl;
}

nlohmann::json makeBaseResult(std::string_view command, const kinopio::KinopioOptions& options) {
  nlohmann::json json = {
      {"command", command},
      {"servers", options.servers},
  };
  if (options.serverSelectionMode.has_value()) {
    switch (*options.serverSelectionMode) {
      case kinopio::ServerSelectionMode::ordered:
        json["serverSelectionMode"] = "ordered";
        break;
      case kinopio::ServerSelectionMode::random:
        json["serverSelectionMode"] = "random";
        break;
      case kinopio::ServerSelectionMode::latency:
        json["serverSelectionMode"] = "latency";
        break;
    }
  } else {
    json["serverSelectionMode"] = nullptr;
  }
  return json;
}

int runConnect(const kinopio::KinopioOptions& options) {
  kinopio::KinopioHub hub(options);
  hub.connected(resolveTimeout());

  auto json = makeBaseResult("connect", options);
  json["ok"] = true;
  json["state"] = hub.state() == kinopio::KinopioState::connected ? "connected" : "other";
  json["isConnected"] = hub.isConnected();
  printJson(json);
  hub.dispose();
  return 0;
}

int runPublish(const kinopio::KinopioOptions& options, int argc, char** argv) {
  if (argc < 5) {
    throw UsageError("Usage: interop_fixture pub <scope> <variable> <payload-kind> [payload-value]");
  }

  const std::string scopeName = argv[2];
  const std::string variableName = argv[3];
  kinopio::Payload payload = nullptr;
  parsePayloadArg(4, argc, argv, payload);

  kinopio::KinopioHub hub(options);
  hub.connected(resolveTimeout());
  auto variable = hub.getScope(scopeName).getVariable(variableName);
  variable.pub(payload);

  auto json = makeBaseResult("pub", options);
  json["ok"] = true;
  json["subject"] = variable.subject();
  json["published"] = payloadToJson(payload);
  json["cachedValue"] = optionalPayloadToJson(variable.value());
  printJson(json);
  hub.dispose();
  return 0;
}

int runSubscribeOnce(const kinopio::KinopioOptions& options, int argc, char** argv) {
  if (argc < 5) {
    throw UsageError("Usage: interop_fixture sub-once <scope> <variable> <timeout-ms>");
  }

  const std::string scopeName = argv[2];
  const std::string variableName = argv[3];
  const auto timeout = std::chrono::milliseconds{std::stoll(argv[4])};

  kinopio::KinopioHub hub(options);
  hub.connected(resolveTimeout());
  auto variable = hub.getScope(scopeName).getVariable(variableName);

  std::mutex mutex;
  std::condition_variable condition;
  bool ready = false;
  kinopio::Payload received = nullptr;
  kinopio::MessageMetadata metadata;

  auto subscription = variable.sub([&](const kinopio::Payload& payload, const kinopio::MessageMetadata& messageMetadata) {
    {
      std::scoped_lock lock(mutex);
      received = payload;
      metadata = messageMetadata;
      ready = true;
    }
    condition.notify_all();
  });

  std::unique_lock lock(mutex);
  if (!condition.wait_for(lock, timeout, [&]() { return ready; })) {
    throw std::runtime_error("Timed out waiting for subscription payload");
  }
  lock.unlock();

  auto json = makeBaseResult("sub-once", options);
  json["ok"] = true;
  json["subject"] = variable.subject();
  json["payload"] = payloadToJson(received);
  json["metadata"] = {
      {"subject", metadata.subject},
      {"reply", metadata.reply ? nlohmann::json(*metadata.reply) : nlohmann::json(nullptr)},
  };
  json["cachedValue"] = optionalPayloadToJson(variable.value());
  printJson(json);
  subscription.unsubscribe();
  hub.dispose();
  return 0;
}

int runValueWait(const kinopio::KinopioOptions& options, int argc, char** argv) {
  if (argc < 5) {
    throw UsageError("Usage: interop_fixture value-wait <scope> <variable> <timeout-ms>");
  }

  const std::string scopeName = argv[2];
  const std::string variableName = argv[3];
  const auto timeout = std::chrono::milliseconds{std::stoll(argv[4])};

  kinopio::KinopioHub hub(options);
  hub.connected(resolveTimeout());
  auto variable = hub.getScope(scopeName).getVariable(variableName);

  std::mutex mutex;
  std::condition_variable condition;
  bool ready = false;
  kinopio::Payload received = nullptr;

  auto subscription = variable.sub([&](const kinopio::Payload& payload, const kinopio::MessageMetadata&) {
    {
      std::scoped_lock lock(mutex);
      received = payload;
      ready = true;
    }
    condition.notify_all();
  });

  std::unique_lock lock(mutex);
  if (!condition.wait_for(lock, timeout, [&]() { return ready; })) {
    throw std::runtime_error("Timed out waiting for value update");
  }
  lock.unlock();

  auto json = makeBaseResult("value-wait", options);
  json["ok"] = true;
  json["subject"] = variable.subject();
  json["received"] = payloadToJson(received);
  json["cachedValue"] = optionalPayloadToJson(variable.value());
  printJson(json);
  subscription.unsubscribe();
  hub.dispose();
  return 0;
}

int runRequest(const kinopio::KinopioOptions& options, int argc, char** argv) {
  if (argc < 5) {
    throw UsageError("Usage: interop_fixture req <scope> <variable> <payload-kind> [payload-value]");
  }

  const std::string scopeName = argv[2];
  const std::string variableName = argv[3];
  kinopio::Payload payload = nullptr;
  parsePayloadArg(4, argc, argv, payload);

  kinopio::KinopioHub hub(options);
  hub.connected(resolveTimeout());
  auto variable = hub.getScope(scopeName).getVariable(variableName);
  const auto response = variable.req(payload);

  auto json = makeBaseResult("req", options);
  json["ok"] = true;
  json["subject"] = variable.subject();
  json["request"] = payloadToJson(payload);
  json["response"] = payloadToJson(response);
  printJson(json);
  hub.dispose();
  return 0;
}

int waitForServiceInvocation(
    const kinopio::KinopioOptions& options,
    const std::string& scopeName,
    const std::string& variableName,
    const kinopio::ServeHandler& handler) {
  kinopio::KinopioHub hub(options);
  hub.connected(resolveTimeout());
  auto variable = hub.getScope(scopeName).getVariable(variableName);

  std::mutex mutex;
  std::condition_variable condition;
  bool ready = false;
  kinopio::Payload requestPayload = nullptr;
  kinopio::MessageMetadata requestMetadata;

  auto service = variable.serve([&](const kinopio::Payload& request, const kinopio::MessageMetadata& metadata) -> kinopio::Payload {
    {
      std::scoped_lock lock(mutex);
      requestPayload = request;
      requestMetadata = metadata;
      ready = true;
    }
    condition.notify_all();
    return handler(request, metadata);
  });

  std::unique_lock lock(mutex);
  if (!condition.wait_for(lock, resolveTimeout(), [&]() { return ready; })) {
    throw std::runtime_error("Timed out waiting for service request");
  }
  lock.unlock();

  // Give the transport a brief window to publish the reply before the
  // one-shot fixture tears the service down.
  std::this_thread::sleep_for(std::chrono::milliseconds{250});

  auto json = makeBaseResult("serve-once", options);
  json["ok"] = true;
  json["subject"] = variable.subject();
  json["request"] = payloadToJson(requestPayload);
  json["metadata"] = {
      {"subject", requestMetadata.subject},
      {"reply", requestMetadata.reply ? nlohmann::json(*requestMetadata.reply) : nlohmann::json(nullptr)},
  };
  printJson(json);
  service.unsubscribe();
  hub.dispose();
  return 0;
}

int runServeOnce(const kinopio::KinopioOptions& options, int argc, char** argv) {
  if (argc < 5) {
    throw UsageError("Usage: interop_fixture serve-once <scope> <variable> <payload-kind> [payload-value]");
  }

  const std::string scopeName = argv[2];
  const std::string variableName = argv[3];
  kinopio::Payload responsePayload = nullptr;
  parsePayloadArg(4, argc, argv, responsePayload);

  return waitForServiceInvocation(
      options,
      scopeName,
      variableName,
      [responsePayload](const kinopio::Payload&, const kinopio::MessageMetadata&) -> kinopio::Payload {
        return responsePayload;
      });
}

int runServeErrorOnce(const kinopio::KinopioOptions& options, int argc, char** argv) {
  if (argc < 5) {
    throw UsageError("Usage: interop_fixture serve-error-once <scope> <variable> <message>");
  }

  const std::string scopeName = argv[2];
  const std::string variableName = argv[3];
  const std::string message = argv[4];

  return waitForServiceInvocation(
      options,
      scopeName,
      variableName,
      [message](const kinopio::Payload&, const kinopio::MessageMetadata&) -> kinopio::Payload {
        throw std::runtime_error(message);
      });
}

}  // namespace

int main(int argc, char** argv) {
  try {
    if (argc < 2) {
      throw UsageError("Usage: interop_fixture <connect|pub|sub-once|value-wait|req|serve-once|serve-error-once> ...");
    }

    const kinopio::KinopioOptions options = makeOptions();
    const std::string command = argv[1];

    if (command == "connect") {
      return runConnect(options);
    }
    if (command == "pub") {
      return runPublish(options, argc, argv);
    }
    if (command == "sub-once") {
      return runSubscribeOnce(options, argc, argv);
    }
    if (command == "value-wait") {
      return runValueWait(options, argc, argv);
    }
    if (command == "req") {
      return runRequest(options, argc, argv);
    }
    if (command == "serve-once") {
      return runServeOnce(options, argc, argv);
    }
    if (command == "serve-error-once") {
      return runServeErrorOnce(options, argc, argv);
    }

    throw UsageError("Unknown command: " + command);
  } catch (const std::exception& error) {
    printJson({
        {"ok", false},
        {"error", error.what()},
    });
    return 1;
  }
}