KinopioHub.CPP

Modern C++20 SDK for KinopioHub, built on top of nats.c.

中文

Current Capabilities

• Supported transport: nats:// and tls://
• Supported patterns: publish / subscribe, request / reply, service mode, scoped variables, latest-value caching, automatic reconnect
• Supported payloads: nlohmann::json, std::string, raw bytes, empty payloads
• Minimum language level: C++20

Quick Start

#include <nlohmann/json.hpp>
#include <kinopio/kinopio.hpp>

int main() {
  kinopio::KinopioOptions options;
  options.servers = {"nats://demo.nats.io:4222"};

  kinopio::KinopioHub hub(options);
  hub.connected(std::chrono::seconds(5));

  auto scope = hub.getScope("chat");
  auto messages = scope.getVariable("messages");

  auto subscription = messages.sub([](const kinopio::Payload& payload, const kinopio::MessageMetadata&) {
    const auto& json = std::get<nlohmann::json>(static_cast<const kinopio::Payload::Base&>(payload));
    std::cout << json.at("message") << std::endl;
  });

  messages.pub(nlohmann::json{{"message", "Hello from C++"}});
}

Public API

ServerSelectionMode

• ordered
• random
• latency

KinopioHub

• KinopioHub(KinopioOptions options = {})
• void connected(std::chrono::milliseconds timeout = 10s) const
• void reconnect() const
• Payload request(std::string_view subject, const Payload& data, const RequestOptions& options = {}) const
• void dispose() const noexcept
• Scope getScope(std::string_view name) const
• StateListenerHandle onStateChange(StateChangeListener listener) const
• KinopioState state() const noexcept
• bool isConnected() const noexcept

Scope

• Variable getVariable(std::string_view name) const

Variable

• const std::string& subject() const
• std::optional<Payload> value() const
• void pub(const Payload& data, const PublishOptions& options = {}) const
• SubscriptionHandle sub(SubscriptionCallback callback, const SubscribeOptions& options = {}) const
• Payload req(const Payload& data, const RequestOptions& options = {}) const
• ServiceHandle serve(ServeHandler handler, const ServeOptions& options = {}) const
• void dispose() const noexcept

KinopioOptions multi-server fields

• servers: candidate NATS server URLs
• serverSelectionMode: optional explicit mode override
• noRandomize: legacy compatibility field; when serverSelectionMode is unset, true maps to ordered and false maps to random

kinopio::leaf

• LeafRuntimeHandle startLeafNode(const LeafRuntimeOptions& options = {})
• std::filesystem::path defaultCacheDirectory()
• LeafRuntimeHandle::snapshot() returns process state, localReady, bridgeReady, resolved binary paths, runtime paths, generated TLS material, and URLs
• LeafRuntimeHandle::stop() stops the local nats-server leaf process and is idempotent

AutoLeafState

• discovering
• followingLeader
• leaderMissingGrace
• electing
• startingLeaf
• leader
• stopped

kinopio::leaf auto coordination

• AutoLeafHandle startAutoLeaf(const AutoLeafOptions& options = {})
• AutoLeafHandle::snapshot() returns the current leader/follower state, stable node identity, elected leader info, manifest path, and optional local leaf snapshot
• AutoLeafHandle::stop() stops background discovery/election work and any owned local leaf runtime

Payload And Codec Model

• Payload is a type-safe variant of: nullptr, nlohmann::json, std::string, std::vector<std::uint8_t>
• Default encoding: JSON -> UTF-8 JSON, string -> UTF-8 text, bytes -> raw bytes, null -> empty payload
• Default decoding: empty payload -> null, UTF-8 JSON -> nlohmann::json, UTF-8 non-JSON -> std::string, invalid UTF-8 -> raw bytes
• Custom codecs replace the default serializer through KinopioOptions::codec

Runtime Behavior

• Construction is non-blocking. Call connected() when you need to wait for readiness.
• When multiple servers are configured and neither serverSelectionMode nor noRandomize is set, the effective mode defaults to latency.
• ordered keeps the configured server order for initial connect and explicit reconnect().
• random shuffles candidates at the start of each connection cycle, then keeps that order stable within the cycle.
• latency probes all configured candidates before initial connect and explicit reconnect(), preferring healthy servers with lower RTT and falling back to input order when all probes fail.
• When latency mode is active with more than one candidate, the hub also runs periodic background probes and may hot-switch to a better server when the measured improvement is at least 30 ms.
• Hot-switching replays value tracking, user subscriptions, and service handlers onto the replacement connection before draining the old one.
• Variable::value() returns the last successfully published or received payload.
• Variable::value(), sub(), serve(), and req() continue to work after a successful hot-switch.
• During the replay window for an external publish, callbacks can briefly be delivered twice because the old and new transport subscriptions may overlap for a moment.
• Repeated publishes with identical serialized bytes are suppressed per variable.
• serve() replies with { "error": true, "message": "..." } when the handler throws.
• req() returns that structured error payload as data; handler failures are not auto-thrown locally.
• Successful hot-switches do not intentionally push the hub back through disconnected.
• Listener, subscription, service, hub, and variable cleanup are all idempotent.

Optional Local Leaf Runtime

