icomm_adapter_generic_v1.hpp

Go to the documentation of this file.

#pragma once
 
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <variant>
#include <vector>
 
// =============================================================================
// Protocol-agnostic communication adapter interface  (contract version 3)
//
// Scope: the value-oriented CONTROL plane (read/write/invoke/browse/subscribe)
// over arbitrary protocols (OPC UA, RTMS, Modbus, REST, MQTT, ...). Bulk media
// streaming (e.g. GenICam image acquisition) is intentionally OUT OF SCOPE and
// belongs to a peer interface composed alongside this one, not derived from it.
//
// Design rules:
//   1. A field lives in a core type only if EVERY conceivable adapter can honor
//      it meaningfully. Protocol-specifics go in `properties` / optional fields.
//   2. The interface grows by ADDITION. After the first adapter ships, new
//      methods MUST carry default implementations (return Unsupported), never
//      pure-virtual, so existing implementers keep compiling. Pre-first-adapter
//      breaking changes are free and acceptable (hence the version bumps).
//
// THREAD-SAFETY CONTRACT (binding):
//   - const query methods (Is*, Get*) are callable from any thread at any time.
//   - Read/Write/Invoke/Browse/Discover may be called concurrently with each
//     other and with active subscriptions; implementations serialize as needed.
//   - Configure/Connect/Disconnect must NOT run concurrently with each other or
//     with data operations; the caller owns lifecycle ordering.
//   - Handlers (subscription + connection-state) run on the adapter's internal
//     thread(s); see each handler's contract below.
// =============================================================================
 
namespace eso::comm {
 
// Bumped to 3: builds on v2 with RAII handler registration, item-level
// unsubscribe, and a flat (but map-semantic) PropertyMap. Pre-implementation,
// so the break is costless.
inline constexpr std::uint32_t kContractVersion = 3;
 
using Address = std::string;
using RequestId = std::string;
using SubscriptionId = std::string;
 
// -----------------------------------------------------------------------------
// PropertyMap: contiguous storage (avoids per-node heap allocs / fragmentation
// for the typical <10-entry case) WHILE preserving map semantics: unique keys
// and a lookup API. This keeps the performance win without silently allowing
// duplicate keys the way a raw vector<pair> would.
// -----------------------------------------------------------------------------
class PropertyMap {
public:
  using value_type = std::pair<std::string, std::string>;
  using storage = std::vector<value_type>;
  using iterator = storage::iterator;
  using const_iterator = storage::const_iterator;
 
  PropertyMap() = default;
  PropertyMap(std::initializer_list<value_type> init) {
    for (const auto& kv : init) set(kv.first, kv.second);
  }
 
  // Insert or overwrite. Guarantees key uniqueness.
  void set(std::string key, std::string val) {
    auto it = find(key);
    if (it != items_.end()) it->second = std::move(val);
    else items_.emplace_back(std::move(key), std::move(val));
  }
 
  const std::string* get(std::string_view key) const {
    auto it = find(key);
    return it == items_.end() ? nullptr : &it->second;
  }
 
  std::string get_or(std::string_view key, std::string_view fallback) const {
    auto* v = get(key);
    return v ? *v : std::string(fallback);
  }
 
  bool contains(std::string_view key) const { return find(key) != items_.end(); }
 
  bool erase(std::string_view key) {
    auto it = find(key);
    if (it == items_.end()) return false;
    items_.erase(it);
    return true;
  }
 
  bool empty() const { return items_.empty(); }
  std::size_t size() const { return items_.size(); }
  void reserve(std::size_t n) { items_.reserve(n); }
 
  iterator begin() { return items_.begin(); }
  iterator end() { return items_.end(); }
  const_iterator begin() const { return items_.begin(); }
  const_iterator end() const { return items_.end(); }
 
private:
  const_iterator find(std::string_view key) const {
    return std::find_if(items_.begin(), items_.end(),
                        [&](const value_type& kv) { return kv.first == key; });
  }
  iterator find(std::string_view key) {
    return std::find_if(items_.begin(), items_.end(),
                        [&](const value_type& kv) { return kv.first == key; });
  }
  storage items_;
};
 
// -----------------------------------------------------------------------------
// RAII handle: ties registration lifetime to scope, eliminating the
// handler-outlives-target teardown trap. Move-only; Reset() detaches early.
// -----------------------------------------------------------------------------
class [[nodiscard]] RegistrationToken {
public:
  using UnregisterCallback = std::function<void()>;
 
