auto_leaf.cpp

#include <chrono>
#include <cstdlib>
#include <iostream>
#include <sstream>
#include <string>
#include <thread>
#include <vector>

#include <nlohmann/json.hpp>

#include "kinopio/kinopio.hpp"
#include "kinopio/leaf.hpp"

namespace {

std::vector<std::string> splitValues(const char* value) {
  if (value == nullptr) {
    return {};
  }

  std::vector<std::string> values;
  std::stringstream stream(value);
  std::string item;
  while (std::getline(stream, item, ',')) {
    if (!item.empty()) {
      values.push_back(item);
    }
  }
  return values;
}

const auto& payloadBase(const kinopio::Payload& payload) {
  return static_cast<const kinopio::Payload::Base&>(payload);
}

std::string stateName(kinopio::leaf::AutoLeafState state) {
  switch (state) {
    case kinopio::leaf::AutoLeafState::discovering:
      return "discovering";
    case kinopio::leaf::AutoLeafState::followingLeader:
      return "following-leader";
    case kinopio::leaf::AutoLeafState::leaderMissingGrace:
      return "leader-missing-grace";
    case kinopio::leaf::AutoLeafState::electing:
      return "electing";
    case kinopio::leaf::AutoLeafState::startingLeaf:
      return "starting-leaf";
    case kinopio::leaf::AutoLeafState::leader:
      return "leader";
    case kinopio::leaf::AutoLeafState::stopped:
      return "stopped";
  }
  return "stopped";
}

}  // namespace

int main() {
  kinopio::leaf::AutoLeafOptions options;
  options.discoveryNamespace = std::getenv("KINOPIO_AUTO_LEAF_NAMESPACE") != nullptr
      ? std::getenv("KINOPIO_AUTO_LEAF_NAMESPACE")
      : "demo";
  options.advertiseHost = std::getenv("KINOPIO_AUTO_LEAF_ADVERTISE_HOST") != nullptr
      ? std::getenv("KINOPIO_AUTO_LEAF_ADVERTISE_HOST")
      : "";
  options.leafRuntime.upstreamLeafUrls = splitValues(std::getenv("KINOPIO_LEAF_UPSTREAM_URLS"));

  if (const auto* discoveryPort = std::getenv("KINOPIO_AUTO_LEAF_DISCOVERY_PORT"); discoveryPort != nullptr) {
    options.discoveryPort = static_cast<std::uint16_t>(std::stoi(discoveryPort));
  }

  auto handle = kinopio::leaf::startAutoLeaf(options);

  kinopio::leaf::AutoLeafSnapshot snapshot;
  for (int attempt = 0; attempt < 20; ++attempt) {
    snapshot = handle.snapshot();
    std::cout << "State: " << stateName(snapshot.state) << '\n';
    std::cout << "Node ID: " << snapshot.nodeId << '\n';
    if (snapshot.leaderNodeId.has_value()) {
      std::cout << "Leader ID: " << *snapshot.leaderNodeId << '\n';
    }
    if (snapshot.leaderLeafUrl.has_value()) {
      std::cout << "Leader leaf URL: " << *snapshot.leaderLeafUrl << '\n';
    }
    if (snapshot.localLeaf.has_value()) {
      break;
    }
    std::this_thread::sleep_for(std::chrono::milliseconds{250});
  }

  if (!snapshot.localLeaf.has_value()) {
    handle.stop();
    return snapshot.state == kinopio::leaf::AutoLeafState::followingLeader ? 0 : 1;
  }

  kinopio::KinopioOptions hubOptions;
  hubOptions.servers = {snapshot.localLeaf->clientUrl};

  kinopio::KinopioHub hub(hubOptions);
  hub.connected(std::chrono::seconds{2});

  auto variable = hub.getScope("demo").getVariable("auto-leaf");
  bool received = false;
  auto subscription = variable.sub([&received](const kinopio::Payload& payload, const kinopio::MessageMetadata&) {
    const auto& json = std::get<nlohmann::json>(payloadBase(payload));
    std::cout << "Received payload: " << json.dump() << '\n';
    received = json.at("ok").get<bool>();
  });

  variable.pub(nlohmann::json{{"ok", true}});
  std::this_thread::sleep_for(std::chrono::milliseconds{100});
  handle.stop();
  return received ? 0 : 1;
}