• The optional leaf runtime lives behind KINOPIO_BUILD_LEAF_RUNTIME=ON, includes kinopio/leaf.hpp, and links as KinopioHub::kinopiohub_leaf.
• It is intentionally separate from the base KinopioHub::kinopiohub target, so pure SDK consumers do not inherit host-process runtime behavior.
• startLeafNode() resolves nats-server in this order: explicit LeafRuntimeOptions::natsServerBinary, newest cached binary under defaultCacheDirectory(), then automatic download of the latest stable official release when downloads are allowed.
• Downloaded binaries are cached by release version. Reusing the cache does not require another network round-trip.
• The runtime auto-generates a private CA, server certificate, and server key inside the runtime directory for the local leaf listener, and exposes those paths through LeafRuntimeSnapshot::tls.
• localReady means the local nats-server process is up and its monitoring endpoint answers. bridgeReady means at least one upstream leaf remote is currently connected.
• If upstream backbone remotes are unreachable, the local runtime can still start and serve local nats:// clients.
• The current implementation uses host tools for lifecycle steps: curl for release download, tar on macOS/Linux or PowerShell archive expansion on Windows, and openssl for TLS material generation.

Optional Auto Leaf Coordination

• startAutoLeaf() adds a higher-level discovery and election loop on top of the optional leaf runtime.
• It is also guarded by KINOPIO_BUILD_LEAF_RUNTIME=ON and stays inside kinopio::leaf, rather than changing the base SDK runtime model.
• Each node joins a discovery namespace, reuses a stable node identity, watches for an existing leader, waits through a grace window after leader loss, then participates in deterministic election.
• Elections prefer higher backbone quality scores. By default the score is derived from live probes of configured LeafRuntimeOptions::upstreamLeafUrls; ties fall back to a stable node-id order.
• A winning node starts its local leaf runtime and begins broadcasting leader metadata; followers expose that leader metadata through AutoLeafSnapshot and a JSON discovery manifest file.
• Recovery is non-preemptive by default: a node that comes back after failover follows the current leader instead of immediately stealing leadership back, which suppresses leader flapping.
• The first discovery transport is IPv4 UDP broadcast, configured through AutoLeafOptions::discoveryBindHost, discoveryBroadcastAddress, and discoveryPort.
• If you want remote consumers to use the announced leader URL, set a routable AutoLeafOptions::advertiseHost and a compatible leafRuntime.leafListenHost.

Multi-Server Example

kinopio::KinopioOptions options;
options.servers = {
    "nats://edge-a:4222",
    "nats://edge-b:4222",
    "nats://edge-c:4222",
};

// Optional. If omitted, multi-server mode defaults to latency.
options.serverSelectionMode = kinopio::ServerSelectionMode::latency;

Build

Preferred: vcpkg manifest

cmake -S . -B build \
  -DCMAKE_TOOLCHAIN_FILE="$VCPKG_ROOT/scripts/buildsystems/vcpkg.cmake" \
  -DBUILD_TESTING=ON
cmake --build build
ctest --test-dir build --output-on-failure

Manifest dependencies:

• cnats
• nlohmann-json

Package lookup / fallback

• Primary package path: find_package(cnats CONFIG REQUIRED) and find_package(nlohmann_json CONFIG REQUIRED)
• If your packaged cnats export links through OpenSSL::SSL, make sure the normal CMake OpenSSL package is resolvable in that environment as well
• Test-only dependency: Catch2 3 is only needed when BUILD_TESTING=ON
• Development fallback: the top-level CMake project can fetch missing dependencies automatically if they are not preinstalled
• Packaging/install exports are intended for the primary package-manager path; for pure FetchContent development builds, use -DKINOPIO_ENABLE_INSTALL=OFF

Optional leaf runtime target

cmake -S . -B build \
  -DKINOPIO_BUILD_LEAF_RUNTIME=ON \
  -DBUILD_TESTING=ON \
  -DKINOPIO_ENABLE_INSTALL=OFF
cmake --build build
ctest --test-dir build --output-on-failure

• The optional leaf runtime adds the KinopioHub::kinopiohub_leaf target and the kinopio/leaf.hpp header.
• The host that runs startLeafNode() must provide curl, openssl, and either tar or PowerShell archive extraction support.
• startAutoLeaf() additionally expects host UDP broadcast capability on the configured discovery port.

Examples

Example programs live under examples/:

• auto_leaf.cpp
• connection.cpp
• leaf_runtime.cpp
• publish.cpp
• subscribe.cpp
• request_reply.cpp
• scope.cpp

By default the connection-oriented examples use nats://demo.nats.io:4222. Override with KINOPIO_NATS_URL for a single server, or KINOPIO_NATS_URLS for a comma-separated multi-server list.

leaf_runtime.cpp is only built when KINOPIO_BUILD_LEAF_RUNTIME=ON. It starts a local leaf runtime, prints the resolved binary and generated TLS paths, and performs a local publish/subscribe self-check through snapshot.clientUrl. You can optionally provide comma-separated upstream remotes with KINOPIO_LEAF_UPSTREAM_URLS.

auto_leaf.cpp is also only built when KINOPIO_BUILD_LEAF_RUNTIME=ON. It starts the discovery/election loop, prints the observed auto-leaf state, and, when elected leader, runs the same local publish/subscribe self-check through the owned leaf runtime. Useful environment variables are KINOPIO_AUTO_LEAF_NAMESPACE, KINOPIO_AUTO_LEAF_DISCOVERY_PORT, KINOPIO_AUTO_LEAF_ADVERTISE_HOST, and KINOPIO_LEAF_UPSTREAM_URLS.

Tests

• Unit tests cover codec behavior, state changes, scope/variable cache, deduplication, error propagation, and RAII cleanup
• When install exports are enabled with packaged dependencies, CTest also runs an install-consumer smoke that verifies find_package(KinopioHub CONFIG REQUIRED) for the base target and, when enabled, the optional KinopioHub::kinopiohub_leaf target
• Integration tests are gated behind BUILD_INTEGRATION_TESTS=ON
• Integration tests also use KINOPIO_NATS_URL, defaulting to nats://demo.nats.io:4222

Detailed capabilities and limitations: docs/capabilities.md

License

GPL-3.0-or-later. See LICENSE.