feat(sdk): lazy per-topic subscription for DataSources (0.15.0)

CHANGELOG.md

@@ -3,7 +3,52 @@
 All notable changes to `plotjuggler_sdk` are recorded here. Versioning policy is in
 [`CLAUDE.md`](./CLAUDE.md) → "Release Versioning".
 
-## [Unreleased] — on branch `feature/plot-markers`, not yet publicly tagged
+## [Unreleased]
+
+### 0.15.0 — Lazy per-topic subscription for DataSources (MINOR, additive only)
+
+Streaming sources can now connect without subscribing, advertise their full topic
+set to the host, and subscribe/unsubscribe individual topics on host demand (driven
+by which topics actually have consumers — plotted curves, scene layers, transform
+inputs). Three additive pieces; every existing plugin keeps working with no
+recompile (`abidiff` additions only):
+
+- **New capability bit** `PJ_DATA_SOURCE_CAPABILITY_LAZY_SUBSCRIPTION` (1<<6, C++
+  mirror `kCapabilityLazySubscription`) declaring that a source supports the flow.
+- **New runtime-host tail slot** `set_advertised_topics` on
+  `PJ_data_source_runtime_host_vtable_t`: declarative full-set topic advertise with
+  parser metadata (reuses `PJ_parser_binding_request_t`), so advertised topics can
+  appear in the host catalog with zero data. C++ wrapper
+  `DataSourceRuntimeHostView::setAdvertisedTopics` returns the distinct error
+  "host does not support lazy subscription" on old hosts so plugins can fall back
+  to eager mode.
+- **New plugin extension** `"pj.topic_subscription.v1"`
+  (`pj_base/data_source_topic_subscription.h`): `set_active_topics` — the host's
+  declarative desired-active set; the plugin reconciles subscriptions. Invoked on
+  the poll thread, strictly serialized with poll/start/stop.
+  `DataSourcePluginBase` wires it automatically to the new
+  `onActiveTopicsChanged()` virtual when the capability is declared;
+  `DataSourceHandle::topicSubscriptionExtension()` / `setActiveTopics()` on the
+  host side.
+- **New parser tail slot** `describe_schema_columns` on
+  `PJ_message_parser_vtable_t`: the flattened scalar-column manifest for a bound
+  schema (JSON `[{"path","type"}]`, parse()-emission order), letting hosts
+  pre-create catalog columns for topics with no samples yet. C++ helpers:
+  `MessageParserPluginBase::describeSchemaColumns` returning
+  `std::vector<sdk::ColumnSpec>` (base serializes the wire JSON),
+  `MessageParserHandle::describeSchemaColumns` host-side.
+
+**UNRELEASED BREAK (in-process C++ parser path).** Adding the
+`describeSchemaColumns` virtual (and `columns_json_buf_` member) to
+`MessageParserPluginBase` changes the class's C++ layout/vtable. The host calls
+`classifySchema`/`parseScalars`/`parseObject` directly on the C++ pointer, so
+parser plugins built against ≤0.14.0 MUST be rebuilt against 0.15.0 — loading a
+stale parser `.so` into a 0.15 host segfaults in `parseObject`. Pure C-vtable
+paths are unaffected. As with the 0.13.0 unification: no public tag ever
+shipped 0.14/0.15, so no released plugin breaks; the first PUBLIC release of
+this line must still be tagged 1.0.0.
+
+## [0.14.0 and earlier] — previously on branch `feature/plot-markers`, not yet publicly tagged
 
 ### Host service: markers + transforms unified into `pj.data_processors.v1` (UNRELEASED BREAK)
 

CMakeLists.txt

@@ -126,7 +126,7 @@ endif()
 if(PJ_INSTALL_SDK)
   include(CMakePackageConfigHelpers)
 