  RegistrationToken() = default;
  explicit RegistrationToken(UnregisterCallback on_destruct)
    : on_destruct_(std::move(on_destruct)) {}
  ~RegistrationToken() { if (on_destruct_) on_destruct_(); }
 
  RegistrationToken(const RegistrationToken&) = delete;
  RegistrationToken& operator=(const RegistrationToken&) = delete;
  RegistrationToken(RegistrationToken&&) noexcept = default;
  RegistrationToken& operator=(RegistrationToken&&) noexcept = default;
 
  void Reset() { on_destruct_ = nullptr; }
 
private:
  UnregisterCallback on_destruct_;
};
 
// -----------------------------------------------------------------------------
// Cooperative cancellation — the seam off which async + abort hang. Sync
// adapters that cannot interrupt a blocking call MAY ignore it but must say so.
// -----------------------------------------------------------------------------
class CancellationToken {
public:
  CancellationToken() = default;
  explicit CancellationToken(std::shared_ptr<std::atomic_bool> flag)
    : flag_(std::move(flag)) {}
  bool cancelled() const {
    return flag_ && flag_->load(std::memory_order_acquire);
  }
private:
  std::shared_ptr<std::atomic_bool> flag_;
};
 
class CancellationSource {
public:
  CancellationSource() : flag_(std::make_shared<std::atomic_bool>(false)) {}
  CancellationToken token() const { return CancellationToken(flag_); }
  void cancel() { flag_->store(true, std::memory_order_release); }
  bool cancelled() const { return flag_->load(std::memory_order_acquire); }
private:
  std::shared_ptr<std::atomic_bool> flag_;
};
 
enum class Status {
  Ok,
  Timeout,
  NotConnected,
  AlreadyConnected,
  InvalidState,
  BadRequest,
  NotFound,
  AccessDenied,
  TypeMismatch,
  Unsupported,
  Cancelled,
  RemoteError,
  TransportError,
  InternalError
};
 
// -----------------------------------------------------------------------------
// Value: universal payload.
//   - monostate = empty / no-value (write-only nodes, failed reads, defaults).
//   - vector<uint8_t> = raw byte blob / register block.
//   - vector<bool>    = boolean ARRAY (distinct from a blob). Kept despite the
//     proxy-reference quirk so a Boolean[] from e.g. OPC UA has an honest type
//     rather than being conflated with a byte blob. Treat with the usual
//     vector<bool> care (no element addresses); convert at the adapter edge.
// -----------------------------------------------------------------------------
using Value = std::variant<
  std::monostate,
  bool,
  std::int32_t,
  std::uint32_t,
  std::int64_t,
  std::uint64_t,
  float,
  double,
  std::string,
  std::vector<std::uint8_t>,
  std::vector<bool>,
  std::vector<std::int32_t>,
  std::vector<std::uint32_t>,
  std::vector<std::int64_t>,
  std::vector<std::uint64_t>,
  std::vector<float>,
  std::vector<double>,
  std::vector<std::string>,
  PropertyMap>;
 
struct Result {
  // Transport/batch-level status. Per-item structs carry their own status,
  // independent of this, so a batch can succeed overall with some items failed.
  Status status = Status::Ok;
  std::string message;
  explicit operator bool() const { return status == Status::Ok; }
};
 
struct RequestOptions {
  std::chrono::milliseconds timeout{5000};
  std::optional<RequestId> requestId;
  std::optional<CancellationToken> cancellation;
  PropertyMap properties;  // protocol-specific request knobs
};
 
struct DataValue {
  Value value;
  Status status = Status::Ok;
  std::string message;
  // Single universal timestamp; protocols with extra clocks (e.g. UA
  // serverTimestamp) record them in properties under documented keys.
  std::optional<std::chrono::system_clock::time_point> timestamp;
  PropertyMap properties;
};
 
struct ConnectionOptions {
  std::string endpoint;
  PropertyMap properties;  // auth/security/tuning — any scheme fits here
};
 
enum class ConnectionState {
  Disconnected,
  Connecting,
  Connected,
  Disconnecting,
  Faulted
};
 
struct ConnectionInfo {
  std::string endpoint;
  std::string protocol;
  std::string adapterName;
  std::string adapterVersion;
  std::uint32_t contractVersion = kContractVersion;
  PropertyMap properties;
};
 
struct ConnectionStatus : Result {
  ConnectionState state = ConnectionState::Disconnected;
  std::optional<ConnectionInfo> info;
};
 
// Fired on the adapter's internal thread when the link transitions.
// RESUBSCRIPTION CONTRACT: unless an adapter documents otherwise, subscriptions
// DO NOT survive a Faulted transition; on recovery the caller must resubscribe.
// Handler must be non-blocking and must not call Configure/Connect/Disconnect.
struct ConnectionStateChange {
  ConnectionState previous = ConnectionState::Disconnected;
  ConnectionState current = ConnectionState::Disconnected;
  Status reason = Status::Ok;
  std::string message;
};
using ConnectionStateHandler = std::function<void(const ConnectionStateChange&)>;
 
// -----------------------------------------------------------------------------
// Discover: find connectable endpoints (pre-connection). All protocol-specific
// request knobs go through options.properties — there is no second bag.
// -----------------------------------------------------------------------------
struct DiscoverRequest {
  std::optional<std::string> query;
  RequestOptions options;
};
struct DiscoveredEndpoint {
  std::string endpoint;
  std::string protocol;
  std::string name;
  PropertyMap properties;
};
struct DiscoverResult : Result {
  std::vector<DiscoveredEndpoint> endpoints;
};
 
// -----------------------------------------------------------------------------
// Browse: inspect the address space of a CONNECTED endpoint. cursor/hasMore
// page large address spaces: pass the returned cursor back to continue.
// -----------------------------------------------------------------------------
struct BrowseRequest {
  std::optional<Address> root;
  bool recursive = false;
  std::optional<std::string> cursor;
  RequestOptions options;
};
struct NodeDescriptor {
  Address address;
  std::string name;
  std::optional<Address> parent;
  bool hasChildren = false;
  PropertyMap properties;  // "access", "type", "displayName", ...
};
struct BrowseResult : Result {
  std::vector<NodeDescriptor> nodes;
  std::optional<std::string> cursor;
  bool hasMore = false;
};
 
// -----------------------------------------------------------------------------
// Read
// -----------------------------------------------------------------------------
struct ReadItem {
  Address address;
  PropertyMap properties;
};
struct ReadRequest {
  std::vector<ReadItem> items;
  RequestOptions options;
};
struct ReadItemResult : Result {  // item status, independent of batch status
  Address address;
  std::optional<DataValue> data;
};
struct ReadResult : Result {
  std::vector<ReadItemResult> items;
};
 
// -----------------------------------------------------------------------------
// Write — batch items are INDEPENDENT and NON-ATOMIC unless documented;
// inspect per-item results.
// -----------------------------------------------------------------------------
struct WriteItem {
  Address address;
  Value value;
  PropertyMap properties;
};
struct WriteRequest {
  std::vector<WriteItem> items;
  RequestOptions options;
};
struct WriteItemResult : Result {  // item status, independent of batch status
  Address address;
};
struct WriteResult : Result {
  std::vector<WriteItemResult> items;
};
 
// -----------------------------------------------------------------------------
// Invoke (method / command / RPC) — optional capability.
// -----------------------------------------------------------------------------
struct InvokeArgument {
  std::string name;  // may be empty for positional protocols
  Value value;
};
struct InvokeRequest {
  Address method;
  std::vector<InvokeArgument> inputArguments;
  RequestOptions options;
};
struct InvokeArgumentResult {
  std::string name;
  DataValue data;
};
struct InvokeResult : Result {
  std::vector<InvokeArgumentResult> outputArguments;
};
 
// -----------------------------------------------------------------------------
// Subscribe.
// SUBSCRIPTION HANDLER THREADING CONTRACT (binding):
//   - Invoked on the adapter's internal delivery thread(s), not the caller's.
//   - Must be non-blocking and must not throw.
//   - Unsubscribe(thisId) / UnsubscribeItems from within the handler is SAFE;
//     honored after the current invocation returns; no re-entry afterwards.
//   - Connect/Disconnect/Configure from within the handler is NOT safe.
//   - After Unsubscribe / UnsubscribeItems / UnsubscribeAll / Disconnect
//     returns, no further invocations for the affected items occur and any
//     in-flight invocation has completed (these calls block until drained).
//   - Backpressure default: drop oldest; adapters may offer alternatives via
//     request properties.
// -----------------------------------------------------------------------------
struct SubscribeItem {
  Address address;
  PropertyMap properties;
};
struct SubscribeRequest {
  std::vector<SubscribeItem> items;
  std::optional<std::chrono::milliseconds> rateHint;  // best-effort
  RequestOptions options;
};
struct Notification {
  SubscriptionId subscriptionId;
  Address address;
  DataValue data;
};
using SubscriptionHandler = std::function<void(const Notification&)>;
struct SubscribeItemResult : Result {
  Address address;
};
struct SubscribeResult : Result {
  SubscriptionId id;
  std::vector<SubscribeItemResult> items;
};
struct UnsubscribeResult : Result {
  SubscriptionId id;
};
 
// -----------------------------------------------------------------------------
// Capabilities — runtime feature negotiation.
// -----------------------------------------------------------------------------
struct Capabilities {
  bool discover = false;
  bool browse = false;
  bool read = false;
  bool write = false;
  bool invoke = false;
  bool subscribe = false;
  bool batch = false;
  bool async = false;
  bool cancellation = false;
  bool connectionStateEvents = false;
  PropertyMap properties;
};
 
struct AdapterDiagnostics {
  std::string adapterName;
  std::string adapterVersion;
  std::string protocol;
  ConnectionState connectionState = ConnectionState::Disconnected;
 
