From d338f4454fef0fca01636e402461cd1bd10c3f2c Mon Sep 17 00:00:00 2001 From: alvvm Date: Mon, 6 Jul 2026 12:31:07 +0200 Subject: [PATCH 1/3] feat(sdk): lazy per-topic subscription for DataSources (0.15.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Additive only — every existing plugin keeps working with no recompile: - PJ_DATA_SOURCE_CAPABILITY_LAZY_SUBSCRIPTION (1<<6) + kCapabilityLazySubscription. - Runtime-host tail slot set_advertised_topics: 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. DataSourceRuntimeHostView::setAdvertisedTopics returns a distinct 'host does not support lazy subscription' error for old-host fallback. - New extension pj.topic_subscription.v1 (data_source_topic_subscription.h): declarative set_active_topics, called on the poll thread strictly serialized with poll/start/stop. DataSourcePluginBase wires it to the new onActiveTopicsChanged() virtual iff the capability is declared; DataSourceHandle::topicSubscriptionExtension()/setActiveTopics() host-side. - MessageParser tail slot describe_schema_columns: flattened scalar-column manifest (JSON [{path,type}], parse() emission order) so hosts can pre-create catalog columns before the first sample. C++ ColumnSpec helper + base-class serialization; MessageParserHandle::describeSchemaColumns. - ABI layout sentinels extended (runtime host 96->104, parser vtable 88->96, extension struct pinned); tests: topic_subscription_extension_test, describe_schema_columns_test. 50/50 green in debug+ASAN and release. - Version 0.14.0 -> 0.15.0 (conanfile, CMakeLists, recipe.yaml), CHANGELOG, docs (ARCHITECTURE, REQUIREMENTS, data-source-guide, message-parser-guide). Note: local abi_check fails on clean main too (toolchain-vs-baseline noise); CI's abi_check is the authoritative additions-only gate. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01WUAom7BH7x39mRZFhBw79z --- CHANGELOG.md | 37 ++- CMakeLists.txt | 2 +- conanfile.py | 9 +- pj_base/CMakeLists.txt | 1 + .../include/pj_base/data_source_protocol.h | 37 +++ .../pj_base/data_source_topic_subscription.h | 72 +++++ .../include/pj_base/message_parser_protocol.h | 29 ++ .../pj_base/sdk/data_source_host_views.hpp | 39 +++ .../pj_base/sdk/data_source_plugin_base.hpp | 33 ++- .../sdk/detail/data_source_trampolines.hpp | 25 ++ pj_base/tests/abi_layout_sentinels_test.cpp | 21 +- .../topic_subscription_extension_test.cpp | 251 ++++++++++++++++++ pj_plugins/CMakeLists.txt | 9 + pj_plugins/docs/ARCHITECTURE.md | 41 +++ pj_plugins/docs/REQUIREMENTS.md | 1 + pj_plugins/docs/data-source-guide.md | 27 ++ pj_plugins/docs/message-parser-guide.md | 28 ++ .../pj_plugins/host/data_source_handle.hpp | 39 +++ .../pj_plugins/host/message_parser_handle.hpp | 22 ++ .../sdk/detail/message_parser_trampolines.hpp | 65 +++++ .../sdk/message_parser_plugin_base.hpp | 81 ++++++ pj_plugins/tests/data_source_library_test.cpp | 2 + .../tests/describe_schema_columns_test.cpp | 108 ++++++++ .../tests/file_source_integration_test.cpp | 1 + .../tests/missing_required_slots_plugin.cpp | 1 + recipe.yaml | 2 +- 26 files changed, 971 insertions(+), 12 deletions(-) create mode 100644 pj_base/include/pj_base/data_source_topic_subscription.h create mode 100644 pj_base/tests/topic_subscription_extension_test.cpp create mode 100644 pj_plugins/tests/describe_schema_columns_test.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b2ba244..d6440abd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,42 @@ 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` (base serializes the wire JSON), + `MessageParserHandle::describeSchemaColumns` host-side. + +## [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) 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..a0032da1 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) @@ -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" diff --git a/pj_base/CMakeLists.txt b/pj_base/CMakeLists.txt index c77c5fe3..56d04f3f 100644 --- a/pj_base/CMakeLists.txt +++ b/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 diff --git a/pj_base/include/pj_base/data_source_protocol.h b/pj_base/include/pj_base/data_source_protocol.h index 4c31f6d4..cb30d46f 100644 --- a/pj_base/include/pj_base/data_source_protocol.h +++ b/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. */ diff --git a/pj_base/include/pj_base/data_source_topic_subscription.h b/pj_base/include/pj_base/data_source_topic_subscription.h new file mode 100644 index 00000000..2cec2154 --- /dev/null +++ b/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 +#include + +#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 diff --git a/pj_base/include/pj_base/message_parser_protocol.h b/pj_base/include/pj_base/message_parser_protocol.h index 32694a7c..9080ed45 100644 --- a/pj_base/include/pj_base/message_parser_protocol.h +++ b/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. */ 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..4cdb9ebf 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 @@ -22,6 +22,7 @@ #include #include #include +#include #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; @@ -304,6 +306,43 @@ 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 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 raw; + raw.reserve(topics.size()); + for (const auto& t : topics) { + raw.push_back( + PJ_parser_binding_request_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), + .parser_config_json = sdk::toAbiString(t.parser_config_json), + }); + } + 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. 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..089399d0 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 @@ -33,8 +33,10 @@ #include #include #include +#include #include "pj_base/data_source_protocol.h" +#include "pj_base/data_source_topic_subscription.h" #include "pj_base/expected.hpp" #include "pj_base/plugin_abi_export.hpp" #include "pj_base/sdk/data_source_host_views.hpp" @@ -125,12 +127,35 @@ class DataSourcePluginBase { return PJ_borrowed_dialog_t{nullptr, nullptr}; } + /// The host's desired-active topic set changed (lazy subscription). + /// Called on the poll thread, strictly serialized with poll()/start()/ + /// stop(), only while the source is started. @p active_topics is the FULL + /// desired set by advertised topic name; reconcile subscriptions against + /// it (the view contents are valid only for the call). Only invoked when + /// the plugin declares kCapabilityLazySubscription. Default is a no-op. + virtual Status onActiveTopicsChanged(Span active_topics) { + (void)active_topics; + return okStatus(); + } + /// Return a pointer to a static plugin-exposed extension for @p id, or /// `nullptr` if unknown. CLAP-style reverse-direction capability query. - /// Default returns nullptr. The returned pointer must be valid for the - /// lifetime of this plugin instance. + /// The returned pointer must be valid for the lifetime of this plugin + /// instance. + /// + /// The default answers PJ_TOPIC_SUBSCRIPTION_EXTENSION_ID (routing to + /// onActiveTopicsChanged) when the plugin declares + /// kCapabilityLazySubscription. Subclasses that override MUST delegate + /// unknown ids to DataSourcePluginBase::pluginExtension to keep that + /// wiring. virtual const void* pluginExtension(std::string_view id) { - (void)id; + if (id == PJ_TOPIC_SUBSCRIPTION_EXTENSION_ID && (capabilities() & PJ_DATA_SOURCE_CAPABILITY_LAZY_SUBSCRIPTION)) { + static const PJ_topic_subscription_extension_t ext = { + sizeof(PJ_topic_subscription_extension_t), + trampoline_set_active_topics, + }; + return &ext; + } return nullptr; } @@ -220,6 +245,8 @@ class DataSourcePluginBase { static PJ_data_source_state_t trampoline_current_state(void* ctx) noexcept; static PJ_borrowed_dialog_t trampoline_get_dialog(void* ctx) noexcept; static const void* trampoline_get_plugin_extension(void* ctx, PJ_string_view_t id) noexcept; + static bool trampoline_set_active_topics( + void* ctx, const PJ_string_view_t* topic_names, uint64_t count, PJ_error_t* out_error) noexcept; }; } // namespace PJ diff --git a/pj_base/include/pj_base/sdk/detail/data_source_trampolines.hpp b/pj_base/include/pj_base/sdk/detail/data_source_trampolines.hpp index b0048d9a..cdea5939 100644 --- a/pj_base/include/pj_base/sdk/detail/data_source_trampolines.hpp +++ b/pj_base/include/pj_base/sdk/detail/data_source_trampolines.hpp @@ -206,4 +206,29 @@ inline const void* DataSourcePluginBase::trampoline_get_plugin_extension(void* c } } +inline bool DataSourcePluginBase::trampoline_set_active_topics( + void* ctx, const PJ_string_view_t* topic_names, uint64_t count, PJ_error_t* out_error) noexcept { + auto* self = static_cast(ctx); + try { + std::vector names; + names.reserve(count); + for (uint64_t i = 0; i < count; ++i) { + const auto& sv = topic_names[i]; + names.emplace_back(sv.data == nullptr ? std::string_view{} : std::string_view(sv.data, sv.size)); + } + auto status = self->onActiveTopicsChanged(Span(names.data(), names.size())); + if (!status) { + self->storeError(out_error, 1, "plugin", std::move(status).error()); + return false; + } + return true; + } catch (const std::exception& e) { + self->storeError(out_error, 1, "plugin", std::string("set_active_topics threw: ") + e.what()); + return false; + } catch (...) { + self->storeError(out_error, 1, "plugin", "unknown exception in set_active_topics"); + return false; + } +} + } // namespace PJ diff --git a/pj_base/tests/abi_layout_sentinels_test.cpp b/pj_base/tests/abi_layout_sentinels_test.cpp index 5be7e748..17a334e3 100644 --- a/pj_base/tests/abi_layout_sentinels_test.cpp +++ b/pj_base/tests/abi_layout_sentinels_test.cpp @@ -30,6 +30,7 @@ #include #include "pj_base/data_source_protocol.h" +#include "pj_base/data_source_topic_subscription.h" #include "pj_base/message_parser_protocol.h" #include "pj_base/plugin_data_api.h" #include "pj_base/toolbox_protocol.h" @@ -86,8 +87,11 @@ static_assert(offsetof(PJ_message_parser_vtable_t, struct_size) == 4, "v4 prefix static_assert(offsetof(PJ_message_parser_vtable_t, bind) == 32, "v4 bind slot pinned"); static_assert(offsetof(PJ_message_parser_vtable_t, parse) == 64, "v4 parse slot pinned"); static_assert(offsetof(PJ_message_parser_vtable_t, get_plugin_extension) == 72, "v4 last baseline slot pinned"); -// 80 baseline (v4.0) + 1 tail slot × 8 bytes = 88. -static_assert(sizeof(PJ_message_parser_vtable_t) == 88, "MessageParser vtable size (update deliberately on append)"); +static_assert(offsetof(PJ_message_parser_vtable_t, classify_schema) == 80, "classify_schema tail slot pinned"); +static_assert( + offsetof(PJ_message_parser_vtable_t, describe_schema_columns) == 88, "describe_schema_columns tail slot pinned"); +// 80 baseline (v4.0) + 2 tail slots × 8 bytes = 96. +static_assert(sizeof(PJ_message_parser_vtable_t) == 96, "MessageParser vtable size (update deliberately on append)"); static_assert(PJ_MESSAGE_PARSER_MIN_VTABLE_SIZE == 80, "MIN vtable size is pinned at v4.0 — NEVER INCREASE"); static_assert(PJ_MESSAGE_PARSER_MIN_VTABLE_SIZE <= sizeof(PJ_message_parser_vtable_t), "MIN must never exceed current"); @@ -153,7 +157,18 @@ 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, set_advertised_topics) == 96, + "set_advertised_topics tail slot pinned"); +static_assert( + sizeof(PJ_data_source_runtime_host_vtable_t) == 104, "Runtime host vtable size (update deliberately on append)"); + +// --- Topic subscription extension ("pj.topic_subscription.v1") --------------- +// Plugin-owned extension POD returned by get_plugin_extension. Appendable via +// struct_size; existing offsets pinned. +static_assert(offsetof(PJ_topic_subscription_extension_t, struct_size) == 0, "struct_size at offset 0"); +static_assert(offsetof(PJ_topic_subscription_extension_t, set_active_topics) == 8, "set_active_topics slot pinned"); +static_assert( + sizeof(PJ_topic_subscription_extension_t) == 16, "Topic subscription extension size (update deliberately)"); // --- 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/topic_subscription_extension_test.cpp b/pj_base/tests/topic_subscription_extension_test.cpp new file mode 100644 index 00000000..ec3da212 --- /dev/null +++ b/pj_base/tests/topic_subscription_extension_test.cpp @@ -0,0 +1,251 @@ +// Copyright 2026 Davide Faconti +// SPDX-License-Identifier: Apache-2.0 + +// Lazy-subscription SDK surface: +// - the "pj.topic_subscription.v1" extension auto-wired by +// DataSourcePluginBase iff kCapabilityLazySubscription is declared, +// routing set_active_topics -> onActiveTopicsChanged; +// - DataSourceRuntimeHostView::setAdvertisedTopics marshalling and its +// distinct old-host fallback error. + +#include + +#include +#include +#include + +#include "pj_base/data_source_protocol.h" +#include "pj_base/data_source_topic_subscription.h" +#include "pj_base/sdk/data_source_plugin_base.hpp" + +namespace { + +constexpr char kManifest[] = R"({"id":"mock-lazy","name":"Mock Lazy","version":"1.0.0"})"; + +// --------------------------------------------------------------------------- +// Plugin-side: extension exposure + onActiveTopicsChanged routing +// --------------------------------------------------------------------------- + +class MockLazySource : public PJ::DataSourcePluginBase { + public: + uint64_t capabilities() const override { + return PJ::kCapabilityContinuousStream | PJ::kCapabilityDelegatedIngest | PJ::kCapabilityLazySubscription; + } + PJ::Status start() override { + return PJ::okStatus(); + } + void stop() override {} + PJ::DataSourceState currentState() const override { + return PJ::DataSourceState::kRunning; + } + + PJ::Status onActiveTopicsChanged(PJ::Span active_topics) override { + if (throw_on_change) { + throw std::runtime_error("reconcile blew up"); + } + if (fail_on_change) { + return PJ::unexpected(std::string("reconcile refused")); + } + received.emplace_back(active_topics.begin(), active_topics.end()); + return PJ::okStatus(); + } + + std::vector> received; + bool throw_on_change = false; + bool fail_on_change = false; +}; + +class MockEagerSource : public MockLazySource { + public: + uint64_t capabilities() const override { + return PJ::kCapabilityContinuousStream | PJ::kCapabilityDelegatedIngest; + } +}; + +template +const PJ_data_source_vtable_t* vtableFor() { + return PJ::DataSourcePluginBase::vtableWithCreate( + []() noexcept -> void* { + try { + return new Plugin(); + } catch (...) { + return nullptr; + } + }, + kManifest); +} + +const PJ_topic_subscription_extension_t* queryExtension(const PJ_data_source_vtable_t* vt, void* ctx) { + PJ_string_view_t id{PJ_TOPIC_SUBSCRIPTION_EXTENSION_ID, sizeof(PJ_TOPIC_SUBSCRIPTION_EXTENSION_ID) - 1}; + return static_cast(vt->get_plugin_extension(ctx, id)); +} + +TEST(TopicSubscriptionExtensionTest, ExposedIffCapabilityDeclared) { + const auto* lazy_vt = vtableFor(); + void* lazy_ctx = lazy_vt->create(); + ASSERT_NE(lazy_ctx, nullptr); + const auto* ext = queryExtension(lazy_vt, lazy_ctx); + ASSERT_NE(ext, nullptr); + EXPECT_GE(ext->struct_size, sizeof(PJ_topic_subscription_extension_t)); + EXPECT_NE(ext->set_active_topics, nullptr); + lazy_vt->destroy(lazy_ctx); + + const auto* eager_vt = vtableFor(); + void* eager_ctx = eager_vt->create(); + ASSERT_NE(eager_ctx, nullptr); + EXPECT_EQ(queryExtension(eager_vt, eager_ctx), nullptr); + eager_vt->destroy(eager_ctx); +} + +TEST(TopicSubscriptionExtensionTest, SetActiveTopicsReachesVirtualWithAllNames) { + const auto* vt = vtableFor(); + void* ctx = vt->create(); + const auto* ext = queryExtension(vt, ctx); + ASSERT_NE(ext, nullptr); + + const PJ_string_view_t names[] = { + {"imu/data", 8}, + {"camera/points", 13}, + }; + PJ_error_t err{}; + ASSERT_TRUE(ext->set_active_topics(ctx, names, 2, &err)); + + auto* self = static_cast(ctx); + ASSERT_EQ(self->received.size(), 1u); + EXPECT_EQ(self->received[0], (std::vector{"imu/data", "camera/points"})); + + // Empty set is a legal desired state (all topics unsubscribed). + ASSERT_TRUE(ext->set_active_topics(ctx, nullptr, 0, &err)); + ASSERT_EQ(self->received.size(), 2u); + EXPECT_TRUE(self->received[1].empty()); + + vt->destroy(ctx); +} + +TEST(TopicSubscriptionExtensionTest, ErrorsAndExceptionsBecomeFalsePlusError) { + const auto* vt = vtableFor(); + void* ctx = vt->create(); + const auto* ext = queryExtension(vt, ctx); + ASSERT_NE(ext, nullptr); + auto* self = static_cast(ctx); + + const PJ_string_view_t names[] = {{"imu/data", 8}}; + + self->fail_on_change = true; + PJ_error_t err{}; + EXPECT_FALSE(ext->set_active_topics(ctx, names, 1, &err)); + EXPECT_NE(std::string_view(err.message).find("reconcile refused"), std::string_view::npos); + + self->fail_on_change = false; + self->throw_on_change = true; + PJ_error_t err2{}; + EXPECT_FALSE(ext->set_active_topics(ctx, names, 1, &err2)); + EXPECT_NE(std::string_view(err2.message).find("reconcile blew up"), std::string_view::npos); + + EXPECT_TRUE(self->received.empty()); + vt->destroy(ctx); +} + +// --------------------------------------------------------------------------- +// Host-view side: setAdvertisedTopics marshalling + old-host fallback +// --------------------------------------------------------------------------- + +struct CapturedTopic { + std::string topic_name; + std::string encoding; + std::string type_name; + std::vector schema; + std::string config_json; +}; + +struct AdvertiseCapture { + std::vector topics; + int calls = 0; +}; + +PJ_data_source_runtime_host_vtable_t makeAdvertiseVtable() { + PJ_data_source_runtime_host_vtable_t vt{}; + vt.protocol_version = 1; + vt.struct_size = sizeof(PJ_data_source_runtime_host_vtable_t); + vt.set_advertised_topics = [](void* ctx, const PJ_parser_binding_request_t* topics, uint64_t count, + PJ_error_t*) noexcept -> bool { + auto* capture = static_cast(ctx); + capture->calls += 1; + capture->topics.clear(); + for (uint64_t i = 0; i < count; ++i) { + const auto& t = topics[i]; + capture->topics.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), + std::string(t.parser_config_json.data, t.parser_config_json.size), + }); + } + return true; + }; + return vt; +} + +TEST(SetAdvertisedTopicsTest, MarshalsAllRequestFields) { + AdvertiseCapture capture; + auto vt = makeAdvertiseVtable(); + PJ::DataSourceRuntimeHostView view(PJ_data_source_runtime_host_t{&capture, &vt}); + + const uint8_t schema_bytes[] = {0xAA, 0xBB}; + const PJ::ParserBindingRequest requests[] = { + { + .topic_name = "imu/data", + .parser_encoding = "ros2msg", + .type_name = "sensor_msgs/msg/Imu", + .schema = PJ::Span(schema_bytes, sizeof(schema_bytes)), + .parser_config_json = R"({"a":1})", + }, + { + .topic_name = "camera/points", + .parser_encoding = "cdr", + .type_name = "sensor_msgs/msg/PointCloud2", + .schema = {}, + .parser_config_json = "", + }, + }; + + auto status = view.setAdvertisedTopics(PJ::Span(requests, 2)); + ASSERT_TRUE(status) << status.error(); + ASSERT_EQ(capture.calls, 1); + ASSERT_EQ(capture.topics.size(), 2u); + EXPECT_EQ(capture.topics[0].topic_name, "imu/data"); + EXPECT_EQ(capture.topics[0].encoding, "ros2msg"); + EXPECT_EQ(capture.topics[0].type_name, "sensor_msgs/msg/Imu"); + EXPECT_EQ(capture.topics[0].schema, (std::vector{0xAA, 0xBB})); + EXPECT_EQ(capture.topics[0].config_json, R"({"a":1})"); + EXPECT_EQ(capture.topics[1].topic_name, "camera/points"); + EXPECT_TRUE(capture.topics[1].schema.empty()); + + // Empty advertise (source has no topics yet) is valid. + auto empty_status = view.setAdvertisedTopics({}); + ASSERT_TRUE(empty_status) << empty_status.error(); + EXPECT_EQ(capture.calls, 2); + EXPECT_TRUE(capture.topics.empty()); +} + +TEST(SetAdvertisedTopicsTest, OldHostYieldsDistinctFallbackError) { + AdvertiseCapture capture; + auto vt = makeAdvertiseVtable(); + // A host built before the tail slot reports a smaller struct_size. + vt.struct_size = offsetof(PJ_data_source_runtime_host_vtable_t, set_advertised_topics); + PJ::DataSourceRuntimeHostView view(PJ_data_source_runtime_host_t{&capture, &vt}); + + auto status = view.setAdvertisedTopics({}); + ASSERT_FALSE(status); + EXPECT_EQ(status.error(), "host does not support lazy subscription"); + EXPECT_EQ(capture.calls, 0); + + PJ::DataSourceRuntimeHostView unbound; + auto unbound_status = unbound.setAdvertisedTopics({}); + ASSERT_FALSE(unbound_status); + EXPECT_EQ(unbound_status.error(), "runtime host is not bound"); +} + +} // namespace diff --git a/pj_plugins/CMakeLists.txt b/pj_plugins/CMakeLists.txt index 4c94110e..7b283e49 100644 --- a/pj_plugins/CMakeLists.txt +++ b/pj_plugins/CMakeLists.txt @@ -330,6 +330,15 @@ target_link_libraries(message_parser_library_test PRIVATE ) add_test(NAME message_parser_library_test COMMAND message_parser_library_test) +# Unit test: describe_schema_columns tail slot (ColumnSpec serialization, +# base-class default, host-side handle gating). +add_executable(describe_schema_columns_test tests/describe_schema_columns_test.cpp) +target_compile_options(describe_schema_columns_test PRIVATE ${PJ_WARNING_FLAGS}) +target_link_libraries(describe_schema_columns_test PRIVATE + pj_message_parser_host pj_base GTest::gtest_main +) +add_test(NAME describe_schema_columns_test COMMAND describe_schema_columns_test) + # Unit test: ObjectIngestPolicyResolver cascade rules. add_executable(object_ingest_policy_test tests/object_ingest_policy_test.cpp) target_compile_options(object_ingest_policy_test PRIVATE ${PJ_WARNING_FLAGS}) diff --git a/pj_plugins/docs/ARCHITECTURE.md b/pj_plugins/docs/ARCHITECTURE.md index a8f874bb..41cd64f6 100644 --- a/pj_plugins/docs/ARCHITECTURE.md +++ b/pj_plugins/docs/ARCHITECTURE.md @@ -586,3 +586,44 @@ 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. + +## Lazy per-topic subscription (0.15.0) + +Streaming sources that declare `PJ_DATA_SOURCE_CAPABILITY_LAZY_SUBSCRIPTION` +connect without subscribing to any topic and let the host drive per-topic +production from actual consumer demand. Three additive pieces cooperate: + +1. **Advertise (plugin → host).** The runtime-host tail slot + `set_advertised_topics(ctx, PJ_parser_binding_request_t*, count, err)` — + the FULL set of currently-available topics with parser metadata, + re-sent whenever the source's topic list changes. The host pre-binds + parsers so advertised topics appear in its catalog with zero data; + topics that disappear are marked unavailable but keep their ingested + data. C++: `DataSourceRuntimeHostView::setAdvertisedTopics`, which + returns the distinct error "host does not support lazy subscription" + on pre-0.15 hosts so plugins can fall back to eager selection. + +2. **Subscribe (host → plugin).** The `"pj.topic_subscription.v1"` + extension (`pj_base/data_source_topic_subscription.h`), queried via + `get_plugin_extension`. Its single verb `set_active_topics` is + DECLARATIVE: the host sends the full desired-active set (by advertised + topic name) and the plugin reconciles — subscribe what's newly + present, unsubscribe what's absent. The plugin may keep its own + always-on selection; the effective set is the union. Threading: called + on the poll thread, strictly serialized with poll/start/stop, only + between start and stop. The C++ SDK wires the extension automatically + in `DataSourcePluginBase::pluginExtension` (iff the capability is + declared) and routes it to the `onActiveTopicsChanged` virtual; + host-side access via `DataSourceHandle::topicSubscriptionExtension()` / + `setActiveTopics()`. + +3. **Column pre-materialization (parser).** The MessageParser tail slot + `describe_schema_columns(ctx, type_name, schema, out_json, err)` returns + the flattened scalar-column manifest (`[{"path","type"}]`, in exact + parse() emission order) so the host can create catalog columns before + the first sample arrives. Optional: parsers that cannot flatten a + schema without a payload (e.g. JSON) simply don't implement it and the + topic shows no fields until first data. C++: + `MessageParserPluginBase::describeSchemaColumns` returns + `std::vector` and the base serializes the wire JSON; + host-side `MessageParserHandle::describeSchemaColumns`. diff --git a/pj_plugins/docs/REQUIREMENTS.md b/pj_plugins/docs/REQUIREMENTS.md index 1dca15ab..7ca9e73f 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 | +| `kCapabilityLazySubscription` | Source advertises topics unsubscribed and subscribes per topic on host demand (`pj.topic_subscription.v1`) | ### Toolbox Capabilities diff --git a/pj_plugins/docs/data-source-guide.md b/pj_plugins/docs/data-source-guide.md index 75e932dc..b204355e 100644 --- a/pj_plugins/docs/data-source-guide.md +++ b/pj_plugins/docs/data-source-guide.md @@ -436,6 +436,33 @@ Key traits of `StreamSourceBase`: override `pause()`/`resume()` from `DataSourcePluginBase` directly and add `kCapabilitySupportsPause` to `extraCapabilities()`. +### Lazy per-topic subscription (0.15.0) + +A streaming source whose transport supports server/broker-side per-topic +filtering (DDS, MQTT, Foxglove WS, …) can let the host drive which topics +are actually subscribed, instead of freezing the selection in the config +dialog before `start()`: + +1. Add `kCapabilityLazySubscription` to `extraCapabilities()`. +2. In `onStart()` (and whenever the transport's topic list changes), call + `runtimeHost().setAdvertisedTopics(requests)` with the FULL set of + available topics — same `ParserBindingRequest` metadata you would pass + to `ensureParserBinding`, but WITHOUT subscribing anything. If this + returns the error "host does not support lazy subscription" (a pre-0.15 + host), fall back to your eager subscribe-selected behavior. +3. Override `onActiveTopicsChanged(Span)` and + reconcile: subscribe topics newly present in the desired set, + unsubscribe topics absent from it. Your own config-selected topics act + as an always-on floor — the effective subscribed set is the union. + +Threading contract: `onActiveTopicsChanged` runs on the same thread as +`poll()`, strictly serialized with `poll()`/`start()`/`stop()` and only +between start and stop — so it may touch the same state as `onPoll()` +without extra locking. The verb is declarative and idempotent: always +reconcile against the full set, never assume incremental pairs. Advertised +topics the host never activates must cost you (nearly) nothing — that is +the point of the feature. + ## Host Services Available to Plugins Two host bindings are provided before `start()` is called: diff --git a/pj_plugins/docs/message-parser-guide.md b/pj_plugins/docs/message-parser-guide.md index 0b7a8697..3a1cf040 100644 --- a/pj_plugins/docs/message-parser-guide.md +++ b/pj_plugins/docs/message-parser-guide.md @@ -254,6 +254,34 @@ The `type_name` is the encoding-specific message type (e.g. `schema` bytes are encoding-specific (e.g. ROS `.msg` definition text, Protobuf `FileDescriptorSet` binary). +### Column manifests (`describeSchemaColumns`, 0.15.0) + +Parsers whose schema fully determines the flattened field layout (ROS CDR, +Protobuf — anything schema-driven) can pre-describe the scalar columns +`parse()` will emit, so the host can create catalog columns for a topic +before its first sample arrives (used by lazy-subscription streaming +sources: all topics appear in the UI tree, data-less, ready to drag): + +```cpp +PJ::Expected> describeSchemaColumns( + std::string_view type_name, PJ::Span schema) const override { + std::vector cols; + cols.push_back({"pose.position.x", PJ::PrimitiveType::kFloat64}); + // ... one entry per leaf, in EXACTLY parse()'s emission order + return cols; +} +``` + +The contract is strict: **same paths, same order, same types as `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). Variable-length array fields may be omitted; their columns then +appear on first data. The base class serializes the specs into the C-ABI +wire JSON (`[{"path":"...","type":"float64"}, ...]`) behind the +`describe_schema_columns` tail slot. The default implementation reports +"unsupported", which is always safe — the topic simply shows no fields +until it produces a sample. + ### Configuration persistence Override `saveConfig()` / `loadConfig()` to support layout save/restore: 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..81db558b 100644 --- a/pj_plugins/include/pj_plugins/host/data_source_handle.hpp +++ b/pj_plugins/include/pj_plugins/host/data_source_handle.hpp @@ -27,8 +27,10 @@ #include #include #include +#include #include "pj_base/data_source_protocol.h" +#include "pj_base/data_source_topic_subscription.h" #include "pj_base/expected.hpp" #include "pj_base/sdk/data_source_host_views.hpp" @@ -178,6 +180,43 @@ class DataSourceHandle { return vt_->get_plugin_extension(ctx_, sv); } + /// The "pj.topic_subscription.v1" extension, or nullptr when the plugin + /// doesn't offer lazy subscription (no capability, older SDK, or a + /// malformed extension struct). Valid for the plugin-instance lifetime. + [[nodiscard]] const PJ_topic_subscription_extension_t* topicSubscriptionExtension() const { + const void* raw = getPluginExtension(PJ_TOPIC_SUBSCRIPTION_EXTENSION_ID); + if (raw == nullptr) { + return nullptr; + } + const auto* ext = static_cast(raw); + constexpr size_t kMinSize = + offsetof(PJ_topic_subscription_extension_t, set_active_topics) + sizeof(ext->set_active_topics); + if (ext->struct_size < kMinSize || ext->set_active_topics == nullptr) { + return nullptr; + } + return ext; + } + + /// Send the full desired-active topic set through the subscription + /// extension. MUST be called on the same thread as poll(), serialized + /// with poll()/start()/stop() — see data_source_topic_subscription.h. + [[nodiscard]] Status setActiveTopics(Span topic_names) { + const auto* ext = topicSubscriptionExtension(); + if (ext == nullptr) { + return unexpected(std::string("plugin does not support topic subscription")); + } + std::vector raw; + raw.reserve(topic_names.size()); + for (const auto& 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/include/pj_plugins/host/message_parser_handle.hpp b/pj_plugins/include/pj_plugins/host/message_parser_handle.hpp index 6aec4b03..6dd72f47 100644 --- a/pj_plugins/include/pj_plugins/host/message_parser_handle.hpp +++ b/pj_plugins/include/pj_plugins/host/message_parser_handle.hpp @@ -139,6 +139,28 @@ class MessageParserHandle { return vt_->get_plugin_extension(ctx_, sv); } + /// Flattened scalar-column manifest for @p type_name as wire JSON (see + /// describe_schema_columns in message_parser_protocol.h). Tail-slot + /// gated: an older plugin (or one that cannot pre-describe this schema) + /// yields an error — the host then simply skips column pre-creation. + /// Out-param instead of Expected for the same degenerate- + /// variant reason as saveConfig. @p out_json is only touched on success. + [[nodiscard]] Status describeSchemaColumns( + std::string_view type_name, Span schema, std::string& out_json) const { + if (!PJ_HAS_TAIL_SLOT(PJ_message_parser_vtable_t, vt_, describe_schema_columns)) { + return unexpected(std::string("parser does not expose describe_schema_columns")); + } + PJ_string_view_t tn{type_name.data(), type_name.size()}; + PJ_bytes_view_t sc{schema.data(), schema.size()}; + PJ_string_view_t out{}; + PJ_error_t err{}; + if (!vt_->describe_schema_columns(ctx_, tn, sc, &out, &err)) { + return unexpected(errorToString(err)); + } + out_json.assign(out.data == nullptr ? "" : out.data, out.size); + return okStatus(); + } + [[nodiscard]] const PJ_message_parser_vtable_t* vtable() const { return vt_; } diff --git a/pj_plugins/include/pj_plugins/sdk/detail/message_parser_trampolines.hpp b/pj_plugins/include/pj_plugins/sdk/detail/message_parser_trampolines.hpp index 1f33bc77..0c32a353 100644 --- a/pj_plugins/include/pj_plugins/sdk/detail/message_parser_trampolines.hpp +++ b/pj_plugins/include/pj_plugins/sdk/detail/message_parser_trampolines.hpp @@ -158,4 +158,69 @@ inline bool MessageParserPluginBase::trampoline_classify_schema( } } +inline bool MessageParserPluginBase::trampoline_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) noexcept { + auto* self = static_cast(ctx); + if (out_columns_json == nullptr) { + self->storeError(out_error, 2, "plugin", "describe_schema_columns called with null out_columns_json"); + return false; + } + try { + auto name_sv = type_name.data == nullptr ? std::string_view{} : std::string_view(type_name.data, type_name.size); + Span schema_span(schema.data, schema.size); + auto columns = self->describeSchemaColumns(name_sv, schema_span); + if (!columns) { + self->storeError(out_error, 1, "plugin", std::move(columns).error()); + return false; + } + // Hand-rolled serialization: paths are the only free-form strings, so a + // minimal JSON string escape keeps pj_plugins' SDK headers dependency-free + // for plugin authors. + std::string& json = self->columns_json_buf_; + json.clear(); + json.push_back('['); + bool first = true; + for (const auto& col : *columns) { + if (!first) { + json.push_back(','); + } + first = false; + json.append("{\"path\":\""); + for (const char c : col.path) { + switch (c) { + case '"': + json.append("\\\""); + break; + case '\\': + json.append("\\\\"); + break; + default: + if (static_cast(c) < 0x20) { + constexpr char kHex[] = "0123456789abcdef"; + json.append("\\u00"); + json.push_back(kHex[(c >> 4) & 0xF]); + json.push_back(kHex[c & 0xF]); + } else { + json.push_back(c); + } + } + } + json.append("\",\"type\":\""); + json.append(sdk::primitiveTypeJsonName(col.type)); + json.append("\"}"); + } + json.push_back(']'); + out_columns_json->data = json.data(); + out_columns_json->size = json.size(); + return true; + } catch (const std::exception& e) { + self->storeError(out_error, 1, "plugin", std::string("describe_schema_columns threw: ") + e.what()); + return false; + } catch (...) { + self->storeError(out_error, 1, "plugin", "unknown exception in describe_schema_columns"); + return false; + } +} + } // namespace PJ diff --git a/pj_plugins/include/pj_plugins/sdk/message_parser_plugin_base.hpp b/pj_plugins/include/pj_plugins/sdk/message_parser_plugin_base.hpp index 041bf591..0bfd3a47 100644 --- a/pj_plugins/include/pj_plugins/sdk/message_parser_plugin_base.hpp +++ b/pj_plugins/include/pj_plugins/sdk/message_parser_plugin_base.hpp @@ -17,6 +17,7 @@ #include #include #include +#include #include #include #include @@ -30,6 +31,7 @@ #include "pj_base/sdk/plugin_data_api.hpp" #include "pj_base/sdk/service_registry.hpp" #include "pj_base/sdk/service_traits.hpp" +#include "pj_base/type_tree.hpp" namespace PJ { namespace sdk { @@ -61,6 +63,63 @@ struct SchemaHandler { std::function(Timestamp, PayloadView)> parse_object; }; +/// One flattened scalar column a parser will emit for a schema, used by +/// describeSchemaColumns to pre-materialize catalog columns before any +/// sample exists (lazy-subscription sources). `path` is the flattened leaf +/// path exactly as parse() will emit it (e.g. "pose.position.x"). +struct ColumnSpec { + std::string path; + PrimitiveType type = PrimitiveType::kFloat64; +}; + +/// Wire name for a PrimitiveType in the describe_schema_columns JSON +/// manifest ("float64", "int32", ...). kUnspecified maps to "float64" — +/// the manifest describes concrete columns, never untyped ones. +[[nodiscard]] constexpr std::string_view primitiveTypeJsonName(PrimitiveType type) noexcept { + switch (type) { + case PrimitiveType::kFloat32: + return "float32"; + case PrimitiveType::kFloat64: + return "float64"; + case PrimitiveType::kInt8: + return "int8"; + case PrimitiveType::kInt16: + return "int16"; + case PrimitiveType::kInt32: + return "int32"; + case PrimitiveType::kInt64: + return "int64"; + case PrimitiveType::kUint8: + return "uint8"; + case PrimitiveType::kUint16: + return "uint16"; + case PrimitiveType::kUint32: + return "uint32"; + case PrimitiveType::kUint64: + return "uint64"; + case PrimitiveType::kBool: + return "bool"; + case PrimitiveType::kString: + return "string"; + case PrimitiveType::kUnspecified: + break; + } + return "float64"; +} + +/// Inverse of primitiveTypeJsonName. Unknown names yield std::nullopt. +[[nodiscard]] inline std::optional primitiveTypeFromJsonName(std::string_view name) noexcept { + for (const auto type : + {PrimitiveType::kFloat32, PrimitiveType::kFloat64, PrimitiveType::kInt8, PrimitiveType::kInt16, + PrimitiveType::kInt32, PrimitiveType::kInt64, PrimitiveType::kUint8, PrimitiveType::kUint16, + PrimitiveType::kUint32, PrimitiveType::kUint64, PrimitiveType::kBool, PrimitiveType::kString}) { + if (name == primitiveTypeJsonName(type)) { + return type; + } + } + return std::nullopt; +} + } // namespace sdk /** @@ -267,6 +326,23 @@ class MessageParserPluginBase { return h->parse_object(timestamp_ns, payload); } + /// Describe the flattened scalar-column manifest for @p type_name: + /// exactly the leaf columns parse() will emit, in emission order. Hosts + /// use it to pre-create catalog columns for topics with no samples yet + /// (lazy-subscription sources). Variable-length array fields MAY be + /// omitted (their columns appear on first data). MUST stay consistent + /// with parse() — same paths, same order, same types. + /// + /// Default: unsupported. Parsers that can flatten their schema without a + /// payload should override; the base serializes the returned specs into + /// the describe_schema_columns wire JSON. + virtual Expected> describeSchemaColumns( + std::string_view type_name, Span schema) const { + (void)type_name; + (void)schema; + return unexpected(std::string("parser does not support column manifests")); + } + /// Return a pointer to a static plugin-exposed extension for @p id, or /// nullptr if unknown. Default returns nullptr. virtual const void* pluginExtension(std::string_view id) { @@ -294,6 +370,7 @@ class MessageParserPluginBase { trampoline_parse, trampoline_get_plugin_extension, trampoline_classify_schema, + trampoline_describe_schema_columns, }; return &vt; } @@ -336,6 +413,7 @@ class MessageParserPluginBase { sdk::ParserWriteHostView write_host_view_{PJ_parser_write_host_t{}}; sdk::ParserObjectWriteHostView object_write_host_view_{}; std::string config_buf_; + std::string columns_json_buf_; // Schema handler table populated by the plugin via registerSchemaHandler(). std::unordered_map handlers_; @@ -356,6 +434,9 @@ class MessageParserPluginBase { static bool trampoline_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) noexcept; + static bool trampoline_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) noexcept; }; } // namespace PJ diff --git a/pj_plugins/tests/data_source_library_test.cpp b/pj_plugins/tests/data_source_library_test.cpp index 7a3d41d2..c46292f9 100644 --- a/pj_plugins/tests/data_source_library_test.cpp +++ b/pj_plugins/tests/data_source_library_test.cpp @@ -111,6 +111,7 @@ PJ_data_source_runtime_host_t makeRuntimeHost(bool with_encodings) { .show_message_box = rhShowMessageBox, .list_available_encodings = rhListEncodings, .push_message = rhPushMessage, + .set_advertised_topics = nullptr, }; static const PJ_data_source_runtime_host_vtable_t no_enc_vt = { .protocol_version = 1, @@ -126,6 +127,7 @@ PJ_data_source_runtime_host_t makeRuntimeHost(bool with_encodings) { .show_message_box = rhShowMessageBox, .list_available_encodings = nullptr, .push_message = rhPushMessage, + .set_advertised_topics = nullptr, }; return PJ_data_source_runtime_host_t{ .ctx = reinterpret_cast(0x2), diff --git a/pj_plugins/tests/describe_schema_columns_test.cpp b/pj_plugins/tests/describe_schema_columns_test.cpp new file mode 100644 index 00000000..6f6a0348 --- /dev/null +++ b/pj_plugins/tests/describe_schema_columns_test.cpp @@ -0,0 +1,108 @@ +// Copyright 2026 Davide Faconti +// SPDX-License-Identifier: Apache-2.0 + +// describe_schema_columns tail slot: SDK-side ColumnSpec -> wire-JSON +// serialization (incl. path escaping), the base-class "unsupported" default, +// host-side MessageParserHandle gating, and the type-name round-trip helpers. + +#include + +#include +#include +#include + +#include "pj_plugins/host/message_parser_handle.hpp" +#include "pj_plugins/sdk/message_parser_plugin_base.hpp" + +namespace { + +constexpr char kManifest[] = + R"({"id":"mock-manifest-parser","name":"Mock Manifest Parser","version":"1.0.0","encoding":["mock"]})"; + +class ManifestParser : public PJ::MessageParserPluginBase { + public: + PJ::Expected> describeSchemaColumns( + std::string_view type_name, PJ::Span schema) const override { + (void)schema; + if (type_name != "mock/Type") { + return PJ::unexpected(std::string("unknown type: ") + std::string(type_name)); + } + return std::vector{ + {"pose.position.x", PJ::PrimitiveType::kFloat64}, + {"status", PJ::PrimitiveType::kString}, + {"count", PJ::PrimitiveType::kUint32}, + {R"(weird"name)", PJ::PrimitiveType::kBool}, + }; + } +}; + +class DefaultParser : public PJ::MessageParserPluginBase {}; + +template +const PJ_message_parser_vtable_t* vtableFor() { + return PJ::MessageParserPluginBase::vtableWithCreate( + []() noexcept -> void* { + try { + return new Plugin(); + } catch (...) { + return nullptr; + } + }, + kManifest); +} + +TEST(DescribeSchemaColumnsTest, SerializesSpecsToWireJsonInOrder) { + PJ::MessageParserHandle handle(vtableFor()); + ASSERT_TRUE(handle.valid()); + + std::string json; + auto status = handle.describeSchemaColumns("mock/Type", {}, json); + ASSERT_TRUE(status) << status.error(); + EXPECT_EQ( + json, R"([{"path":"pose.position.x","type":"float64"},{"path":"status","type":"string"},)" + R"({"path":"count","type":"uint32"},{"path":"weird\"name","type":"bool"}])"); +} + +TEST(DescribeSchemaColumnsTest, PluginErrorPropagates) { + PJ::MessageParserHandle handle(vtableFor()); + std::string json = "untouched"; + auto status = handle.describeSchemaColumns("other/Type", {}, json); + ASSERT_FALSE(status); + EXPECT_NE(status.error().find("unknown type: other/Type"), std::string::npos); + EXPECT_EQ(json, "untouched"); +} + +TEST(DescribeSchemaColumnsTest, BaseDefaultIsUnsupported) { + PJ::MessageParserHandle handle(vtableFor()); + std::string json; + auto status = handle.describeSchemaColumns("mock/Type", {}, json); + ASSERT_FALSE(status); + EXPECT_NE(status.error().find("does not support column manifests"), std::string::npos); +} + +TEST(DescribeSchemaColumnsTest, OldPluginVtableIsGated) { + // A plugin compiled before the tail slot reports a smaller struct_size. + static PJ_message_parser_vtable_t old_vt = *vtableFor(); + old_vt.struct_size = offsetof(PJ_message_parser_vtable_t, describe_schema_columns); + PJ::MessageParserHandle handle(&old_vt); + std::string json; + auto status = handle.describeSchemaColumns("mock/Type", {}, json); + ASSERT_FALSE(status); + EXPECT_NE(status.error().find("does not expose describe_schema_columns"), std::string::npos); +} + +TEST(DescribeSchemaColumnsTest, PrimitiveTypeNamesRoundTrip) { + for (const auto type : + {PJ::PrimitiveType::kFloat32, PJ::PrimitiveType::kFloat64, PJ::PrimitiveType::kInt8, PJ::PrimitiveType::kInt16, + PJ::PrimitiveType::kInt32, PJ::PrimitiveType::kInt64, PJ::PrimitiveType::kUint8, PJ::PrimitiveType::kUint16, + PJ::PrimitiveType::kUint32, PJ::PrimitiveType::kUint64, PJ::PrimitiveType::kBool, PJ::PrimitiveType::kString}) { + const auto name = PJ::sdk::primitiveTypeJsonName(type); + const auto parsed = PJ::sdk::primitiveTypeFromJsonName(name); + ASSERT_TRUE(parsed.has_value()) << name; + EXPECT_EQ(*parsed, type) << name; + } + EXPECT_FALSE(PJ::sdk::primitiveTypeFromJsonName("double").has_value()); + EXPECT_EQ(PJ::sdk::primitiveTypeJsonName(PJ::PrimitiveType::kUnspecified), "float64"); +} + +} // namespace diff --git a/pj_plugins/tests/file_source_integration_test.cpp b/pj_plugins/tests/file_source_integration_test.cpp index fca6ecdb..285fbdd2 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, + .set_advertised_topics = nullptr, }; return PJ_data_source_runtime_host_t{.ctx = state, .vtable = &vtable}; } diff --git a/pj_plugins/tests/missing_required_slots_plugin.cpp b/pj_plugins/tests/missing_required_slots_plugin.cpp index aa42215c..a7068596 100644 --- a/pj_plugins/tests/missing_required_slots_plugin.cpp +++ b/pj_plugins/tests/missing_required_slots_plugin.cpp @@ -98,6 +98,7 @@ extern "C" PJ_MESSAGE_PARSER_EXPORT const PJ_message_parser_vtable_t* PJ_get_mes .parse = nullptr, .get_plugin_extension = extension, .classify_schema = nullptr, + .describe_schema_columns = nullptr, }; return &vt; } 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 From b0b73ba084b89d79317ef3cd09c64e7f9669fbae Mon Sep 17 00:00:00 2001 From: alvvm Date: Mon, 6 Jul 2026 16:40:51 +0200 Subject: [PATCH 2/3] =?UTF-8?q?docs,test(sdk):=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20capability=20table=20row,=20C++=20parser-path=20bre?= =?UTF-8?q?ak=20marker,=20DataSourceHandle=20subscription=20test?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01WUAom7BH7x39mRZFhBw79z --- CHANGELOG.md | 10 ++ pj_plugins/CMakeLists.txt | 9 ++ pj_plugins/docs/data-source-guide.md | 1 + .../data_source_handle_subscription_test.cpp | 100 ++++++++++++++++++ 4 files changed, 120 insertions(+) create mode 100644 pj_plugins/tests/data_source_handle_subscription_test.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index d6440abd..5cf1bedc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,16 @@ recompile (`abidiff` additions only): `std::vector` (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) diff --git a/pj_plugins/CMakeLists.txt b/pj_plugins/CMakeLists.txt index 7b283e49..01fda30b 100644 --- a/pj_plugins/CMakeLists.txt +++ b/pj_plugins/CMakeLists.txt @@ -339,6 +339,15 @@ target_link_libraries(describe_schema_columns_test PRIVATE ) add_test(NAME describe_schema_columns_test COMMAND describe_schema_columns_test) +# Unit test: DataSourceHandle wrapper over the topic_subscription extension +# (topicSubscriptionExtension() gating, setActiveTopics() marshalling/errors). +add_executable(data_source_handle_subscription_test tests/data_source_handle_subscription_test.cpp) +target_compile_options(data_source_handle_subscription_test PRIVATE ${PJ_WARNING_FLAGS}) +target_link_libraries(data_source_handle_subscription_test PRIVATE + pj_data_source_host pj_base GTest::gtest_main +) +add_test(NAME DataSourceHandleSubscriptionTest COMMAND data_source_handle_subscription_test) + # Unit test: ObjectIngestPolicyResolver cascade rules. add_executable(object_ingest_policy_test tests/object_ingest_policy_test.cpp) target_compile_options(object_ingest_policy_test PRIVATE ${PJ_WARNING_FLAGS}) diff --git a/pj_plugins/docs/data-source-guide.md b/pj_plugins/docs/data-source-guide.md index b204355e..be176238 100644 --- a/pj_plugins/docs/data-source-guide.md +++ b/pj_plugins/docs/data-source-guide.md @@ -614,6 +614,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 | +| `kCapabilityLazySubscription` | `1 << 6` | Source advertises topics unsubscribed and subscribes per topic on host demand (pj.topic_subscription.v1) | Combine with bitwise OR. diff --git a/pj_plugins/tests/data_source_handle_subscription_test.cpp b/pj_plugins/tests/data_source_handle_subscription_test.cpp new file mode 100644 index 00000000..ca315986 --- /dev/null +++ b/pj_plugins/tests/data_source_handle_subscription_test.cpp @@ -0,0 +1,100 @@ +// Copyright 2026 Davide Faconti +// SPDX-License-Identifier: Apache-2.0 + +// Host-side DataSourceHandle wrapper over the "pj.topic_subscription.v1" +// extension: topicSubscriptionExtension() gating and setActiveTopics() +// marshalling/error-path, exercised through the real handle (not the raw +// vtable, which pj_base/tests/topic_subscription_extension_test.cpp already +// covers at the SDK level). + +#include + +#include +#include + +#include "pj_base/sdk/data_source_plugin_base.hpp" +#include "pj_plugins/host/data_source_handle.hpp" + +namespace { + +constexpr char kManifest[] = R"({"id":"h","name":"h","version":"1.0.0"})"; + +class LazySource : public PJ::DataSourcePluginBase { + public: + uint64_t capabilities() const override { + return PJ::kCapabilityContinuousStream | PJ::kCapabilityLazySubscription; + } + PJ::Status start() override { + return PJ::okStatus(); + } + void stop() override {} + PJ::DataSourceState currentState() const override { + return PJ::DataSourceState::kRunning; + } + + PJ::Status onActiveTopicsChanged(PJ::Span active_topics) override { + received.emplace_back(active_topics.begin(), active_topics.end()); + return PJ::okStatus(); + } + + std::vector> received; +}; + +class EagerSource : 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::kRunning; + } +}; + +template +const PJ_data_source_vtable_t* vtableFor() { + return PJ::DataSourcePluginBase::vtableWithCreate( + []() noexcept -> void* { + try { + return new Plugin(); + } catch (...) { + return nullptr; + } + }, + kManifest); +} + +TEST(DataSourceHandleSubscriptionTest, LazyHandleExposesExtensionAndForwardsTopics) { + PJ::DataSourceHandle handle(vtableFor()); + ASSERT_TRUE(handle.valid()); + ASSERT_NE(handle.topicSubscriptionExtension(), nullptr); + + const std::vector names{"imu/data", "camera/points"}; + auto status = handle.setActiveTopics(PJ::Span(names)); + ASSERT_TRUE(status) << status.error(); + + auto* plugin = static_cast(handle.context()); + ASSERT_EQ(plugin->received.size(), 1u); + EXPECT_EQ(plugin->received[0], (std::vector{"imu/data", "camera/points"})); + + auto empty_status = handle.setActiveTopics({}); + ASSERT_TRUE(empty_status) << empty_status.error(); + ASSERT_EQ(plugin->received.size(), 2u); + EXPECT_TRUE(plugin->received[1].empty()); +} + +TEST(DataSourceHandleSubscriptionTest, EagerHandleHasNoExtensionAndRejectsSetActiveTopics) { + PJ::DataSourceHandle handle(vtableFor()); + ASSERT_TRUE(handle.valid()); + EXPECT_EQ(handle.topicSubscriptionExtension(), nullptr); + + const std::vector names{"imu/data"}; + auto status = handle.setActiveTopics(PJ::Span(names)); + ASSERT_FALSE(status); + EXPECT_NE(status.error().find("does not support topic subscription"), std::string::npos); +} + +} // namespace From 6b558aff4a4f51b88e2c6ace1d9fd76c60d43498 Mon Sep 17 00:00:00 2001 From: alvvm Date: Mon, 6 Jul 2026 16:47:16 +0200 Subject: [PATCH 3/3] refactor(sdk): share ParserBindingRequest ABI marshalling setAdvertisedTopics duplicated ensureParserBinding's field-for-field struct conversion; both now go through toAbiParserBindingRequest. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01WUAom7BH7x39mRZFhBw79z --- .../pj_base/sdk/data_source_host_views.hpp | 30 +++++++++---------- 1 file changed, 15 insertions(+), 15 deletions(-) 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 4cdb9ebf..e4157f1d 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 @@ -97,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) { @@ -199,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{}; @@ -327,14 +334,7 @@ class DataSourceRuntimeHostView { std::vector raw; raw.reserve(topics.size()); for (const auto& t : topics) { - raw.push_back( - PJ_parser_binding_request_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), - .parser_config_json = sdk::toAbiString(t.parser_config_json), - }); + raw.push_back(toAbiParserBindingRequest(t)); } PJ_error_t err{}; if (!host_.vtable->set_advertised_topics(host_.ctx, raw.data(), raw.size(), &err)) {