-  set(PJ_PACKAGE_VERSION "0.14.0")
+  set(PJ_PACKAGE_VERSION "0.15.0")
   set(PJ_PACKAGE_CMAKE_DIR ${CMAKE_INSTALL_LIBDIR}/cmake/plotjuggler_sdk)
 
   install(EXPORT plotjuggler_sdkTargets

conanfile.py

@@ -6,7 +6,7 @@
   plugin_sdk   — umbrella for plugin authors (base + dialog SDK + parser SDK)
   plugin_host  — umbrella for host loaders (data_source/parser/toolbox/dialog)
 
-A consuming Conan recipe declares e.g. `plotjuggler_sdk/0.14.0` and then:
+A consuming Conan recipe declares e.g. `plotjuggler_sdk/0.15.0` and then:
 
     find_package(plotjuggler_sdk REQUIRED COMPONENTS plugin_sdk)
     target_link_libraries(my_plugin PRIVATE plotjuggler_sdk::plugin_sdk)
@@ -35,10 +35,13 @@ class PlotjugglerSdkConan(ConanFile):
     # `pj.markers.v1` and the interim `pj.generators.v1`; generalized
     # `create_data_processor`/`validate_data_processor_script` with kind/language/flags).
     # 0.14.0 then adds the dialog-protocol additions (radio column + interactive
-    # sub-panel). All pre-1.0 unreleased — no PUBLIC tag ever shipped
+    # sub-panel). 0.15.0 adds lazy per-topic subscription for DataSources (the
+    # `pj.topic_subscription.v1` extension, the runtime-host `set_advertised_topics`
+    # tail slot, and the parser `describe_schema_columns` tail slot) — additive only.
+    # All pre-1.0 unreleased — no PUBLIC tag ever shipped
     # `pj.data_processors.v1`, so no released plugin breaks. The FIRST public release
     # that carries the unified `pj.data_processors.v1` MUST be tagged 1.0.0. See CHANGELOG.md.
-    version = "0.14.0"
+    version = "0.15.0"
     # Apache-2.0 covers the whole SDK (pj_base + pj_plugins). See LICENSE.
     license = "Apache-2.0"
     url = "https://github.com/PlotJuggler/plotjuggler_sdk"

pj_base/CMakeLists.txt

@@ -76,6 +76,7 @@ if(PJ_BUILD_TESTS)
     tests/data_processors_api_test.cpp
     tests/settings_store_host_test.cpp
     tests/data_source_protocol_test.cpp
+    tests/topic_subscription_extension_test.cpp
     # TODO: data_source_plugin_base_test.cpp and
     # message_parser_plugin_base_test.cpp exercised old bind_write_host /
     # bind_runtime_host slots that were already removed by the time the

pj_base/include/pj_base/data_source_protocol.h

@@ -131,6 +131,14 @@ enum {
   PJ_DATA_SOURCE_CAPABILITY_DELEGATED_INGEST = 1ull << 3,  /**< Plugin pushes raw bytes for host-side parsing. */
   PJ_DATA_SOURCE_CAPABILITY_SUPPORTS_PAUSE = 1ull << 4,    /**< pause()/resume() are implemented. */
   PJ_DATA_SOURCE_CAPABILITY_HAS_DIALOG = 1ull << 5,        /**< Plugin provides a configuration dialog. */
+  /**
+   * Source supports lazy per-topic subscription: it advertises its full topic
+   * set via the runtime host's `set_advertised_topics` tail slot without
+   * subscribing, and subscribes/unsubscribes individual topics on host demand
+   * through the "pj.topic_subscription.v1" plugin extension (see
+   * pj_base/data_source_topic_subscription.h).
+   */
+  PJ_DATA_SOURCE_CAPABILITY_LAZY_SUBSCRIPTION = 1ull << 6,
 };
 
 /** Opaque handle returned by ensure_parser_binding, used with push_message. */
@@ -327,6 +335,35 @@ typedef struct PJ_data_source_runtime_host_vtable_t {
   bool (*push_message)(
       void* ctx, PJ_parser_binding_handle_t handle, int64_t host_timestamp_ns,
       PJ_message_data_fetcher_t fetch_message_data, PJ_error_t* out_error) PJ_NOEXCEPT;
+
+  /**
+   * [stream-thread] Declarative topic advertise for lazy-subscription sources.
+   *
+   * The plugin reports the FULL set of topics currently available from the
+   * source, each carrying the same parser metadata as ensure_parser_binding
+   * (topic name, encoding, type name, schema, parser config). The host
+   * reconciles against the previously advertised set:
+   *
+   *   - new entries are pre-bound (parser instantiated, schema registered)
+   *     so the topics appear in the host catalog with zero data;
+   *   - entries that disappeared are marked unavailable — already-ingested
+   *     data and catalog entries persist, and the host stops requesting the
+   *     topic through the "pj.topic_subscription.v1" extension.
+   *
+   * Idempotent: call again with the full set whenever the source's topic
+   * list changes (mid-stream advertise/unadvertise, reconnect). Advertising
+   * a topic does NOT subscribe it; production starts only after the host
+   * requests the topic via the subscription extension (or the plugin's own
+   * always-on selection). All views are valid only for the duration of the
+   * call.
+   *
+   * Only meaningful for plugins declaring
+   * PJ_DATA_SOURCE_CAPABILITY_LAZY_SUBSCRIPTION. Gate reads with
+   * PJ_HAS_TAIL_SLOT; a host without this slot does not support lazy
+   * subscription and the plugin should fall back to its eager behavior.
+   */
+  bool (*set_advertised_topics)(
+      void* ctx, const PJ_parser_binding_request_t* topics, uint64_t count, PJ_error_t* out_error) PJ_NOEXCEPT;
 } PJ_data_source_runtime_host_vtable_t;
 
 /** Fat pointer pairing a runtime host context with its vtable. */

pj_base/include/pj_base/data_source_topic_subscription.h

@@ -0,0 +1,72 @@
+/**
+ * @file data_source_topic_subscription.h
+ * @brief "pj.topic_subscription.v1" plugin extension — host-driven lazy
+ *        per-topic subscription for DataSource plugins.
+ *
+ * A DataSource plugin that declares PJ_DATA_SOURCE_CAPABILITY_LAZY_SUBSCRIPTION
+ * connects without subscribing to any topic. It advertises the available topic
+ * set through the runtime host's `set_advertised_topics` tail slot (see
+ * data_source_protocol.h) and lets the host drive which topics are actually
+ * produced through this extension:
+ *
+ *   host: get_plugin_extension(ctx, "pj.topic_subscription.v1")
+ *      -> PJ_topic_subscription_extension_t*
+ *   host: ext->set_active_topics(ctx, names, count, &err)   // desired set
+ *
+ * The verb is DECLARATIVE: the host always sends the full desired-active set
+ * and the plugin reconciles — subscribing topics newly present, unsubscribing
+ * topics absent. Declarative reconcile stays correct across reconnects and
+ * mid-stream advertise/unadvertise without any imperative pairing.
+ */
+// Copyright 2026 Davide Faconti
+// SPDX-License-Identifier: Apache-2.0
+
+#ifndef PJ_DATA_SOURCE_TOPIC_SUBSCRIPTION_H
+#define PJ_DATA_SOURCE_TOPIC_SUBSCRIPTION_H
+
+#include <stdbool.h>
+#include <stdint.h>
+
+#include "pj_base/plugin_data_api.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/** Extension id passed to PJ_data_source_vtable_t::get_plugin_extension. */
+#define PJ_TOPIC_SUBSCRIPTION_EXTENSION_ID "pj.topic_subscription.v1"
+
+/**
+ * Returned by get_plugin_extension(ctx, PJ_TOPIC_SUBSCRIPTION_EXTENSION_ID).
+ * Static plugin-owned POD, valid for the plugin-instance lifetime.
+ */
+typedef struct PJ_topic_subscription_extension_t {
+  uint32_t struct_size; /**< sizeof(PJ_topic_subscription_extension_t) — future tail slots. */
+
+  /**
+   * [stream-thread] Set the host's desired-active topic set (full set, by
+   * topic name as advertised). The plugin reconciles its actual
+   * subscriptions: subscribe topics newly present, unsubscribe topics
+   * absent. The plugin MAY additionally keep its own always-on
+   * subscriptions (e.g. a config-selected eager floor); the effective
+   * subscribed set is the union.
+   *
+   * Threading contract: the host invokes this on the SAME thread that calls
+   * the plugin's poll(), strictly serialized with poll()/start()/stop() —
+   * never concurrently with any of them. May be called between start()
+   * returning and the first poll(); never before start() or after stop().
+   * Unknown (not currently advertised) names are ignored, not errors — the
+   * host and the source's live topic set converge asynchronously.
+   *
+   * @p ctx is the plugin instance context (the same pointer passed to every
+   * PJ_data_source_vtable_t slot). Name views are valid only for the call.
+   */
+  bool (*set_active_topics)(void* ctx, const PJ_string_view_t* topic_names, uint64_t count, PJ_error_t* out_error)
+      PJ_NOEXCEPT;
+} PJ_topic_subscription_extension_t;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // PJ_DATA_SOURCE_TOPIC_SUBSCRIPTION_H

pj_base/include/pj_base/message_parser_protocol.h

@@ -136,6 +136,35 @@ typedef struct PJ_message_parser_vtable_t {
   bool (*classify_schema)(
       void* ctx, PJ_string_view_t type_name, PJ_bytes_view_t schema, PJ_schema_classification_t* out_classification,
       PJ_error_t* out_error) PJ_NOEXCEPT;
+
+  /**
+   * [thread-safe, same thread class as bind_schema] OPTIONAL. The flattened
+   * scalar-column manifest for a schema: exactly the leaf columns parse()
+   * will emit for messages of @p type_name, in emission order, as a JSON
+   * array:
+   *
+   *   [{"path":"pose.position.x","type":"float64"}, ...]
+   *
+   * "type" is one of: "float32","float64","int8","int16","int32","int64",
+   * "uint8","uint16","uint32","uint64","bool","string".
+   *
+   * Hosts use this to pre-create catalog columns for topics that have not
+   * yet produced a sample (lazy-subscription sources). Variable-length
+   * array fields MAY be omitted — their columns then appear on first data.
+   *
+   * MUST be consistent with parse(): a later parse() emitting field N with
+   * a different path or type than manifest entry N is a plugin bug (the
+   * host write path fails loudly on the mismatch).
+   *
+   * The returned view points to plugin-owned storage valid until the next
+   * call to this function on the same ctx. NULL slot or false return means
+   * the parser cannot pre-describe columns for this schema (the topic
+   * appears field-less until first data). Pure-functional contract: no
+   * host side-effects.
+   */
+  bool (*describe_schema_columns)(
+      void* ctx, PJ_string_view_t type_name, PJ_bytes_view_t schema, PJ_string_view_t* out_columns_json,
+      PJ_error_t* out_error) PJ_NOEXCEPT;
 } PJ_message_parser_vtable_t;
 /* The vtable above is ABI-APPENDABLE: new slots may be added at the tail;
  * host reads guard with PJ_HAS_TAIL_SLOT. See PJ_MESSAGE_PARSER_MIN_VTABLE_SIZE. */

pj_base/include/pj_base/sdk/data_source_host_views.hpp

@@ -22,6 +22,7 @@
 #include <string>
 #include <string_view>
 #include <type_traits>
+#include <vector>
 
 #include "pj_base/buffer_anchor.hpp"
 #include "pj_base/data_source_protocol.h"
@@ -83,6 +84,7 @@ constexpr uint64_t kCapabilityDirectIngest = PJ_DATA_SOURCE_CAPABILITY_DIRECT_IN
 constexpr uint64_t kCapabilityDelegatedIngest = PJ_DATA_SOURCE_CAPABILITY_DELEGATED_INGEST;
 constexpr uint64_t kCapabilitySupportsPause = PJ_DATA_SOURCE_CAPABILITY_SUPPORTS_PAUSE;
 constexpr uint64_t kCapabilityHasDialog = PJ_DATA_SOURCE_CAPABILITY_HAS_DIALOG;
+constexpr uint64_t kCapabilityLazySubscription = PJ_DATA_SOURCE_CAPABILITY_LAZY_SUBSCRIPTION;
 
 using ParserBindingHandle = PJ_parser_binding_handle_t;
 
@@ -95,6 +97,19 @@ struct ParserBindingRequest {
   std::string_view parser_config_json;
 };
 
+/// Marshal a ParserBindingRequest into its ABI counterpart. The returned
+/// struct borrows @p request's string/byte views — valid only as long as
+/// @p request is.
+[[nodiscard]] inline PJ_parser_binding_request_t toAbiParserBindingRequest(const ParserBindingRequest& request) {
+  return PJ_parser_binding_request_t{
+      .topic_name = sdk::toAbiString(request.topic_name),
+      .parser_encoding = sdk::toAbiString(request.parser_encoding),
+      .type_name = sdk::toAbiString(request.type_name),
+      .schema = sdk::toAbiBytes(request.schema),
+      .parser_config_json = sdk::toAbiString(request.parser_config_json),
+  };
+}
+
 /// Convert a PJ_error_t populated by the ABI into a descriptive std::string.
 /// Safe to call on a zero-initialized error (returns "unspecified error").
 [[nodiscard]] inline std::string errorToString(const PJ_error_t& err) {
@@ -197,13 +212,7 @@ class DataSourceRuntimeHostView {
       return unexpected("runtime host is not bound");
     }
 
-    PJ_parser_binding_request_t raw{
-        .topic_name = sdk::toAbiString(request.topic_name),
-        .parser_encoding = sdk::toAbiString(request.parser_encoding),
-        .type_name = sdk::toAbiString(request.type_name),
-        .schema = sdk::toAbiBytes(request.schema),
-        .parser_config_json = sdk::toAbiString(request.parser_config_json),
-    };
+    PJ_parser_binding_request_t raw = toAbiParserBindingRequest(request);
 
     ParserBindingHandle handle{};
     PJ_error_t err{};
@@ -304,6 +313,36 @@ class DataSourceRuntimeHostView {
     return okStatus();
   }
 
+  /**
+   * Declarative topic advertise for lazy-subscription sources: report the
+   * FULL set of topics currently available (with parser metadata), without
+   * subscribing any of them. Call again with the full set whenever the
+   * source's topic list changes. See `set_advertised_topics` in
+   * data_source_protocol.h for the reconcile semantics.
+   *
+   * Returns the distinct error "host does not support lazy subscription"
+   * when the host predates the tail slot — plugins use this to fall back
+   * to their eager subscribe-selected behavior on old hosts.
+   */
+  [[nodiscard]] Status setAdvertisedTopics(Span<const ParserBindingRequest> topics) const {
+    if (!valid()) {
+      return unexpected(std::string("runtime host is not bound"));
+    }
+    if (!PJ_HAS_TAIL_SLOT(PJ_data_source_runtime_host_vtable_t, host_.vtable, set_advertised_topics)) {
+      return unexpected(std::string("host does not support lazy subscription"));
+    }
+    std::vector<PJ_parser_binding_request_t> raw;
+    raw.reserve(topics.size());
+    for (const auto& t : topics) {
+      raw.push_back(toAbiParserBindingRequest(t));
+    }
+    PJ_error_t err{};
+    if (!host_.vtable->set_advertised_topics(host_.ctx, raw.data(), raw.size(), &err)) {
+      return unexpected(errorToString(err));
+    }
+    return okStatus();
+  }
+
   /**
    * Display a modal message box and wait for user response.
    * @return The button clicked, or kOk if the host does not support dialogs.