  std::uint64_t discoverCount = 0;
  std::uint64_t browseCount = 0;
  std::uint64_t readCount = 0;
  std::uint64_t writeCount = 0;
  std::uint64_t invokeCount = 0;
  std::uint64_t notificationCount = 0;
  std::uint64_t errorCount = 0;
 
  std::string lastError;
  PropertyMap properties;
};
 
// Callbacks take the result BY VALUE so consumers can std::move out of it
// across thread boundaries without a deep copy. Delivered on the adapter's
// internal thread (same discipline as subscription handlers).
using DiscoverCallback = std::function<void(DiscoverResult result)>;
using BrowseCallback   = std::function<void(BrowseResult result)>;
using ReadCallback     = std::function<void(ReadResult result)>;
using WriteCallback    = std::function<void(WriteResult result)>;
using InvokeCallback   = std::function<void(InvokeResult result)>;
 
// =============================================================================
// Core interface.
//
// LIFECYCLE CONTRACT:
//   - Configure only while Disconnected (else InvalidState).
//   - Connect requires a prior successful Configure (else InvalidState).
//   - Connect/Disconnect are idempotent w.r.t. target state and block until it
//     is reached or failure is determined.
//   - Data ops require Connected (else NotConnected).
//   - Unimplemented optional ops return Status::Unsupported; consult
//     GetCapabilities() to avoid relying on them.
// =============================================================================
class ICommAdapter {
public:
  virtual ~ICommAdapter() = default;
 
