diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b2ba244..f5056496 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,9 +3,35 @@ 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] — not yet publicly tagged -### Host service: markers + transforms unified into `pj.data_processors.v1` (UNRELEASED BREAK) +### DataSource: per-topic pause/resume — advertise-without-subscribe + demand-driven control (ADDITIVE) + +Lets a streaming source expose *all* its topics cheaply while only transmitting data +for topics the host is actually displaying. Strictly additive (tail slot + plugin +extension, both `struct_size`/`PJ_HAS_TAIL_SLOT`-gated); every existing plugin keeps +working with no recompile, and a new plugin degrades gracefully on an old host. + +- **New capability** `PJ_DATA_SOURCE_CAPABILITY_PER_TOPIC_PAUSE = 1 << 6` (C++ mirror + `kCapabilityPerTopicPause`). +- **Plugin → host advertise**: new runtime-host tail slot + `notify_available_topics(topics, count)` carrying `PJ_available_topic_t` + (`topic_name`/`parser_encoding`/`type_name`/`schema`, mirroring + `PJ_parser_binding_request_t`), so the host can list and a-priori classify topics + (via `classify_schema`) before any data flows. C++ helper + `DataSourceRuntimeHostView::notifyAvailableTopics(Span)`. +- **Host → plugin control**: new `get_plugin_extension("pj.topic_subscription.v1")` + extension `PJ_topic_subscription_v1_t::set_active_topics(names, count)` — declarative + full active-set; the plugin diffs and subscribes/unsubscribes. Host wrapper + `DataSourceHandle::setActiveTopics(...)` (no-op when the extension is absent). +- Runtime-host vtable size grows 96 → 104 (`notify_available_topics` at offset 96); + `PJ_ABI_VERSION` (5), `PJ_DATA_SOURCE_PROTOCOL_VERSION` (4), and + `PJ_DATA_SOURCE_MIN_VTABLE_SIZE` (128) unchanged. `abi/baseline.abi` unchanged + (additions only). + +## [0.14.0] + +### Host service: markers + transforms unified into `pj.data_processors.v1` The two whole-series host-driven services were collapsed into ONE contract — Pablo's `pj.data_processors.v1` — with a string `kind` discriminator, so a plugin chooses @@ -31,7 +57,6 @@ so a host that IMPLEMENTS the service (e.g. `DataProcessorsRuntimeHost`) must up vtable fill; callers using `DataProcessorsHostView::createTransform` keep working via the shim, while `validateScript` call-sites gain a leading `kind` argument (`"transform"`). -**Versioning note.** This is an ABI/API change to a service that merged to `main`. It -ships as `0.13.0` because no PUBLIC tag has carried `pj.data_processors.v1` yet, so no -released plugin is broken. **The first public release that carries the unified -`pj.data_processors.v1` must be tagged `1.0.0`** per the pre-1.0 break rule in `CLAUDE.md`. +**Versioning note.** This was an ABI/API break to a service that had merged to `main`. +It shipped as `0.14.0` rather than `1.0.0` because no public tag had ever carried +`pj.data_processors.v1`, so no released plugin was broken by the change. diff --git a/CMakeLists.txt b/CMakeLists.txt index 6e7a09ed..55b54c79 100644 --- a/CMakeLists.txt +++ b/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 diff --git a/conanfile.py b/conanfile.py index 104accd8..2669f681 100644 --- a/conanfile.py +++ b/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) @@ -30,15 +30,17 @@ class PlotjugglerSdkConan(ConanFile): name = "plotjuggler_sdk" - # UNRELEASED BREAK: 0.13.0 unified markers + transforms into the single host - # service `pj.data_processors.v1` via a `kind` discriminator (removed the old + # 0.14.0 unified markers + transforms into the single host service + # `pj.data_processors.v1` via a `kind` discriminator (removed the old # `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 - # `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" + # `create_data_processor`/`validate_data_processor_script` with + # kind/language/flags) and added the dialog-protocol additions (radio + # column + interactive sub-panel). Shipped as a MINOR bump rather than + # 1.0.0 because no public tag had ever carried `pj.data_processors.v1`, + # so no released plugin broke. + # 0.15.0 adds DataSource per-topic pause (advertise + demand-driven + # subscription) — strictly additive. See CHANGELOG.md. + 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" diff --git a/pj_base/CLAUDE.md b/pj_base/CLAUDE.md index ab14c462..7ab1133b 100644 --- a/pj_base/CLAUDE.md +++ b/pj_base/CLAUDE.md @@ -15,7 +15,7 @@ pj_base is the **Level 0** foundation and the **SDK boundary** for plugin author - ABI numbering is **frozen**: `BuiltinObjectType` (builtin_object.hpp) and `PJ_builtin_object_type_t` (builtin_object_abi.h) share stable numeric values — never renumber; types 2 and 12 are permanently reserved. Append-only. - Every vtable slot is `PJ_NOEXCEPT`; a throw across the ABI boundary calls `std::terminate`. See the header block in `plugin_data_api.h`. - `BuiltinObject` is `std::any`, deliberately not `std::variant`, for forward-compat — recover via `std::any_cast` / `sdk::typeOf`. See `builtin/builtin_object.hpp`. -- ABI/struct-layout or signature changes require a Conan **MINOR** bump and a refreshed `abi/baseline.abi`; tail-appended slots are a PATCH (see submodule CLAUDE.md → Release Versioning). +- Versioning follows the authoritative submodule CLAUDE.md → Release Versioning: a **tail-appended slot / new capability** (backward-compatible ABI *addition*, `abidiff` shows additions-only) is a **MINOR** and does **not** touch `abi/baseline.abi`; a **struct-layout reorder or signature change** (an ABI *break*) is a **MAJOR** — that is what refreshes `abi/baseline.abi` and bumps `PJ_*_PROTOCOL_VERSION`/`PJ_ABI_VERSION`. ## Read deeper | For | Read | diff --git a/pj_base/CMakeLists.txt b/pj_base/CMakeLists.txt index c77c5fe3..80b22919 100644 --- a/pj_base/CMakeLists.txt +++ b/pj_base/CMakeLists.txt @@ -21,6 +21,12 @@ add_library(pj_base STATIC src/builtin/voxel_grid_codec.cpp src/number_parse.cpp src/type_tree.cpp + src/data_source_host_views.cpp + src/data_source_patterns.cpp + src/data_source_plugin_base.cpp + src/data_source_trampolines.cpp + src/toolbox_plugin_base.cpp + src/toolbox_trampolines.cpp ) target_include_directories(pj_base PUBLIC $ @@ -92,6 +98,7 @@ if(PJ_BUILD_TESTS) tests/image_annotations_decoder_test.cpp tests/media_metadata_test.cpp tests/push_message_test.cpp + tests/notify_available_topics_test.cpp tests/image_codec_test.cpp tests/depth_image_codec_test.cpp tests/point_cloud_codec_test.cpp diff --git a/pj_base/include/pj_base/data_source_protocol.h b/pj_base/include/pj_base/data_source_protocol.h index 4c31f6d4..77fe3fd9 100644 --- a/pj_base/include/pj_base/data_source_protocol.h +++ b/pj_base/include/pj_base/data_source_protocol.h @@ -131,6 +131,8 @@ 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. */ + PJ_DATA_SOURCE_CAPABILITY_PER_TOPIC_PAUSE = + 1ull << 6, /**< Host may pause/resume individual topics via the "pj.topic_subscription.v1" extension. */ }; /** Opaque handle returned by ensure_parser_binding, used with push_message. */ @@ -200,6 +202,20 @@ typedef struct { PJ_string_view_t parser_config_json; /**< Optional JSON config for the parser. */ } PJ_parser_binding_request_t; +/** + * A topic the source knows it can stream but has not (necessarily) + * subscribed. Handed to the host via notify_available_topics so the host can + * list it and classify it a priori (via the parser's classify_schema) with no + * data flowing. Carries the same parser inputs as PJ_parser_binding_request_t + * so the host can classify and, later, bind identically. + */ +typedef struct { + PJ_string_view_t topic_name; /**< Canonical topic name (the host/engine key). */ + PJ_string_view_t parser_encoding; /**< Encoding name, e.g. "protobuf", "ros2msg". */ + PJ_string_view_t type_name; /**< Message type name (encoding-specific). */ + PJ_bytes_view_t schema; /**< Optional schema bytes for a-priori classify_schema. */ +} PJ_available_topic_t; + /** * DataSource runtime host vtable — control-plane callbacks provided by the * host and delivered to the plugin via the service registry under the name @@ -207,8 +223,10 @@ typedef struct { * * The plugin calls these to report progress, send diagnostic messages, * notify state changes, and (for delegated ingest) bind parsers and push - * raw message payloads. All calls are made on the thread that called - * start(). + * raw message payloads. The per-slot thread tag governs: the host may drive + * start()/stop() on one thread and poll() (with its [stream-thread] + * callbacks) on another, so a host must treat [stream-thread] callbacks as + * arriving off its UI thread and marshal accordingly. * * Fallible calls take a `PJ_error_t* out_error` which the callee populates * on failure. Callers may pass NULL if they don't need the detail. @@ -327,6 +345,19 @@ 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] Advertise the set of topics this source can stream but has + * not (necessarily) subscribed. The host lists them (e.g. in the curve tree) + * and classifies each a priori via classify_schema, WITHOUT any data + * flowing; subscription happens separately (see the "pj.topic_subscription.v1" + * plugin extension). Call again to update as topics appear/vanish; each call + * is the full current set for this source. The plugin calls this on the + * poll/stream thread; the host is responsible for marshaling to its own + * thread. Tail slot — readers MUST gate access with PJ_HAS_TAIL_SLOT. + */ + bool (*notify_available_topics)(void* ctx, const PJ_available_topic_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. */ @@ -454,6 +485,40 @@ typedef struct PJ_data_source_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_DATA_SOURCE_MIN_VTABLE_SIZE. */ +/** + * Reverse-DNS id of the per-topic subscription control extension, queried via + * get_plugin_extension. A source advertising PJ_DATA_SOURCE_CAPABILITY_PER_TOPIC_PAUSE + * returns a pointer to a PJ_topic_subscription_v1_t for this id; every other + * source returns NULL. + */ +#define PJ_TOPIC_SUBSCRIPTION_EXTENSION_V1 "pj.topic_subscription.v1" + +/** + * Host→plugin per-topic subscription control ("pj.topic_subscription.v1"). + * + * The host passes the desired ACTIVE topic set — declarative and complete — + * and the plugin diffs it against its current subscriptions, subscribing to + * newly-active topics and unsubscribing (pausing) the rest, so no data flows + * for a topic that is not being displayed. This is a strict superset control: + * an empty set pauses everything. + * + * The plugin owns this struct; it must stay valid for the plugin instance + * lifetime. IMPORTANT: the `ctx` passed to set_active_topics is the plugin + * INSTANCE (the host's DataSourceHandle::context()), NOT the extension pointer + * returned by get_plugin_extension. + */ +typedef struct { + uint32_t struct_size; /**< sizeof(PJ_topic_subscription_v1_t) — for forward-compatible growth. */ + /** + * [main-thread] Set the full active-topic set. Entered on the host thread; + * the plugin marshals to its own stream thread. Returns false + populates + * out_error on failure. Names the source does not recognize are ignored + * (not an error). + */ + 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_v1_t; + /** Signature of the exported entry point: `PJ_get_data_source_vtable`. */ typedef const PJ_data_source_vtable_t* (*PJ_get_data_source_vtable_fn)(void); diff --git a/pj_base/include/pj_base/sdk/data_source_host_views.hpp b/pj_base/include/pj_base/sdk/data_source_host_views.hpp index 848e0d2e..3c42366f 100644 --- a/pj_base/include/pj_base/sdk/data_source_host_views.hpp +++ b/pj_base/include/pj_base/sdk/data_source_host_views.hpp @@ -83,6 +83,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 kCapabilityPerTopicPause = PJ_DATA_SOURCE_CAPABILITY_PER_TOPIC_PAUSE; using ParserBindingHandle = PJ_parser_binding_handle_t; @@ -95,22 +96,21 @@ struct ParserBindingRequest { std::string_view parser_config_json; }; +/// A topic the source can stream but has not (necessarily) subscribed. Handed +/// to the host via DataSourceRuntimeHostView::notifyAvailableTopics so the host +/// lists and a-priori classifies it (via classify_schema) with no data flowing. +/// Mirrors ParserBindingRequest's parser inputs so the host can classify and, +/// later, bind identically. `schema` may be empty when the encoding needs none. +struct AvailableTopic { + std::string_view topic_name; + std::string_view parser_encoding; + std::string_view type_name; + Span schema; +}; + /// 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) { - std::string out; - if (err.domain[0] != '\0') { - out.append(err.domain); - out.append(": "); - } - if (err.message[0] != '\0') { - out.append(err.message); - } - if (out.empty()) { - out = "unspecified error"; - } - return out; -} +[[nodiscard]] std::string errorToString(const PJ_error_t& err); class ParserIngestHostView; @@ -142,16 +142,7 @@ class DataSourceRuntimeHostView { } /// Begin a progress bar. Returns an error if the host refused to start it. - [[nodiscard]] Status progressStart(std::string_view label, uint64_t total_steps, bool cancellable) const { - if (!valid() || host_.vtable->progress_start == nullptr) { - return unexpected("runtime host is not bound"); - } - PJ_error_t err{}; - if (!host_.vtable->progress_start(host_.ctx, sdk::toAbiString(label), total_steps, cancellable, &err)) { - return unexpected(errorToString(err)); - } - return okStatus(); - } + [[nodiscard]] Status progressStart(std::string_view label, uint64_t total_steps, bool cancellable) const; /// Advance progress. Returns true to continue, false if the user cancelled. [[nodiscard]] bool progressUpdate(uint64_t current_step) const { @@ -192,26 +183,15 @@ class DataSourceRuntimeHostView { } /// Bind (or look up) a parser for delegated ingest. - [[nodiscard]] Expected ensureParserBinding(const ParserBindingRequest& request) const { - if (!valid() || host_.vtable->ensure_parser_binding == nullptr) { - return unexpected("runtime host is not bound"); - } + [[nodiscard]] Expected ensureParserBinding(const ParserBindingRequest& request) const; - 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), - }; - - ParserBindingHandle handle{}; - PJ_error_t err{}; - if (!host_.vtable->ensure_parser_binding(host_.ctx, &raw, &handle, &err)) { - return unexpected(errorToString(err)); - } - return handle; - } + /// Advertise the full set of topics this source can stream but has not + /// (necessarily) subscribed, so the host can list and a-priori classify them + /// with no data flowing. Each call replaces the previously-advertised set for + /// this source. Returns an error if the host does not expose the tail slot + /// (old host) — callers may treat that as "advertisement unsupported" and + /// fall back to legacy behavior. Call on the poll/stream thread. + [[nodiscard]] Status notifyAvailableTopics(Span topics) const; /// Push a message via a deferred FetchMessageData callable. The DataSource /// hands the host a callable that produces the payload bytes when invoked. @@ -309,21 +289,7 @@ class DataSourceRuntimeHostView { * @return The button clicked, or kOk if the host does not support dialogs. */ [[nodiscard]] MessageBoxButton showMessageBox( - MessageBoxType type, std::string_view title, std::string_view message, int buttons = 0) const { - if (!valid() || host_.vtable->show_message_box == nullptr) { - if (buttons & static_cast(MessageBoxButton::kContinue)) { - return MessageBoxButton::kContinue; - } - if (buttons & static_cast(MessageBoxButton::kYes)) { - return MessageBoxButton::kYes; - } - return MessageBoxButton::kOk; - } - int result = host_.vtable->show_message_box( - host_.ctx, static_cast(type), sdk::toAbiString(title), sdk::toAbiString(message), - buttons == 0 ? PJ_MSG_BTN_OK : buttons); - return static_cast(result); - } + MessageBoxType type, std::string_view title, std::string_view message, int buttons = 0) const; void showInfo(std::string_view title, std::string_view message) const { (void)showMessageBox(MessageBoxType::kInfo, title, message, static_cast(MessageBoxButton::kOk)); diff --git a/pj_base/include/pj_base/sdk/data_source_patterns.hpp b/pj_base/include/pj_base/sdk/data_source_patterns.hpp index e9c5d7d7..608d0a04 100644 --- a/pj_base/include/pj_base/sdk/data_source_patterns.hpp +++ b/pj_base/include/pj_base/sdk/data_source_patterns.hpp @@ -79,23 +79,7 @@ class FileSourceBase : public DataSourcePluginBase { /// writeHost() and runtimeHost() are available. virtual Status importData() = 0; - Status start() final { - state_ = DataSourceState::kStarting; - runtimeHost().notifyState(state_); - - auto status = importData(); - runtimeHost().progressFinish(); // safe no-op if no progress was started - if (!status) { - state_ = DataSourceState::kFailed; - runtimeHost().notifyState(state_); - return status; - } - - state_ = DataSourceState::kStopped; - runtimeHost().notifyState(state_); - runtimeHost().requestStop(DataSourceState::kStopped, "import complete"); - return okStatus(); - } + Status start() final; void stop() final { state_ = DataSourceState::kStopped; @@ -143,31 +127,13 @@ class StreamSourceBase : public DataSourcePluginBase { /// Must be idempotent. virtual void onStop() = 0; - Status start() final { - state_ = DataSourceState::kStarting; - runtimeHost().notifyState(state_); - - auto status = onStart(); - if (!status) { - state_ = DataSourceState::kFailed; - runtimeHost().notifyState(state_); - return status; - } - - state_ = DataSourceState::kRunning; - runtimeHost().notifyState(state_); - return okStatus(); - } + Status start() final; Status poll() final { return onPoll(); } - void stop() final { - onStop(); - state_ = DataSourceState::kStopped; - runtimeHost().notifyState(state_); - } + void stop() final; DataSourceState currentState() const final { return state_; diff --git a/pj_base/include/pj_base/sdk/data_source_plugin_base.hpp b/pj_base/include/pj_base/sdk/data_source_plugin_base.hpp index 23f29100..93632e35 100644 --- a/pj_base/include/pj_base/sdk/data_source_plugin_base.hpp +++ b/pj_base/include/pj_base/sdk/data_source_plugin_base.hpp @@ -67,26 +67,7 @@ class DataSourcePluginBase { /// unused; hosts without an ObjectStore bound simply don't register it. /// Override to request additional services (e.g. colormap), or to relax /// the default requirement. - virtual Status bind(sdk::ServiceRegistry services) { - auto write = services.require(); - if (!write) { - return unexpected(std::move(write).error()); - } - write_host_view_ = *write; - - auto runtime = services.require(); - if (!runtime) { - return unexpected(std::move(runtime).error()); - } - runtime_host_view_ = *runtime; - - if (auto object_write = services.get()) { - object_write_host_view_ = *object_write; - } - - service_registry_ = services; - return okStatus(); - } + virtual Status bind(sdk::ServiceRegistry services); /// Serialize plugin configuration to JSON. Default returns "{}". virtual std::string saveConfig() const { @@ -224,9 +205,6 @@ class DataSourcePluginBase { } // namespace PJ -// Out-of-line trampoline definitions — separated to keep the public API header concise. -#include "pj_base/sdk/detail/data_source_trampolines.hpp" - /** * Export a DataSourcePluginBase subclass as a shared-library plugin. * diff --git a/pj_base/include/pj_base/sdk/toolbox_plugin_base.hpp b/pj_base/include/pj_base/sdk/toolbox_plugin_base.hpp index 3cc6849f..7876bbc7 100644 --- a/pj_base/include/pj_base/sdk/toolbox_plugin_base.hpp +++ b/pj_base/include/pj_base/sdk/toolbox_plugin_base.hpp @@ -63,36 +63,11 @@ class ToolboxRuntimeHostView { /// pushMessage() per record. Drive the returned view from a single worker /// thread; unlike reportMessage/notifyDataChanged, parser ingest is not /// GUI-marshalled. - [[nodiscard]] Expected createParserIngest(uint32_t data_source_id) const { - if (!valid()) { - return unexpected("toolbox runtime host is not bound"); - } - if (!PJ_HAS_TAIL_SLOT(PJ_toolbox_runtime_host_vtable_t, host_.vtable, create_parser_ingest)) { - return unexpected("toolbox runtime host does not support create_parser_ingest (older host)"); - } - PJ_data_source_runtime_host_t raw{}; - PJ_error_t err{}; - if (!host_.vtable->create_parser_ingest(host_.ctx, data_source_id, &raw, &err)) { - return unexpected(errorToString(err)); - } - return ParserIngestHostView{raw}; - } + [[nodiscard]] Expected createParserIngest(uint32_t data_source_id) const; /// Flush + destroy the context. Idempotent. The view returned by /// createParserIngest must not be used afterwards. - [[nodiscard]] Status releaseParserIngest(uint32_t data_source_id) const { - if (!valid()) { - return unexpected("toolbox runtime host is not bound"); - } - if (!PJ_HAS_TAIL_SLOT(PJ_toolbox_runtime_host_vtable_t, host_.vtable, release_parser_ingest)) { - return unexpected("toolbox runtime host does not support release_parser_ingest (older host)"); - } - PJ_error_t err{}; - if (!host_.vtable->release_parser_ingest(host_.ctx, data_source_id, &err)) { - return unexpected(errorToString(err)); - } - return okStatus(); - } + [[nodiscard]] Status releaseParserIngest(uint32_t data_source_id) const; [[nodiscard]] const PJ_toolbox_runtime_host_t& raw() const { return host_; @@ -139,32 +114,7 @@ class ToolboxPluginBase { /// - "pj.toolbox_object_read.v1" → ObjectReadHost (optional) /// /// Override to acquire additional services or relax defaults. - virtual Status bind(sdk::ServiceRegistry services) { - auto host = services.require(); - if (!host) { - return unexpected(std::move(host).error()); - } - toolbox_host_view_ = *host; - - auto runtime = services.require(); - if (!runtime) { - return unexpected(std::move(runtime).error()); - } - runtime_host_view_ = *runtime; - - // Colormap is optional — acquire opportunistically. - if (auto cm = services.get()) { - colormap_view_ = *cm; - } - - // Object read is optional — transformer-style toolboxes resolve it. - if (auto obj = services.get()) { - object_read_host_view_ = *obj; - } - - service_registry_ = services; - return okStatus(); - } + virtual Status bind(sdk::ServiceRegistry services); virtual std::string saveConfig() const { return "{}"; @@ -272,8 +222,6 @@ class ToolboxPluginBase { } // namespace PJ -#include "pj_base/sdk/detail/toolbox_trampolines.hpp" - #define PJ_TOOLBOX_PLUGIN(ClassName, manifest) \ PJ_EXPORT_PLUGIN_ABI_VERSION(PJ_TOOLBOX_EXPORT) \ extern "C" PJ_TOOLBOX_EXPORT const PJ_toolbox_vtable_t* PJ_get_toolbox_vtable() noexcept { \ diff --git a/pj_base/src/data_source_host_views.cpp b/pj_base/src/data_source_host_views.cpp new file mode 100644 index 00000000..1fa73a28 --- /dev/null +++ b/pj_base/src/data_source_host_views.cpp @@ -0,0 +1,100 @@ +// Copyright 2026 Davide Faconti +// SPDX-License-Identifier: Apache-2.0 + +#include "pj_base/sdk/data_source_host_views.hpp" + +#include + +namespace PJ { + +std::string errorToString(const PJ_error_t& err) { + std::string out; + if (err.domain[0] != '\0') { + out.append(err.domain); + out.append(": "); + } + if (err.message[0] != '\0') { + out.append(err.message); + } + if (out.empty()) { + out = "unspecified error"; + } + return out; +} + +Status DataSourceRuntimeHostView::progressStart(std::string_view label, uint64_t total_steps, bool cancellable) const { + if (!valid() || host_.vtable->progress_start == nullptr) { + return unexpected("runtime host is not bound"); + } + PJ_error_t err{}; + if (!host_.vtable->progress_start(host_.ctx, sdk::toAbiString(label), total_steps, cancellable, &err)) { + return unexpected(errorToString(err)); + } + return okStatus(); +} + +Expected DataSourceRuntimeHostView::ensureParserBinding( + const ParserBindingRequest& request) const { + if (!valid() || host_.vtable->ensure_parser_binding == nullptr) { + 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), + }; + + ParserBindingHandle handle{}; + PJ_error_t err{}; + if (!host_.vtable->ensure_parser_binding(host_.ctx, &raw, &handle, &err)) { + return unexpected(errorToString(err)); + } + return handle; +} + +Status DataSourceRuntimeHostView::notifyAvailableTopics(Span 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, notify_available_topics)) { + return unexpected(std::string("runtime host does not expose notify_available_topics")); + } + std::vector raw; + raw.reserve(topics.size()); + for (const auto& t : topics) { + raw.push_back( + PJ_available_topic_t{ + .topic_name = sdk::toAbiString(t.topic_name), + .parser_encoding = sdk::toAbiString(t.parser_encoding), + .type_name = sdk::toAbiString(t.type_name), + .schema = sdk::toAbiBytes(t.schema), + }); + } + PJ_error_t err{}; + if (!host_.vtable->notify_available_topics(host_.ctx, raw.data(), raw.size(), &err)) { + return unexpected(errorToString(err)); + } + return okStatus(); +} + +MessageBoxButton DataSourceRuntimeHostView::showMessageBox( + MessageBoxType type, std::string_view title, std::string_view message, int buttons) const { + if (!valid() || host_.vtable->show_message_box == nullptr) { + if (buttons & static_cast(MessageBoxButton::kContinue)) { + return MessageBoxButton::kContinue; + } + if (buttons & static_cast(MessageBoxButton::kYes)) { + return MessageBoxButton::kYes; + } + return MessageBoxButton::kOk; + } + int result = host_.vtable->show_message_box( + host_.ctx, static_cast(type), sdk::toAbiString(title), sdk::toAbiString(message), + buttons == 0 ? PJ_MSG_BTN_OK : buttons); + return static_cast(result); +} + +} // namespace PJ diff --git a/pj_base/src/data_source_patterns.cpp b/pj_base/src/data_source_patterns.cpp new file mode 100644 index 00000000..2b616f1b --- /dev/null +++ b/pj_base/src/data_source_patterns.cpp @@ -0,0 +1,48 @@ +// Copyright 2026 Davide Faconti +// SPDX-License-Identifier: Apache-2.0 + +#include "pj_base/sdk/data_source_patterns.hpp" + +namespace PJ { + +Status FileSourceBase::start() { + state_ = DataSourceState::kStarting; + runtimeHost().notifyState(state_); + + auto status = importData(); + runtimeHost().progressFinish(); // safe no-op if no progress was started + if (!status) { + state_ = DataSourceState::kFailed; + runtimeHost().notifyState(state_); + return status; + } + + state_ = DataSourceState::kStopped; + runtimeHost().notifyState(state_); + runtimeHost().requestStop(DataSourceState::kStopped, "import complete"); + return okStatus(); +} + +Status StreamSourceBase::start() { + state_ = DataSourceState::kStarting; + runtimeHost().notifyState(state_); + + auto status = onStart(); + if (!status) { + state_ = DataSourceState::kFailed; + runtimeHost().notifyState(state_); + return status; + } + + state_ = DataSourceState::kRunning; + runtimeHost().notifyState(state_); + return okStatus(); +} + +void StreamSourceBase::stop() { + onStop(); + state_ = DataSourceState::kStopped; + runtimeHost().notifyState(state_); +} + +} // namespace PJ diff --git a/pj_base/src/data_source_plugin_base.cpp b/pj_base/src/data_source_plugin_base.cpp new file mode 100644 index 00000000..3f45077e --- /dev/null +++ b/pj_base/src/data_source_plugin_base.cpp @@ -0,0 +1,29 @@ +// Copyright 2026 Davide Faconti +// SPDX-License-Identifier: Apache-2.0 + +#include "pj_base/sdk/data_source_plugin_base.hpp" + +namespace PJ { + +Status DataSourcePluginBase::bind(sdk::ServiceRegistry services) { + auto write = services.require(); + if (!write) { + return unexpected(std::move(write).error()); + } + write_host_view_ = *write; + + auto runtime = services.require(); + if (!runtime) { + return unexpected(std::move(runtime).error()); + } + runtime_host_view_ = *runtime; + + if (auto object_write = services.get()) { + object_write_host_view_ = *object_write; + } + + service_registry_ = services; + return okStatus(); +} + +} // namespace PJ diff --git a/pj_base/include/pj_base/sdk/detail/data_source_trampolines.hpp b/pj_base/src/data_source_trampolines.cpp similarity index 80% rename from pj_base/include/pj_base/sdk/detail/data_source_trampolines.hpp rename to pj_base/src/data_source_trampolines.cpp index b0048d9a..c39f325b 100644 --- a/pj_base/include/pj_base/sdk/detail/data_source_trampolines.hpp +++ b/pj_base/src/data_source_trampolines.cpp @@ -1,8 +1,7 @@ /** - * @file detail/data_source_trampolines.hpp - * @brief Out-of-line definitions for DataSourcePluginBase C ABI trampolines (v4). + * @file data_source_trampolines.cpp + * @brief C ABI trampolines for DataSourcePluginBase (v4). * - * Included automatically by data_source_plugin_base.hpp — do not include directly. * Each trampoline wraps a virtual call with try-catch for full exception * safety across the C ABI boundary and populates `PJ_error_t*` out-params * via the plugin's per-instance error buffer. Every trampoline is `noexcept` @@ -11,17 +10,17 @@ // Copyright 2026 Davide Faconti // SPDX-License-Identifier: Apache-2.0 -#pragma once +#include "pj_base/sdk/data_source_plugin_base.hpp" namespace PJ { -inline void DataSourcePluginBase::trampoline_destroy(void* ctx) noexcept { +void DataSourcePluginBase::trampoline_destroy(void* ctx) noexcept { try { delete static_cast(ctx); } catch (...) {} } -inline uint64_t DataSourcePluginBase::trampoline_capabilities(void* ctx) noexcept { +uint64_t DataSourcePluginBase::trampoline_capabilities(void* ctx) noexcept { auto* self = static_cast(ctx); try { return self->capabilities(); @@ -34,8 +33,7 @@ inline uint64_t DataSourcePluginBase::trampoline_capabilities(void* ctx) noexcep } } -inline bool DataSourcePluginBase::trampoline_bind( - void* ctx, PJ_service_registry_t registry, PJ_error_t* out_error) noexcept { +bool DataSourcePluginBase::trampoline_bind(void* ctx, PJ_service_registry_t registry, PJ_error_t* out_error) noexcept { auto* self = static_cast(ctx); try { auto status = self->bind(sdk::ServiceRegistry(registry)); @@ -53,7 +51,7 @@ inline bool DataSourcePluginBase::trampoline_bind( } } -inline bool DataSourcePluginBase::trampoline_save_config( +bool DataSourcePluginBase::trampoline_save_config( void* ctx, PJ_string_view_t* out_json, PJ_error_t* out_error) noexcept { auto* self = static_cast(ctx); if (out_json == nullptr) { @@ -74,7 +72,7 @@ inline bool DataSourcePluginBase::trampoline_save_config( } } -inline bool DataSourcePluginBase::trampoline_load_config( +bool DataSourcePluginBase::trampoline_load_config( void* ctx, PJ_string_view_t config_json, PJ_error_t* out_error) noexcept { auto* self = static_cast(ctx); try { @@ -95,7 +93,7 @@ inline bool DataSourcePluginBase::trampoline_load_config( } } -inline bool DataSourcePluginBase::trampoline_start(void* ctx, PJ_error_t* out_error) noexcept { +bool DataSourcePluginBase::trampoline_start(void* ctx, PJ_error_t* out_error) noexcept { auto* self = static_cast(ctx); try { auto status = self->start(); @@ -113,7 +111,7 @@ inline bool DataSourcePluginBase::trampoline_start(void* ctx, PJ_error_t* out_er } } -inline void DataSourcePluginBase::trampoline_stop(void* ctx) noexcept { +void DataSourcePluginBase::trampoline_stop(void* ctx) noexcept { auto* self = static_cast(ctx); try { self->stop(); @@ -124,7 +122,7 @@ inline void DataSourcePluginBase::trampoline_stop(void* ctx) noexcept { } } -inline bool DataSourcePluginBase::trampoline_pause(void* ctx, PJ_error_t* out_error) noexcept { +bool DataSourcePluginBase::trampoline_pause(void* ctx, PJ_error_t* out_error) noexcept { auto* self = static_cast(ctx); try { auto status = self->pause(); @@ -142,7 +140,7 @@ inline bool DataSourcePluginBase::trampoline_pause(void* ctx, PJ_error_t* out_er } } -inline bool DataSourcePluginBase::trampoline_resume(void* ctx, PJ_error_t* out_error) noexcept { +bool DataSourcePluginBase::trampoline_resume(void* ctx, PJ_error_t* out_error) noexcept { auto* self = static_cast(ctx); try { auto status = self->resume(); @@ -160,7 +158,7 @@ inline bool DataSourcePluginBase::trampoline_resume(void* ctx, PJ_error_t* out_e } } -inline bool DataSourcePluginBase::trampoline_poll(void* ctx, PJ_error_t* out_error) noexcept { +bool DataSourcePluginBase::trampoline_poll(void* ctx, PJ_error_t* out_error) noexcept { auto* self = static_cast(ctx); try { auto status = self->poll(); @@ -178,7 +176,7 @@ inline bool DataSourcePluginBase::trampoline_poll(void* ctx, PJ_error_t* out_err } } -inline PJ_data_source_state_t DataSourcePluginBase::trampoline_current_state(void* ctx) noexcept { +PJ_data_source_state_t DataSourcePluginBase::trampoline_current_state(void* ctx) noexcept { auto* self = static_cast(ctx); try { return static_cast(self->currentState()); @@ -187,7 +185,7 @@ inline PJ_data_source_state_t DataSourcePluginBase::trampoline_current_state(voi } } -inline PJ_borrowed_dialog_t DataSourcePluginBase::trampoline_get_dialog(void* ctx) noexcept { +PJ_borrowed_dialog_t DataSourcePluginBase::trampoline_get_dialog(void* ctx) noexcept { auto* self = static_cast(ctx); try { return self->getDialog(); @@ -196,7 +194,7 @@ inline PJ_borrowed_dialog_t DataSourcePluginBase::trampoline_get_dialog(void* ct } } -inline const void* DataSourcePluginBase::trampoline_get_plugin_extension(void* ctx, PJ_string_view_t id) noexcept { +const void* DataSourcePluginBase::trampoline_get_plugin_extension(void* ctx, PJ_string_view_t id) noexcept { auto* self = static_cast(ctx); try { std::string_view sv = id.data == nullptr ? std::string_view{} : std::string_view(id.data, id.size); diff --git a/pj_base/src/toolbox_plugin_base.cpp b/pj_base/src/toolbox_plugin_base.cpp new file mode 100644 index 00000000..39dd0c41 --- /dev/null +++ b/pj_base/src/toolbox_plugin_base.cpp @@ -0,0 +1,64 @@ +// Copyright 2026 Davide Faconti +// SPDX-License-Identifier: Apache-2.0 + +#include "pj_base/sdk/toolbox_plugin_base.hpp" + +namespace PJ { + +Expected ToolboxRuntimeHostView::createParserIngest(uint32_t data_source_id) const { + if (!valid()) { + return unexpected("toolbox runtime host is not bound"); + } + if (!PJ_HAS_TAIL_SLOT(PJ_toolbox_runtime_host_vtable_t, host_.vtable, create_parser_ingest)) { + return unexpected("toolbox runtime host does not support create_parser_ingest (older host)"); + } + PJ_data_source_runtime_host_t raw{}; + PJ_error_t err{}; + if (!host_.vtable->create_parser_ingest(host_.ctx, data_source_id, &raw, &err)) { + return unexpected(errorToString(err)); + } + return ParserIngestHostView{raw}; +} + +Status ToolboxRuntimeHostView::releaseParserIngest(uint32_t data_source_id) const { + if (!valid()) { + return unexpected("toolbox runtime host is not bound"); + } + if (!PJ_HAS_TAIL_SLOT(PJ_toolbox_runtime_host_vtable_t, host_.vtable, release_parser_ingest)) { + return unexpected("toolbox runtime host does not support release_parser_ingest (older host)"); + } + PJ_error_t err{}; + if (!host_.vtable->release_parser_ingest(host_.ctx, data_source_id, &err)) { + return unexpected(errorToString(err)); + } + return okStatus(); +} + +Status ToolboxPluginBase::bind(sdk::ServiceRegistry services) { + auto host = services.require(); + if (!host) { + return unexpected(std::move(host).error()); + } + toolbox_host_view_ = *host; + + auto runtime = services.require(); + if (!runtime) { + return unexpected(std::move(runtime).error()); + } + runtime_host_view_ = *runtime; + + // Colormap is optional — acquire opportunistically. + if (auto cm = services.get()) { + colormap_view_ = *cm; + } + + // Object read is optional — transformer-style toolboxes resolve it. + if (auto obj = services.get()) { + object_read_host_view_ = *obj; + } + + service_registry_ = services; + return okStatus(); +} + +} // namespace PJ diff --git a/pj_base/include/pj_base/sdk/detail/toolbox_trampolines.hpp b/pj_base/src/toolbox_trampolines.cpp similarity index 76% rename from pj_base/include/pj_base/sdk/detail/toolbox_trampolines.hpp rename to pj_base/src/toolbox_trampolines.cpp index 1e953df7..d7f230d6 100644 --- a/pj_base/include/pj_base/sdk/detail/toolbox_trampolines.hpp +++ b/pj_base/src/toolbox_trampolines.cpp @@ -1,23 +1,23 @@ /** - * @file detail/toolbox_trampolines.hpp - * @brief Out-of-line C ABI trampolines for ToolboxPluginBase (v4). + * @file toolbox_trampolines.cpp + * @brief C ABI trampolines for ToolboxPluginBase (v4). * * Every trampoline is `noexcept` — the v4 vtable requires it. */ // Copyright 2026 Davide Faconti // SPDX-License-Identifier: Apache-2.0 -#pragma once +#include "pj_base/sdk/toolbox_plugin_base.hpp" namespace PJ { -inline void ToolboxPluginBase::trampoline_destroy(void* ctx) noexcept { +void ToolboxPluginBase::trampoline_destroy(void* ctx) noexcept { try { delete static_cast(ctx); } catch (...) {} } -inline uint64_t ToolboxPluginBase::trampoline_capabilities(void* ctx) noexcept { +uint64_t ToolboxPluginBase::trampoline_capabilities(void* ctx) noexcept { auto* self = static_cast(ctx); try { return self->capabilities(); @@ -26,8 +26,7 @@ inline uint64_t ToolboxPluginBase::trampoline_capabilities(void* ctx) noexcept { } } -inline bool ToolboxPluginBase::trampoline_bind( - void* ctx, PJ_service_registry_t registry, PJ_error_t* out_error) noexcept { +bool ToolboxPluginBase::trampoline_bind(void* ctx, PJ_service_registry_t registry, PJ_error_t* out_error) noexcept { auto* self = static_cast(ctx); try { auto status = self->bind(sdk::ServiceRegistry(registry)); @@ -45,8 +44,7 @@ inline bool ToolboxPluginBase::trampoline_bind( } } -inline bool ToolboxPluginBase::trampoline_save_config( - void* ctx, PJ_string_view_t* out_json, PJ_error_t* out_error) noexcept { +bool ToolboxPluginBase::trampoline_save_config(void* ctx, PJ_string_view_t* out_json, PJ_error_t* out_error) noexcept { auto* self = static_cast(ctx); if (out_json == nullptr) { self->storeError(out_error, 2, "plugin", "save_config called with null out_json"); @@ -66,7 +64,7 @@ inline bool ToolboxPluginBase::trampoline_save_config( } } -inline bool ToolboxPluginBase::trampoline_load_config( +bool ToolboxPluginBase::trampoline_load_config( void* ctx, PJ_string_view_t config_json, PJ_error_t* out_error) noexcept { auto* self = static_cast(ctx); try { @@ -87,7 +85,7 @@ inline bool ToolboxPluginBase::trampoline_load_config( } } -inline PJ_borrowed_dialog_t ToolboxPluginBase::trampoline_get_dialog(void* ctx) noexcept { +PJ_borrowed_dialog_t ToolboxPluginBase::trampoline_get_dialog(void* ctx) noexcept { auto* self = static_cast(ctx); try { return self->getDialog(); @@ -96,14 +94,14 @@ inline PJ_borrowed_dialog_t ToolboxPluginBase::trampoline_get_dialog(void* ctx) } } -inline void ToolboxPluginBase::trampoline_on_data_changed(void* ctx) noexcept { +void ToolboxPluginBase::trampoline_on_data_changed(void* ctx) noexcept { auto* self = static_cast(ctx); try { self->onDataChanged(); } catch (...) {} } -inline const void* ToolboxPluginBase::trampoline_get_plugin_extension(void* ctx, PJ_string_view_t id) noexcept { +const void* ToolboxPluginBase::trampoline_get_plugin_extension(void* ctx, PJ_string_view_t id) noexcept { auto* self = static_cast(ctx); try { std::string_view sv = id.data == nullptr ? std::string_view{} : std::string_view(id.data, id.size); diff --git a/pj_base/tests/abi_layout_sentinels_test.cpp b/pj_base/tests/abi_layout_sentinels_test.cpp index 5be7e748..4f4d4fcf 100644 --- a/pj_base/tests/abi_layout_sentinels_test.cpp +++ b/pj_base/tests/abi_layout_sentinels_test.cpp @@ -153,7 +153,10 @@ static_assert( "list_available_encodings slot pinned"); static_assert(offsetof(PJ_data_source_runtime_host_vtable_t, push_message) == 88, "push_message tail slot pinned"); static_assert( - sizeof(PJ_data_source_runtime_host_vtable_t) == 96, "Runtime host vtable size (update deliberately on append)"); + offsetof(PJ_data_source_runtime_host_vtable_t, notify_available_topics) == 96, + "notify_available_topics tail slot pinned"); +static_assert( + sizeof(PJ_data_source_runtime_host_vtable_t) == 104, "Runtime host vtable size (update deliberately on append)"); // --- Write-host vtables (ABI-APPENDABLE within v4) -------------------------- static_assert(offsetof(PJ_source_write_host_vtable_t, abi_version) == 0, "source write host prefix pinned"); diff --git a/pj_base/tests/notify_available_topics_test.cpp b/pj_base/tests/notify_available_topics_test.cpp new file mode 100644 index 00000000..8b102840 --- /dev/null +++ b/pj_base/tests/notify_available_topics_test.cpp @@ -0,0 +1,219 @@ +// Copyright 2026 Davide Faconti +// SPDX-License-Identifier: Apache-2.0 + +// Tests for the per-topic pause/resume ABI additions: +// +// 1. DataSourceRuntimeHostView::notifyAvailableTopics flows each advertised +// topic (name/encoding/type/schema) through the new runtime-host tail +// slot notify_available_topics. +// 2. When the host predates the slot (short struct_size / NULL field), the +// call returns an explicit error rather than degrading silently — so a +// NEW plugin on an OLD host can detect the absence and fall back. +// 3. The "pj.topic_subscription.v1" plugin extension round-trips a full +// active-topic set to the plugin, and an unknown extension id yields +// nullptr (an OLD host on a NEW plugin gets a clean no-op). + +#include + +#include +#include +#include + +#include "pj_base/data_source_protocol.h" +#include "pj_base/sdk/data_source_host_views.hpp" +#include "pj_base/sdk/data_source_plugin_base.hpp" + +namespace { + +// One captured advertised topic (strings/bytes copied out of the borrowed views). +struct CapturedTopic { + std::string topic_name; + std::string parser_encoding; + std::string type_name; + std::vector schema; +}; + +// Mock runtime host — captures notify_available_topics calls. +class MockHost { + public: + MockHost() { + vtable_.protocol_version = 1; + vtable_.struct_size = sizeof(PJ_data_source_runtime_host_vtable_t); + vtable_.notify_available_topics = &MockHost::notifyThunk; + host_.ctx = this; + host_.vtable = &vtable_; + } + + // Simulate an older host that predates the slot: shrink struct_size to the + // offset of notify_available_topics AND null the field. + void dropNotifyAvailableTopics() { + vtable_.notify_available_topics = nullptr; + vtable_.struct_size = offsetof(PJ_data_source_runtime_host_vtable_t, notify_available_topics); + } + + PJ::DataSourceRuntimeHostView view() const { + return PJ::DataSourceRuntimeHostView(host_); + } + + std::vector& captured() { + return captured_; + } + int callCount() const { + return call_count_; + } + + private: + static bool notifyThunk(void* ctx, const PJ_available_topic_t* topics, uint64_t count, PJ_error_t* /*err*/) noexcept { + auto* self = static_cast(ctx); + self->call_count_++; + self->captured_.clear(); + for (uint64_t i = 0; i < count; ++i) { + const auto& t = topics[i]; + self->captured_.push_back( + CapturedTopic{ + std::string(t.topic_name.data, t.topic_name.size), + std::string(t.parser_encoding.data, t.parser_encoding.size), + std::string(t.type_name.data, t.type_name.size), + std::vector(t.schema.data, t.schema.data + t.schema.size), + }); + } + return true; + } + + PJ_data_source_runtime_host_vtable_t vtable_{}; + PJ_data_source_runtime_host_t host_{}; + std::vector captured_; + int call_count_ = 0; +}; + +TEST(NotifyAvailableTopicsTest, AdvertisedTopicsFlowThroughSlot) { + MockHost host; + const std::vector schema_bytes{0xDE, 0xAD, 0xBE, 0xEF}; + const std::vector topics{ + {"/camera/image", "protobuf", "foxglove.RawImage", + PJ::Span(schema_bytes.data(), schema_bytes.size())}, + {"/odom", "ros2msg", "nav_msgs/msg/Odometry", {}}, + }; + + auto status = host.view().notifyAvailableTopics(PJ::Span(topics.data(), topics.size())); + ASSERT_TRUE(status) << (status ? "" : status.error()); + ASSERT_EQ(host.captured().size(), 2U); + + EXPECT_EQ(host.captured()[0].topic_name, "/camera/image"); + EXPECT_EQ(host.captured()[0].parser_encoding, "protobuf"); + EXPECT_EQ(host.captured()[0].type_name, "foxglove.RawImage"); + EXPECT_EQ(host.captured()[0].schema, schema_bytes); + + EXPECT_EQ(host.captured()[1].topic_name, "/odom"); + EXPECT_EQ(host.captured()[1].parser_encoding, "ros2msg"); + EXPECT_TRUE(host.captured()[1].schema.empty()); +} + +TEST(NotifyAvailableTopicsTest, EmptySetIsValid) { + MockHost host; + auto status = host.view().notifyAvailableTopics(PJ::Span{}); + EXPECT_TRUE(status); + EXPECT_EQ(host.callCount(), 1); + EXPECT_TRUE(host.captured().empty()); +} + +TEST(NotifyAvailableTopicsTest, ReturnsErrorWhenSlotMissing) { + MockHost host; + host.dropNotifyAvailableTopics(); + const std::vector topics{{"/x", "protobuf", "T", {}}}; + + auto status = host.view().notifyAvailableTopics(PJ::Span(topics.data(), topics.size())); + EXPECT_FALSE(status); // explicit failure so a new plugin can fall back + EXPECT_EQ(host.callCount(), 0); +} + +// ---------- pj.topic_subscription.v1 extension (plugin side) ---------- + +// A minimal source exposing the per-topic subscription extension. The extension +// records the active-topic set the host pushes in, so the test can assert the +// full set round-tripped with the plugin instance as ctx. +class ExtensionSource : public PJ::DataSourcePluginBase { + public: + uint64_t capabilities() const override { + return PJ::kCapabilityContinuousStream | PJ::kCapabilityDelegatedIngest | PJ::kCapabilityPerTopicPause; + } + PJ::Status start() override { + return PJ::okStatus(); + } + void stop() override {} + PJ::DataSourceState currentState() const override { + return PJ::DataSourceState::kIdle; + } + const void* pluginExtension(std::string_view id) override { + if (id == PJ_TOPIC_SUBSCRIPTION_EXTENSION_V1) { + return &ext_; + } + return nullptr; + } + + std::vector active_topics; + + private: + static bool setActiveTopicsThunk( + void* ctx, const PJ_string_view_t* names, uint64_t count, PJ_error_t* /*err*/) noexcept { + auto* self = static_cast(ctx); + self->active_topics.assign(count, {}); + for (uint64_t i = 0; i < count; ++i) { + self->active_topics[i].assign(names[i].data, names[i].size); + } + return true; + } + + PJ_topic_subscription_v1_t ext_{sizeof(PJ_topic_subscription_v1_t), &ExtensionSource::setActiveTopicsThunk}; +}; + +TEST(TopicSubscriptionExtensionTest, SetActiveTopicsRoundTrips) { + ExtensionSource src; + EXPECT_TRUE(src.capabilities() & PJ::kCapabilityPerTopicPause); + + const auto* ext = + static_cast(src.pluginExtension(PJ_TOPIC_SUBSCRIPTION_EXTENSION_V1)); + ASSERT_NE(ext, nullptr); + ASSERT_GE(ext->struct_size, offsetof(PJ_topic_subscription_v1_t, set_active_topics) + sizeof(ext->set_active_topics)); + ASSERT_NE(ext->set_active_topics, nullptr); + + const PJ_string_view_t names[2] = {{"a", 1}, {"/topic/b", 8}}; + PJ_error_t err{}; + ASSERT_TRUE(ext->set_active_topics(&src, names, 2, &err)); + ASSERT_EQ(src.active_topics.size(), 2U); + EXPECT_EQ(src.active_topics[0], "a"); + EXPECT_EQ(src.active_topics[1], "/topic/b"); + + // Empty set pauses everything. + ASSERT_TRUE(ext->set_active_topics(&src, nullptr, 0, &err)); + EXPECT_TRUE(src.active_topics.empty()); +} + +TEST(TopicSubscriptionExtensionTest, UnknownExtensionIdIsNull) { + ExtensionSource src; + EXPECT_EQ(src.pluginExtension("pj.nonexistent.v1"), nullptr); +} + +// A source WITHOUT the extension (old plugin) — pluginExtension defaults to +// nullptr, so a host querying the id gets a clean no-op signal. +class PlainSource : public PJ::DataSourcePluginBase { + public: + uint64_t capabilities() const override { + return PJ::kCapabilityContinuousStream; + } + PJ::Status start() override { + return PJ::okStatus(); + } + void stop() override {} + PJ::DataSourceState currentState() const override { + return PJ::DataSourceState::kIdle; + } +}; + +TEST(TopicSubscriptionExtensionTest, PluginWithoutExtensionReturnsNull) { + PlainSource src; + EXPECT_FALSE(src.capabilities() & PJ::kCapabilityPerTopicPause); + EXPECT_EQ(src.pluginExtension(PJ_TOPIC_SUBSCRIPTION_EXTENSION_V1), nullptr); +} + +} // namespace diff --git a/pj_plugins/docs/ARCHITECTURE.md b/pj_plugins/docs/ARCHITECTURE.md index a8f874bb..7d45c09d 100644 --- a/pj_plugins/docs/ARCHITECTURE.md +++ b/pj_plugins/docs/ARCHITECTURE.md @@ -543,7 +543,7 @@ out_array, err)`: ## Builtin-object pipeline (PR #86) The v4 DataSource runtime host adds a tail slot `push_message` -(offset 96 in `PJ_data_source_runtime_host_vtable_t`) that takes a +(offset 88 in `PJ_data_source_runtime_host_vtable_t`) that takes a deferred byte-fetch callable instead of bytes: ```c @@ -586,3 +586,35 @@ forward compatibility is automatic. Concrete builtins live under `ImageAnnotations`, `FrameTransforms`); see `docs/builtin_type.md` for the type catalog and `docs/image_annotations_format.md` for the canonical annotation wire format. + +## Per-topic pause (demand-driven subscription) + +`kCapabilityPerTopicPause` (`1 << 6`) lets a multi-topic streaming source +expose its full topic universe cheaply while only transmitting data for +topics the host is actually displaying, instead of an all-or-nothing +whole-source pause. Two independent additions, both `struct_size`/ +`PJ_HAS_TAIL_SLOT`-gated so an old plugin or an old host degrades cleanly: + +- **Plugin → host advertise.** A second tail slot on the runtime host, + `notify_available_topics(ctx, topics, count, out_error)` (offset 96, + growing `sizeof(PJ_data_source_runtime_host_vtable_t)` 96 → 104), carrying + `PJ_available_topic_t{topic_name, parser_encoding, type_name, schema}` — + the same parser-identifying fields as `PJ_parser_binding_request_t` minus + `parser_config_json` (not yet known pre-subscription), so the host can + a-priori `classify_schema` every topic with no data flowing. Every call + carries the **full current set** (declarative, not a delta), entered from + the plugin's poll/stream thread. `DataSourceRuntimeHostView::notifyAvailableTopics` + returns an error on a host that lacks the slot, so a new plugin can detect + an old host and fall back to subscribing its preselected set at `start()`. +- **Host → plugin control.** The first stable (`pj..v1`) instance of + the §0a CLAP-style `get_plugin_extension` mechanism: + `"pj.topic_subscription.v1"` → `PJ_topic_subscription_v1_t::set_active_topics(ctx, + names, count, out_error)`. Also declarative-full-set, entered from the + host's GUI thread; the plugin's expected implementation is a mutex-protected + latest-wins slot drained on its own poll thread (not a command queue — the + host may call it faster than the plugin can act, and only the most recent + call matters). `DataSourceHandle::setActiveTopics` is a no-op when the + plugin does not expose the extension. + +See `docs/data-source-guide.md` → "Per-topic pause (demand-driven +subscription)" for the plugin-author walkthrough. diff --git a/pj_plugins/docs/REQUIREMENTS.md b/pj_plugins/docs/REQUIREMENTS.md index 1dca15ab..94f1fa0d 100644 --- a/pj_plugins/docs/REQUIREMENTS.md +++ b/pj_plugins/docs/REQUIREMENTS.md @@ -176,6 +176,7 @@ synchronize and call host methods only from callback context. | `kCapabilityDelegatedIngest` | Plugin pushes raw bytes for host-side parsing | | `kCapabilitySupportsPause` | pause()/resume() are implemented | | `kCapabilityHasDialog` | Plugin provides a configuration dialog | +| `kCapabilityPerTopicPause` | Host may pause/resume individual topics on a live connection (advertise via `notifyAvailableTopics` + `pj.topic_subscription.v1`) — see `docs/ARCHITECTURE.md` → "Per-topic pause" and `docs/data-source-guide.md` | ### Toolbox Capabilities diff --git a/pj_plugins/docs/data-source-guide.md b/pj_plugins/docs/data-source-guide.md index 75e932dc..961cab5f 100644 --- a/pj_plugins/docs/data-source-guide.md +++ b/pj_plugins/docs/data-source-guide.md @@ -468,6 +468,7 @@ Access via `runtimeHost()`. Use this for lifecycle coordination and diagnostics. | `isStopRequested()` | Check if the host wants you to stop. | | `ensureParserBinding(request)` | Bind a parser for delegated ingest (see below). | | `pushMessage(handle, timestamp, fetch_message_data)` | Push a message through a parser binding via a deferred fetcher callable; the host invokes it per the active ObjectIngestPolicy (eager/lazy). | +| `notifyAvailableTopics(topics)` | Advertise the full set of topics you *can* stream but have not subscribed, so the host lists and a-priori classifies them with no data flowing. See *Per-topic pause* below. | ## Optional Features @@ -492,6 +493,55 @@ PJ::Status resume() override { } ``` +### Per-topic pause (demand-driven subscription) + +For a multi-topic transport (a ROS/foxglove bridge, DDS, etc.) the pause above +is *whole-source*. A source that can subscribe/unsubscribe individual topics on +its live connection should instead let the host drive subscription per topic: +advertise everything cheaply, but only pull data for topics the host is actually +displaying. Declare it and the host will subscribe on demand: + +```cpp +uint64_t extraCapabilities() const override { + return PJ::kCapabilityDelegatedIngest | PJ::kCapabilityPerTopicPause; +} +``` + +Two halves: + +**1. Advertise (plugin → host).** From your poll/stream thread, tell the host +every topic you *could* stream (not just the subscribed ones). The host lists +them and classifies each a priori via `classify_schema` — no data flows yet. +Call again whenever the available set changes. + +```cpp +std::vector topics; // {topic_name, parser_encoding, type_name, schema} +for (const auto& ch : advertised_channels_) topics.push_back(toAvailableTopic(ch)); +runtimeHost().notifyAvailableTopics(topics); +``` + +`notifyAvailableTopics` returns an **error on an old host** that lacks the tail +slot — treat that as "advertisement unsupported" and fall back to your legacy +(subscribe-a-preselected-set-on-start) behavior, so a new plugin never strands +on an old host. + +**2. Subscribe on demand (host → plugin).** Expose the +`"pj.topic_subscription.v1"` extension. The host hands you the **full** desired +active-topic set (declarative); you diff it against your current subscriptions +and subscribe/unsubscribe on your connection. An empty set pauses everything. + +```cpp +const void* pluginExtension(std::string_view id) override { + if (id == PJ_TOPIC_SUBSCRIPTION_EXTENSION_V1) return &topic_subscription_ext_; + return nullptr; +} +// topic_subscription_ext_ is a PJ_topic_subscription_v1_t{sizeof(...), &setActiveTopicsThunk}; +// setActiveTopicsThunk receives the PLUGIN INSTANCE as ctx (not the extension +// pointer). It is entered on the host thread — marshal the desired set to your +// stream thread (e.g. a command queue drained in onPoll), do not touch your +// subscription state directly here. +``` + ### Periodic polling Override `onPoll()` (via `StreamSourceBase`) or `poll()` (via @@ -587,6 +637,7 @@ Any state --> failed | `kCapabilityDelegatedIngest` | `1 << 3` | Plugin pushes raw bytes for host-side parsing | | `kCapabilitySupportsPause` | `1 << 4` | pause()/resume() are implemented | | `kCapabilityHasDialog` | `1 << 5` | Plugin provides a configuration dialog | +| `kCapabilityPerTopicPause` | `1 << 6` | Host may pause/resume individual topics (advertise + `pj.topic_subscription.v1`) | Combine with bitwise OR. diff --git a/pj_plugins/include/pj_plugins/host/data_source_handle.hpp b/pj_plugins/include/pj_plugins/host/data_source_handle.hpp index feb7dcc7..348902c6 100644 --- a/pj_plugins/include/pj_plugins/host/data_source_handle.hpp +++ b/pj_plugins/include/pj_plugins/host/data_source_handle.hpp @@ -23,10 +23,12 @@ #pragma once #include +#include #include #include #include #include +#include #include "pj_base/data_source_protocol.h" #include "pj_base/expected.hpp" @@ -178,6 +180,32 @@ class DataSourceHandle { return vt_->get_plugin_extension(ctx_, sv); } + /// Push the full set of topics the source should keep actively subscribed + /// (the "pj.topic_subscription.v1" extension). The plugin diffs this against + /// its current subscriptions; topics not in the set are paused. Safe no-op + /// (returns success) when the plugin does not expose the extension — i.e. it + /// lacks PJ_DATA_SOURCE_CAPABILITY_PER_TOPIC_PAUSE or is an older `.so`. + /// The plugin instance ctx is passed (NOT the extension pointer). + [[nodiscard]] Status setActiveTopics(Span topic_names) const { + const auto* ext = + static_cast(getPluginExtension(PJ_TOPIC_SUBSCRIPTION_EXTENSION_V1)); + if (ext == nullptr || + ext->struct_size < offsetof(PJ_topic_subscription_v1_t, set_active_topics) + sizeof(ext->set_active_topics) || + ext->set_active_topics == nullptr) { + return okStatus(); + } + std::vector raw; + raw.reserve(topic_names.size()); + for (const std::string_view name : topic_names) { + raw.push_back(PJ_string_view_t{name.data(), name.size()}); + } + PJ_error_t err{}; + if (!ext->set_active_topics(ctx_, raw.data(), raw.size(), &err)) { + return unexpected(errorToString(err)); + } + return okStatus(); + } + [[nodiscard]] const PJ_data_source_vtable_t* vtable() const { return vt_; } diff --git a/pj_plugins/tests/data_source_library_test.cpp b/pj_plugins/tests/data_source_library_test.cpp index 7a3d41d2..5503ea54 100644 --- a/pj_plugins/tests/data_source_library_test.cpp +++ b/pj_plugins/tests/data_source_library_test.cpp @@ -9,6 +9,8 @@ #include #include #include +#include +#include #include "pj_base/plugin_data_api.h" #include "pj_base/sdk/service_traits.hpp" @@ -111,6 +113,7 @@ PJ_data_source_runtime_host_t makeRuntimeHost(bool with_encodings) { .show_message_box = rhShowMessageBox, .list_available_encodings = rhListEncodings, .push_message = rhPushMessage, + .notify_available_topics = nullptr, }; static const PJ_data_source_runtime_host_vtable_t no_enc_vt = { .protocol_version = 1, @@ -126,6 +129,7 @@ PJ_data_source_runtime_host_t makeRuntimeHost(bool with_encodings) { .show_message_box = rhShowMessageBox, .list_available_encodings = nullptr, .push_message = rhPushMessage, + .notify_available_topics = nullptr, }; return PJ_data_source_runtime_host_t{ .ctx = reinterpret_cast(0x2), @@ -155,6 +159,19 @@ TEST(DataSourceLibraryTest, LoadsSharedPluginAndDrivesInstance) { EXPECT_EQ(handle.currentState(), PJ::DataSourceState::kStopped); } +TEST(DataSourceLibraryTest, SetActiveTopicsIsNoOpWithoutExtension) { + auto library = PJ::DataSourceLibrary::load(PJ_MOCK_DATA_SOURCE_PLUGIN_PATH); + ASSERT_TRUE(library) << library.error(); + auto handle = library->createHandle(); + ASSERT_TRUE(handle.valid()); + + // The stock mock does not expose "pj.topic_subscription.v1"; the host wrapper + // must degrade to a successful no-op rather than erroring. + const std::vector names{"/a", "/b"}; + auto status = handle.setActiveTopics(PJ::Span(names.data(), names.size())); + EXPECT_TRUE(status) << (status ? "" : status.error()); +} + TEST(DataSourceLibraryTest, RejectsMissingRequiredVtableSlot) { auto library = PJ::DataSourceLibrary::load(PJ_MISSING_REQUIRED_SLOTS_PLUGIN_PATH); ASSERT_FALSE(library); diff --git a/pj_plugins/tests/file_source_integration_test.cpp b/pj_plugins/tests/file_source_integration_test.cpp index fca6ecdb..220a2afc 100644 --- a/pj_plugins/tests/file_source_integration_test.cpp +++ b/pj_plugins/tests/file_source_integration_test.cpp @@ -154,6 +154,7 @@ PJ_data_source_runtime_host_t makeRuntimeHost(RuntimeHostState* state) { .show_message_box = rhShowMessageBox, .list_available_encodings = nullptr, .push_message = nullptr, + .notify_available_topics = nullptr, }; return PJ_data_source_runtime_host_t{.ctx = state, .vtable = &vtable}; } diff --git a/recipe.yaml b/recipe.yaml index 1fa08def..6231db80 100644 --- a/recipe.yaml +++ b/recipe.yaml @@ -1,7 +1,7 @@ schema_version: 1 context: - version: "0.14.0" + version: "0.15.0" package: name: plotjuggler_sdk