  virtual std::uint32_t GetContractVersion() const { return kContractVersion; }
 
  virtual Capabilities GetCapabilities() const = 0;
  virtual AdapterDiagnostics GetDiagnostics() const = 0;
 
  // --- Lifecycle ---
  virtual Result Configure(const ConnectionOptions& options) = 0;
  virtual Result Connect() = 0;
  virtual Result Disconnect() = 0;
  virtual ConnectionStatus GetConnectionStatus() const = 0;
  virtual bool IsConnected() const = 0;
 
  // RAII registration: unregistration is guaranteed when the returned token
  // leaves scope (or via Reset()). Pass nullptr semantics by simply dropping
  // the token. See ConnectionStateHandler contract.
  virtual RegistrationToken SetConnectionStateHandler(ConnectionStateHandler handler) = 0;
 
  // --- Synchronous operations ---
  virtual DiscoverResult Discover(const DiscoverRequest& request) = 0;
  virtual BrowseResult Browse(const BrowseRequest& request) = 0;
  virtual ReadResult Read(const ReadRequest& request) = 0;
  virtual WriteResult Write(const WriteRequest& request) = 0;
  virtual InvokeResult Invoke(const InvokeRequest& request) = 0;
 
  // --- Asynchronous operations (callback on internal thread; cancel via the
  //     token in request.options.cancellation) ---
  virtual void DiscoverAsync(const DiscoverRequest& request, DiscoverCallback cb) = 0;
  virtual void BrowseAsync(const BrowseRequest& request, BrowseCallback cb) = 0;
  virtual void ReadAsync(const ReadRequest& request, ReadCallback cb) = 0;
  virtual void WriteAsync(const WriteRequest& request, WriteCallback cb) = 0;
  virtual void InvokeAsync(const InvokeRequest& request, InvokeCallback cb) = 0;
 
  // --- Pub/Sub ---
  virtual SubscribeResult Subscribe(const SubscribeRequest& request,
                                    SubscriptionHandler handler) = 0;
  virtual UnsubscribeResult Unsubscribe(const SubscriptionId& id) = 0;
  // Scale a subscription back without destroying the whole batch.
  virtual UnsubscribeResult UnsubscribeItems(const SubscriptionId& id,
                                             const std::vector<Address>& addresses) = 0;
  virtual Result UnsubscribeAll() = 0;
};
 
}  // namespace eso::comm