From df673f08a0b3e4fc6dc8c00b74b85632091e4127 Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Sun, 25 Oct 2026 13:40:37 +0100 Subject: [PATCH 1/7] refactor(gateway): the core registers the built-in cyclic-subscription samplers register_builtin_samplers() registers data, faults, configurations, logs and updates from GatewayServices, so a program that hosts the core gets samplers without the ROS node. The data sampler reads through the same provider resolution as GET /{entity}/data/{item}, so plugin-served data can be subscribed to. --- scripts/check_cores_without_ros/main.cpp | 309 +++++++++++++++++- .../core/builtin_samplers.hpp | 31 ++ .../core/data/data_item_reader.hpp | 51 +++ .../src/core/builtin_samplers.cpp | 196 +++++++++++ .../src/core/data/data_item_reader.cpp | 93 ++++++ src/ros2_medkit_gateway/src/gateway_node.cpp | 146 +-------- .../src/http/handlers/data_handlers.cpp | 63 +--- 7 files changed, 682 insertions(+), 207 deletions(-) create mode 100644 src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/builtin_samplers.hpp create mode 100644 src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/data/data_item_reader.hpp create mode 100644 src/ros2_medkit_gateway/src/core/builtin_samplers.cpp create mode 100644 src/ros2_medkit_gateway/src/core/data/data_item_reader.cpp diff --git a/scripts/check_cores_without_ros/main.cpp b/scripts/check_cores_without_ros/main.cpp index 6a8cf8d90..1b2f45752 100644 --- a/scripts/check_cores_without_ros/main.cpp +++ b/scripts/check_cores_without_ros/main.cpp @@ -17,12 +17,18 @@ #include #include +#include +#include #include #include #include #include #include +#include +#include #include +#include +#include #include #include #include @@ -40,9 +46,12 @@ #include "ros2_medkit_fault_manager/sqlite_fault_storage.hpp" #include "ros2_medkit_fault_reporter/core/reporter_core.hpp" #include "ros2_medkit_gateway/core/auth/auth_config.hpp" +#include "ros2_medkit_gateway/core/builtin_samplers.hpp" #include "ros2_medkit_gateway/core/config.hpp" +#include "ros2_medkit_gateway/core/data/topic_data_provider.hpp" #include "ros2_medkit_gateway/core/discovery/entity_discovery.hpp" #include "ros2_medkit_gateway/core/gateway_services.hpp" +#include "ros2_medkit_gateway/core/http/handlers/sse_transport_provider.hpp" #include "ros2_medkit_gateway/core/http/rate_limiter.hpp" #include "ros2_medkit_gateway/core/http/rest_server.hpp" #include "ros2_medkit_gateway/core/http/sse_client_tracker.hpp" @@ -52,6 +61,9 @@ #include "ros2_medkit_gateway/core/managers/operation_manager.hpp" #include "ros2_medkit_gateway/core/managers/subscription_manager.hpp" #include "ros2_medkit_gateway/core/models/thread_safe_entity_cache.hpp" +#include "ros2_medkit_gateway/core/plugins/gateway_plugin.hpp" +#include "ros2_medkit_gateway/core/plugins/plugin_manager.hpp" +#include "ros2_medkit_gateway/core/providers/data_provider.hpp" #include "ros2_medkit_gateway/core/resource_sampler.hpp" #include "ros2_medkit_gateway/core/subscription_transport.hpp" #include "ros2_medkit_gateway/core/transports/action_transport.hpp" @@ -118,10 +130,17 @@ class NoFaultTransport : public gw::FaultServiceTransport { const std::string & /*description*/, const std::string & /*source_id*/) override { return none(); } + // Once set, every list answers one fault whose occurrence_count grows with each call. gw::FaultResult list_faults(const std::string & /*source_id*/, bool /*include_prefailed*/, bool /*include_confirmed*/, bool /*include_cleared*/, bool /*include_healed*/, bool /*include_muted*/, bool /*include_clusters*/) override { - return none(); + if (!reports_faults.load()) { + return none(); + } + gw::json fault = {{"fault_code", "CHECK_FAULT"}, + {"occurrence_count", ++list_calls}, + {"reporting_sources", gw::json::array({"hosted_app"})}}; + return {true, {{"faults", gw::json::array({fault})}}, {}, gw::FaultFailure::Declined}; } gw::FaultWithEnvJsonResult get_fault_with_env(const std::string & /*fault_code*/, const std::string & /*source_id*/) override { @@ -148,6 +167,8 @@ class NoFaultTransport : public gw::FaultServiceTransport { bool is_available() const override { return false; } + std::atomic reports_faults{false}; + std::atomic list_calls{0}; private: static gw::FaultResult none() { @@ -155,6 +176,90 @@ class NoFaultTransport : public gw::FaultServiceTransport { } }; +// Answers every sample with a value that grows by one per call. +class CountingTopicProvider : public gw::TopicDataProvider { + public: + tl::expected sample(const std::string & topic, + std::chrono::milliseconds /*timeout*/) override { + gw::TopicSampleResult out; + out.topic_name = topic; + out.message_type = "check/msg/Counter"; + out.data = gw::json{{"value", ++samples}}; + out.has_data = true; + return out; + } + tl::expected, gw::ErrorInfo> + sample_parallel(const std::vector & topics, std::chrono::milliseconds timeout) override { + std::vector out; + out.reserve(topics.size()); + for (const auto & topic : topics) { + out.push_back(*sample(topic, timeout)); + } + return out; + } + std::optional get_topic_info(const std::string & /*topic*/) override { + return std::nullopt; + } + bool has_publishers(const std::string & /*topic*/) override { + return true; + } + std::vector discover(const std::string & /*namespace_prefix*/) override { + return {}; + } + std::vector discover_all() override { + return {}; + } + std::map build_component_topic_map() override { + return {}; + } + gw::ComponentTopics get_component_topics(const std::string & /*component_fqn*/) override { + return {}; + } + gw::TopicDiscoveryResult discover_topics_by_namespace() override { + return {}; + } + std::set discover_topic_namespaces() override { + return {}; + } + gw::ComponentTopics get_topics_for_namespace(const std::string & /*ns_prefix*/) override { + return {}; + } + std::vector get_topic_publishers(const std::string & /*topic*/) override { + return {}; + } + std::vector get_topic_subscribers(const std::string & /*topic*/) override { + return {}; + } + gw::TopicConnection get_topic_connection(const std::string & /*topic*/) override { + return {}; + } + std::atomic samples{0}; +}; + +// A plugin that serves the data of the entities it owns; every read counts up. +class CheckDataPlugin : public gw::GatewayPlugin, public gw::DataProvider { + public: + std::string name() const override { + return "check_plugin"; + } + void configure(const gw::json & /*config*/) override { + } + tl::expected + list_data(const std::string & /*entity_id*/) override { + return gw::dto::DataListResult{{{"items", gw::json::array()}}}; + } + tl::expected read_data(const std::string & /*entity_id*/, + const std::string & resource_name) override { + return gw::dto::DataValue{{{"source", "plugin"}, {"name", resource_name}, {"value", ++reads}}}; + } + tl::expected write_data(const std::string & /*entity_id*/, + const std::string & /*resource_name*/, + const gw::json & /*value*/) override { + return tl::make_unexpected(gw::DataProviderErrorInfo{gw::DataProviderError::ReadOnly, "read only", 405}); + } + std::atomic reads{0}; +}; + class NoParameterTransport : public gw::ParameterTransport { public: bool is_self_node(const std::string & /*node_name*/) const override { @@ -302,7 +407,8 @@ class NoTypeSchemas : public gw::TypeSchemaProvider { class MinimalServices : public gw::GatewayServices { public: MinimalServices() - : fault_(std::make_shared()) + : fault_transport_(std::make_shared()) + , fault_(fault_transport_) , config_(std::make_shared()) , data_(std::make_shared()) , operations_(std::make_shared(), std::make_shared(), nullptr) @@ -345,7 +451,7 @@ class MinimalServices : public gw::GatewayServices { return nullptr; } gw::PluginManager * plugin_manager() override { - return nullptr; + return plugins_.get(); } gw::ScriptManager * script_manager() override { return nullptr; @@ -378,10 +484,28 @@ class MinimalServices : public gw::GatewayServices { return nullptr; } + gw::ThreadSafeEntityCache & mutable_cache() { + return cache_; + } + NoFaultTransport & fault_transport() { + return *fault_transport_; + } + // Plugins exist only after this call; the plugin owns the entity ids it is given. + CheckDataPlugin & add_data_plugin(const std::vector & owned_entities) { + plugins_ = std::make_unique(); + auto plugin = std::make_unique(); + auto & ref = *plugin; + plugins_->add_plugin(std::move(plugin)); + plugins_->register_entity_ownership("check_plugin", owned_entities); + return ref; + } + private: gw::ThreadSafeEntityCache cache_; EmptyDiscovery discovery_; NoTypeSchemas schemas_; + std::shared_ptr fault_transport_; + std::unique_ptr plugins_; gw::FaultManager fault_; gw::ConfigurationManager config_; gw::DataAccessManager data_; @@ -411,6 +535,111 @@ bool check(bool ok, const char * what) { std::printf("%s: %s\n", ok ? "ok " : "FAIL", what); return ok; } + +// A REST server on an ephemeral port, stopped and joined on scope exit. +class RunningServer { + public: + explicit RunningServer(MinimalServices & services) : port_(free_port()) { + gw::RestConfig rest; + rest.http_thread_pool_size = 6; + server_ = std::make_unique(services, nullptr, "127.0.0.1", port_, gw::CorsConfig{}, + gw::AuthConfig{}, gw::RateLimitConfig{}, rest); + thread_ = std::thread([this] { + server_->start(); + }); + for (int i = 0; i < 100 && !server_->is_running(); ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + } + } + ~RunningServer() { + server_->stop(); + thread_.join(); + } + RunningServer(const RunningServer &) = delete; + RunningServer & operator=(const RunningServer &) = delete; + RunningServer(RunningServer &&) = delete; + RunningServer & operator=(RunningServer &&) = delete; + int port() const { + return port_; + } + + private: + int port_; + std::unique_ptr server_; + std::thread thread_; +}; + +// Reads SSE frames from `path` until `want` of them arrived, the stream ended or the read timed out. +std::vector read_events(httplib::Client & client, const std::string & path, size_t want) { + std::vector events; + std::string buffer; + client.Get( + path, httplib::Headers{}, + [](const httplib::Response &) { + return true; + }, + [&](const char * data, size_t len) { + buffer.append(data, len); + for (size_t end = buffer.find("\n\n"); end != std::string::npos; end = buffer.find("\n\n")) { + const std::string frame = buffer.substr(0, end); + buffer.erase(0, end + 2); + if (frame.rfind("data: ", 0) == 0) { + events.push_back(nlohmann::json::parse(frame.substr(6), nullptr, false)); + } + } + return events.size() < want; + }); + return events; +} + +httplib::Client make_client(int port) { + httplib::Client client("127.0.0.1", port); + client.set_connection_timeout(5); + client.set_read_timeout(5); + return client; +} + +// Creates a fast cyclic subscription on `resource`; returns {subscription id, events path}. +std::pair subscribe(httplib::Client & client, const std::string & resource, + const std::string & app = "hosted_app") { + const nlohmann::json body = {{"resource", resource}, {"interval", "fast"}, {"duration", 60}}; + auto res = client.Post("/api/v1/apps/" + app + "/cyclic-subscriptions", body.dump(), "application/json"); + if (!res || res->status != 201) { + std::printf("subscribe %s -> %d %s\n", resource.c_str(), res ? res->status : 0, res ? res->body.c_str() : ""); + return {}; + } + const auto json = nlohmann::json::parse(res->body, nullptr, false); + return {json.value("id", ""), json.value("event_source", "")}; +} + +// True when the first two sampled payloads exist and `field` differs between them. +bool values_change(const std::vector & events, + const std::function & pick) { + if (events.size() < 2 || !events[0].contains("payload") || !events[1].contains("payload")) { + return false; + } + const auto first = pick(events[0]["payload"]); + const auto second = pick(events[1]["payload"]); + return !first.is_null() && !second.is_null() && first != second; +} + +gw::App hosted_app() { + gw::App app; + app.id = "hosted_app"; + app.name = "Hosted app"; + app.external = true; + return app; +} + +void add_sse_transport(MinimalServices & services) { + services.transport_registry()->register_transport( + std::make_unique(*services.subscription_manager(), services.sse_client_tracker())); + services.subscription_manager()->set_on_removed([&services](const gw::CyclicSubscriptionInfo & info) { + if (auto * transport = services.transport_registry()->get_transport(info.protocol)) { + transport->stop(info.id); + } + }); +} } // namespace static int run() { @@ -709,6 +938,80 @@ static int run() { server_thread.join(); } + // A host that registers the samplers and a transport gets cyclic subscriptions. + { + MinimalServices services; + CountingTopicProvider topics; + services.data_access_manager()->set_topic_data_provider(&topics); + services.fault_transport().reports_faults = true; + auto plugin_app = hosted_app(); + plugin_app.id = "plugin_app"; + services.mutable_cache().update_apps({hosted_app(), plugin_app}); + services.add_data_plugin({"plugin_app"}); + gw::register_builtin_samplers(services); + add_sse_transport(services); + { + RunningServer running(services); + auto client = make_client(running.port()); + + // Cyclic subscription on a data resource. + const auto data_sub = subscribe(client, "/api/v1/apps/hosted_app/data/temperature"); + ok &= check(!data_sub.second.empty(), "a cyclic subscription on a data resource is created"); + const auto data_events = + data_sub.second.empty() ? std::vector{} : read_events(client, data_sub.second, 2); + ok &= check(data_events.size() == 2, "the data subscription streams two events"); + ok &= check(values_change(data_events, + [](const nlohmann::json & p) { + return p.contains("data") ? p["data"].value("value", nlohmann::json()) + : nlohmann::json(); + }), + "the data value differs between the two events"); + ok &= check(!data_events.empty() && data_events[0].contains("payload") && + data_events[0]["payload"].value("id", "") == "/temperature", + "the data payload names the resource"); + if (!data_sub.first.empty()) { + client.Delete("/api/v1/apps/hosted_app/cyclic-subscriptions/" + data_sub.first); + } + + // Cyclic subscription on a resource a plugin serves. + const auto plugin_sub = subscribe(client, "/api/v1/apps/plugin_app/data/speed", "plugin_app"); + ok &= check(!plugin_sub.second.empty(), "a cyclic subscription on a plugin-served data resource is created"); + const auto plugin_events = + plugin_sub.second.empty() ? std::vector{} : read_events(client, plugin_sub.second, 2); + ok &= check(plugin_events.size() == 2, "the plugin-data subscription streams two events"); + ok &= check(!plugin_events.empty() && plugin_events[0].contains("payload") && + plugin_events[0]["payload"].value("source", "") == "plugin" && + plugin_events[0]["payload"].value("name", "") == "speed", + "the plugin-data payload comes from the plugin's provider"); + ok &= check(values_change(plugin_events, + [](const nlohmann::json & p) { + return p.value("value", nlohmann::json()); + }), + "the plugin value differs between the two events"); + if (!plugin_sub.first.empty()) { + client.Delete("/api/v1/apps/plugin_app/cyclic-subscriptions/" + plugin_sub.first); + } + + // Cyclic subscription on faults. + const auto fault_sub = subscribe(client, "/api/v1/apps/hosted_app/faults"); + ok &= check(!fault_sub.second.empty(), "a cyclic subscription on faults is created"); + const auto fault_events = + fault_sub.second.empty() ? std::vector{} : read_events(client, fault_sub.second, 2); + ok &= check(fault_events.size() == 2, "the faults subscription streams two events"); + ok &= check(values_change(fault_events, + [](const nlohmann::json & p) { + return p.contains("faults") && p["faults"].is_array() && !p["faults"].empty() + ? p["faults"][0].value("occurrence_count", nlohmann::json()) + : nlohmann::json(); + }), + "the fault occurrence_count differs between the two events"); + if (!fault_sub.first.empty()) { + client.Delete("/api/v1/apps/hosted_app/cyclic-subscriptions/" + fault_sub.first); + } + } + services.transport_registry()->shutdown_all(*services.subscription_manager()); + } + return ok ? 0 : 1; } diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/builtin_samplers.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/builtin_samplers.hpp new file mode 100644 index 000000000..087d51d99 --- /dev/null +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/builtin_samplers.hpp @@ -0,0 +1,31 @@ +// Copyright 2026 bburda +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#pragma once + +namespace ros2_medkit_gateway { + +class GatewayServices; + +/// Registers the built-in cyclic-subscription samplers on `services.sampler_registry()`: +/// `data`, `faults`, `configurations` and `logs`, plus `updates` when `services.update_manager()` +/// is set. Each sampler reads through the managers and helpers of the matching GET route, so a +/// resource the route serves can be subscribed to, whoever provides it. A collection the host has +/// no manager for answers with the error text of its GET route. +/// +/// `services` must outlive every sampler: the samplers keep a reference and read the manager +/// getters on each call. Call it once, before the first subscription is created. +void register_builtin_samplers(GatewayServices & services); + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/data/data_item_reader.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/data/data_item_reader.hpp new file mode 100644 index 000000000..6d4236327 --- /dev/null +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/data/data_item_reader.hpp @@ -0,0 +1,51 @@ +// Copyright 2026 bburda +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#pragma once + +#include + +#include +#include + +#include "ros2_medkit_gateway/core/data/data_types.hpp" +#include "ros2_medkit_gateway/core/models/error_info.hpp" +#include "ros2_medkit_gateway/core/providers/data_provider.hpp" +#include "ros2_medkit_gateway/dto/data.hpp" + +namespace ros2_medkit_gateway { + +class DataAccessManager; +class PluginManager; + +/// An `x-medkit-plugin-error` with the status clamped to [400, 599] and the message cut to 512 characters. +ErrorInfo make_plugin_error(int http_status, const std::string & message, nlohmann::json extra_params = {}); + +/// The plugin error for a failed `DataProvider` call. +ErrorInfo make_provider_error(const DataProviderErrorInfo & info, const std::string & entity_id); + +/// A data item is addressed by its full topic path. The leading slash is optional in a route segment. +std::string to_full_topic_path(const std::string & topic_name); + +/// Reads one data item of a plugin-owned entity: through its `DataProvider`, else through its +/// `x-plc-data` route. `GET /{entity}/data/{item}` and the cyclic-subscription sampler both use it. +tl::expected read_plugin_data_item(PluginManager * plugins, const std::string & entity_id, + const std::string & topic_name); + +/// Samples one topic through the `TopicDataProvider` of `data`. Answers 503 when none is configured. +/// `GET /{entity}/data/{item}` and the cyclic-subscription sampler both use it. +tl::expected sample_data_topic(DataAccessManager & data, + const std::string & full_topic_path); + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/core/builtin_samplers.cpp b/src/ros2_medkit_gateway/src/core/builtin_samplers.cpp new file mode 100644 index 000000000..71163b93e --- /dev/null +++ b/src/ros2_medkit_gateway/src/core/builtin_samplers.cpp @@ -0,0 +1,196 @@ +// Copyright 2026 bburda +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "ros2_medkit_gateway/core/builtin_samplers.hpp" + +#include +#include + +#include +#include + +#include "ros2_medkit_gateway/core/data/data_item_reader.hpp" +#include "ros2_medkit_gateway/core/faults/fault_scope.hpp" +#include "ros2_medkit_gateway/core/gateway_services.hpp" +#include "ros2_medkit_gateway/core/logs/log_scope.hpp" +#include "ros2_medkit_gateway/core/managers/configuration_manager.hpp" +#include "ros2_medkit_gateway/core/managers/data_access_manager.hpp" +#include "ros2_medkit_gateway/core/managers/fault_manager.hpp" +#include "ros2_medkit_gateway/core/managers/log_manager.hpp" +#include "ros2_medkit_gateway/core/managers/update_manager.hpp" +#include "ros2_medkit_gateway/core/models/thread_safe_entity_cache.hpp" +#include "ros2_medkit_gateway/core/plugins/plugin_manager.hpp" +#include "ros2_medkit_gateway/core/providers/update_types.hpp" +#include "ros2_medkit_gateway/core/resource_sampler.hpp" + +namespace ros2_medkit_gateway { + +namespace { + +using Sampled = tl::expected; + +Sampled sample_data(GatewayServices & services, const std::string & entity_id, const std::string & resource_path) { + // A plugin-owned entity is read through its provider, as GET /{entity}/data/{item} does. + auto * plugins = services.plugin_manager(); + if (plugins != nullptr && plugins->get_entity_owner(entity_id)) { + const std::string item_name = + (!resource_path.empty() && resource_path.front() == '/') ? resource_path.substr(1) : resource_path; + auto item = read_plugin_data_item(plugins, entity_id, item_name); + if (!item) { + return tl::make_unexpected(item.error().message); + } + return std::move(item->content); + } + auto sample = sample_data_topic(*services.data_access_manager(), to_full_topic_path(resource_path)); + if (!sample) { + return tl::make_unexpected(std::string{"Topic sample failed: "} + sample.error().message); + } + if (sample->has_data && sample->data.has_value()) { + nlohmann::json payload; + payload["id"] = resource_path; + payload["data"] = *sample->data; + return payload; + } + return tl::make_unexpected("Topic data not available: " + resource_path); +} + +Sampled sample_faults(GatewayServices & services, const std::string & entity_id) { + auto * fault_mgr = services.fault_manager(); + if (!fault_mgr) { + return tl::make_unexpected(std::string("FaultManager not available")); + } + const auto & cache = services.entity_cache(); + auto entity_ref = cache.find_entity(entity_id); + if (!entity_ref) { + return tl::make_unexpected("Entity not found: " + entity_id); + } + // Same scoping as GET /{entity}/faults: shared source resolution (external app -> bare id, + // external component owns its own id) and the shared boundary-aware filter. + auto source_fqns = faults::resolve_entity_source_fqns(cache, entity_ref->type, entity_id); + auto result = fault_mgr->list_faults(""); + if (!result.success) { + return tl::make_unexpected(result.error_message); + } + auto filtered = faults::filter_faults_by_sources(result.data["faults"], source_fqns); + result.data["count"] = filtered.size(); + result.data["faults"] = std::move(filtered); + return result.data; +} + +Sampled sample_configurations(GatewayServices & services, const std::string & entity_id) { + auto * config_mgr = services.configuration_manager(); + if (!config_mgr) { + return tl::make_unexpected(std::string("ConfigurationManager not available")); + } + auto entity_configs = services.entity_cache().get_entity_configurations(entity_id); + nlohmann::json items = nlohmann::json::array(); + for (const auto & node : entity_configs.nodes) { + auto result = config_mgr->list_parameters(node.node_fqn); + if (result.success && result.data.is_array()) { + for (auto & param : result.data) { + items.push_back(std::move(param)); + } + } + } + nlohmann::json payload; + payload["items"] = std::move(items); + return payload; +} + +Sampled sample_logs(GatewayServices & services, const std::string & entity_id) { + auto * log_mgr = services.log_manager(); + if (!log_mgr) { + return tl::make_unexpected(std::string("LogManager not available")); + } + const auto & cache = services.entity_cache(); + auto entity_ref = cache.find_entity(entity_id); + if (!entity_ref) { + return tl::make_unexpected("Entity not found: " + entity_id); + } + // Same scoping as GET /{entity}/logs via the shared helper (external app -> bare id, external + // component owns its own id, plus the namespace-prefix merge for COMPONENT / AREA). + std::string entity_fqn; + if (entity_ref->type == SovdEntityType::COMPONENT) { + if (auto comp = cache.get_component(entity_id)) { + entity_fqn = comp->fqn; + } + } else if (entity_ref->type == SovdEntityType::AREA) { + if (auto area = cache.get_area(entity_id)) { + entity_fqn = area->namespace_path; + } + } else if (entity_ref->type == SovdEntityType::APP) { + if (auto app = cache.get_app(entity_id)) { + entity_fqn = app->effective_fqn(); + } + } + auto result = logs::query_entity_logs(*log_mgr, cache, entity_ref->type, entity_id, entity_fqn, "", ""); + if (!result.has_value()) { + return tl::make_unexpected(result.error()); + } + nlohmann::json payload; + payload["items"] = std::move(*result); + return payload; +} + +Sampled sample_updates(GatewayServices & services, const std::string & resource_path) { + auto * mgr = services.update_manager(); + if (!mgr || !mgr->has_backend()) { + return tl::make_unexpected(std::string("Update backend not available")); + } + auto result = mgr->get_status(resource_path); + if (!result) { + return tl::make_unexpected(result.error().message); + } + return update_status_to_json(*result); +} + +} // namespace + +void register_builtin_samplers(GatewayServices & services) { + auto & registry = *services.sampler_registry(); + registry.register_sampler( + "data", + [&services](const std::string & entity_id, const std::string & resource_path) { + return sample_data(services, entity_id, resource_path); + }, + /*is_builtin=*/true, /*honours_resource_path=*/true); + registry.register_sampler( + "faults", + [&services](const std::string & entity_id, const std::string & /*resource_path*/) { + return sample_faults(services, entity_id); + }, + /*is_builtin=*/true); + registry.register_sampler( + "configurations", + [&services](const std::string & entity_id, const std::string & /*resource_path*/) { + return sample_configurations(services, entity_id); + }, + /*is_builtin=*/true); + registry.register_sampler( + "logs", + [&services](const std::string & entity_id, const std::string & /*resource_path*/) { + return sample_logs(services, entity_id); + }, + /*is_builtin=*/true); + if (services.update_manager() != nullptr) { + registry.register_sampler( + "updates", + [&services](const std::string & /*entity_id*/, const std::string & resource_path) { + return sample_updates(services, resource_path); + }, + /*is_builtin=*/true, /*honours_resource_path=*/true); + } +} + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/core/data/data_item_reader.cpp b/src/ros2_medkit_gateway/src/core/data/data_item_reader.cpp new file mode 100644 index 000000000..4a974620b --- /dev/null +++ b/src/ros2_medkit_gateway/src/core/data/data_item_reader.cpp @@ -0,0 +1,93 @@ +// Copyright 2026 bburda +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "ros2_medkit_gateway/core/data/data_item_reader.hpp" + +#include +#include +#include +#include +#include + +#include "ros2_medkit_gateway/core/data/topic_data_provider.hpp" +#include "ros2_medkit_gateway/core/http/error_codes.hpp" +#include "ros2_medkit_gateway/core/logging.hpp" +#include "ros2_medkit_gateway/core/managers/data_access_manager.hpp" +#include "ros2_medkit_gateway/core/plugins/plugin_manager.hpp" +#include "ros2_medkit_gateway/http/handlers/handler_support.hpp" + +namespace ros2_medkit_gateway { + +using json = nlohmann::json; +using handlers::make_error; + +ErrorInfo make_plugin_error(int http_status, const std::string & message, json extra_params) { + static constexpr size_t kMaxMessageLength = 512; + int status = http_status < 400 ? 400 : (http_status > 599 ? 599 : http_status); + std::string msg = message.size() > kMaxMessageLength ? message.substr(0, kMaxMessageLength) + "..." : message; + return make_error(status, ERR_PLUGIN_ERROR, std::move(msg), std::move(extra_params)); +} + +ErrorInfo make_provider_error(const DataProviderErrorInfo & info, const std::string & entity_id) { + return make_plugin_error(info.http_status, info.message, json{{"entity_id", entity_id}}); +} + +std::string to_full_topic_path(const std::string & topic_name) { + if (topic_name.empty() || topic_name.front() == '/') { + return topic_name; + } + return "/" + topic_name; +} + +tl::expected read_plugin_data_item(PluginManager * plugins, const std::string & entity_id, + const std::string & topic_name) { + const log::Logger logger("rest_server"); + auto * data_prov = plugins ? plugins->get_data_provider_for_entity(entity_id) : nullptr; + if (data_prov == nullptr) { + // Same x-plc-data route fallback as the data list, single-item form. + if (auto via_route = plugins ? plugins->fetch_entity_data_via_route(entity_id, topic_name) : std::nullopt) { + return dto::DataValue{std::move(*via_route)}; + } + return tl::make_unexpected( + make_error(404, ERR_RESOURCE_NOT_FOUND, "No data provider for plugin entity '" + entity_id + "'")); + } + try { + auto result = data_prov->read_data(entity_id, topic_name); + if (!result) { + return tl::make_unexpected(make_provider_error(result.error(), entity_id)); + } + return std::move(*result); + } catch (const std::exception & e) { + MEDKIT_LOG_ERROR(logger, "Plugin DataProvider threw for entity '%s': %s", entity_id.c_str(), e.what()); + return tl::make_unexpected(make_plugin_error(500, "Plugin threw exception", json{{"entity_id", entity_id}})); + } catch (...) { + MEDKIT_LOG_ERROR(logger, "Plugin DataProvider threw unknown exception for entity '%s'", entity_id.c_str()); + return tl::make_unexpected( + make_plugin_error(500, "Plugin threw unknown exception", json{{"entity_id", entity_id}})); + } +} + +tl::expected sample_data_topic(DataAccessManager & data, + const std::string & full_topic_path) { + // Sampling goes through the pool-backed TopicDataProvider. + auto * provider = data.get_topic_data_provider(); + if (provider == nullptr) { + return tl::make_unexpected(make_error(503, ERR_SERVICE_UNAVAILABLE, "Topic sampling is not configured")); + } + const auto timeout_ms = + std::chrono::milliseconds{static_cast(std::max(data.get_topic_sample_timeout(), 0.0) * 1000.0)}; + return provider->sample(full_topic_path, timeout_ms); +} + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/gateway_node.cpp b/src/ros2_medkit_gateway/src/gateway_node.cpp index cb546c81e..322d831aa 100644 --- a/src/ros2_medkit_gateway/src/gateway_node.cpp +++ b/src/ros2_medkit_gateway/src/gateway_node.cpp @@ -30,12 +30,11 @@ #include #include "ros2_medkit_gateway/core/aggregation/network_utils.hpp" +#include "ros2_medkit_gateway/core/builtin_samplers.hpp" #include "ros2_medkit_gateway/core/data/topic_data_provider.hpp" #include "ros2_medkit_gateway/core/discovery/merge_types.hpp" #include "ros2_medkit_gateway/core/discovery/refresh_debounce.hpp" #include "ros2_medkit_gateway/core/entity_validation.hpp" -#include "ros2_medkit_gateway/core/faults/fault_scope.hpp" -#include "ros2_medkit_gateway/core/logs/log_scope.hpp" #include "ros2_medkit_gateway/core/thread_pool_config.hpp" #include "ros2_medkit_gateway/fault_manager_paths.hpp" #include "ros2_medkit_gateway/param_utils.hpp" @@ -1332,148 +1331,7 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki } // Register built-in resource samplers - sampler_registry_->register_sampler( - "data", - [this](const std::string & /*entity_id*/, - const std::string & resource_path) -> tl::expected { - auto * dam = get_data_access_manager(); - if (!dam) { - return tl::make_unexpected(std::string("DataAccessManager not available")); - } - auto * provider = dam->get_topic_data_provider(); - if (!provider) { - return tl::make_unexpected(std::string("TopicDataProvider not configured")); - } - const auto timeout_ms = std::chrono::milliseconds{ - static_cast(std::max(dam->get_topic_sample_timeout(), 0.0) * 1000.0)}; - auto r = provider->sample(resource_path, timeout_ms); - if (!r) { - return tl::make_unexpected(std::string{"Topic sample failed: "} + r.error().message); - } - TopicSampleResult sample = *r; - if (sample.has_data && sample.data.has_value()) { - nlohmann::json payload; - payload["id"] = resource_path; - payload["data"] = *sample.data; - return payload; - } - return tl::make_unexpected("Topic data not available: " + resource_path); - }, - /*is_builtin=*/true, /*honours_resource_path=*/true); - - sampler_registry_->register_sampler( - "faults", - [this](const std::string & entity_id, - const std::string & /*resource_path*/) -> tl::expected { - auto * fault_mgr = get_fault_manager(); - if (!fault_mgr) { - return tl::make_unexpected(std::string("FaultManager not available")); - } - const auto & cache = get_thread_safe_cache(); - auto entity_ref = cache.find_entity(entity_id); - if (!entity_ref) { - return tl::make_unexpected("Entity not found: " + entity_id); - } - // Same scoping as GET /{entity}/faults: shared source resolution - // (external app -> bare id, external component owns its own id) and - // the shared boundary-aware filter. A private variant here answered - // "Entity no longer available" for external apps whose effective_fqn - // is empty while REST returned the fault. - auto source_fqns = faults::resolve_entity_source_fqns(cache, entity_ref->type, entity_id); - auto result = fault_mgr->list_faults(""); - if (!result.success) { - return tl::make_unexpected(result.error_message); - } - auto filtered = faults::filter_faults_by_sources(result.data["faults"], source_fqns); - result.data["count"] = filtered.size(); - result.data["faults"] = std::move(filtered); - return result.data; - }, - /*is_builtin=*/true); - - sampler_registry_->register_sampler( - "configurations", - [this](const std::string & entity_id, - const std::string & /*resource_path*/) -> tl::expected { - auto * config_mgr = get_configuration_manager(); - if (!config_mgr) { - return tl::make_unexpected(std::string("ConfigurationManager not available")); - } - const auto & cache = get_thread_safe_cache(); - auto entity_configs = cache.get_entity_configurations(entity_id); - nlohmann::json items = nlohmann::json::array(); - for (const auto & node : entity_configs.nodes) { - auto result = config_mgr->list_parameters(node.node_fqn); - if (result.success && result.data.is_array()) { - for (auto & param : result.data) { - items.push_back(std::move(param)); - } - } - } - nlohmann::json payload; - payload["items"] = std::move(items); - return payload; - }, - /*is_builtin=*/true); - - sampler_registry_->register_sampler( - "logs", - [this](const std::string & entity_id, - const std::string & /*resource_path*/) -> tl::expected { - auto * log_mgr = get_log_manager(); - if (!log_mgr) { - return tl::make_unexpected(std::string("LogManager not available")); - } - const auto & cache = get_thread_safe_cache(); - auto entity_ref = cache.find_entity(entity_id); - if (!entity_ref) { - return tl::make_unexpected("Entity not found: " + entity_id); - } - // Same scoping as GET /{entity}/logs via the shared helper (external - // app -> bare id, external component owns its own id, plus the - // namespace-prefix merge for COMPONENT / AREA). - std::string entity_fqn; - if (entity_ref->type == SovdEntityType::COMPONENT) { - if (auto comp = cache.get_component(entity_id)) { - entity_fqn = comp->fqn; - } - } else if (entity_ref->type == SovdEntityType::AREA) { - if (auto area = cache.get_area(entity_id)) { - entity_fqn = area->namespace_path; - } - } else if (entity_ref->type == SovdEntityType::APP) { - if (auto app = cache.get_app(entity_id)) { - entity_fqn = app->effective_fqn(); - } - } - auto result = logs::query_entity_logs(*log_mgr, cache, entity_ref->type, entity_id, entity_fqn, "", ""); - if (!result.has_value()) { - return tl::make_unexpected(result.error()); - } - nlohmann::json payload; - payload["items"] = std::move(*result); - return payload; - }, - /*is_builtin=*/true); - - // Register update status sampler (server-level, uses UpdateManager) - if (update_mgr_) { - sampler_registry_->register_sampler( - "updates", - [this](const std::string & /*entity_id*/, - const std::string & resource_path) -> tl::expected { - auto * mgr = get_update_manager(); - if (!mgr || !mgr->has_backend()) { - return tl::make_unexpected(std::string("Update backend not available")); - } - auto result = mgr->get_status(resource_path); - if (!result) { - return tl::make_unexpected(result.error().message); - } - return update_status_to_json(*result); - }, - /*is_builtin=*/true, /*honours_resource_path=*/true); - } + register_builtin_samplers(*this); RCLCPP_INFO(get_logger(), "Registered built-in resource samplers: data, faults, configurations, logs%s", update_mgr_ ? ", updates" : ""); diff --git a/src/ros2_medkit_gateway/src/http/handlers/data_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/data_handlers.cpp index eb5533d89..43c7b7722 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/data_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/data_handlers.cpp @@ -29,6 +29,7 @@ #include #include +#include "ros2_medkit_gateway/core/data/data_item_reader.hpp" #include "ros2_medkit_gateway/core/data/topic_data_provider.hpp" #include "ros2_medkit_gateway/core/exceptions.hpp" #include "ros2_medkit_gateway/core/http/error_codes.hpp" @@ -55,21 +56,6 @@ using json = nlohmann::json; // Helper free functions // ============================================================================= -/// Sanitize a plugin-supplied error into the standard `x-medkit-plugin-error` -/// shape: clamp HTTP status to [400, 599] and truncate message at 512 chars. -ErrorInfo make_plugin_error(int http_status, const std::string & message, json extra_params = {}) { - static constexpr size_t kMaxMessageLength = 512; - int status = http_status < 400 ? 400 : (http_status > 599 ? 599 : http_status); - std::string msg = message.size() > kMaxMessageLength ? message.substr(0, kMaxMessageLength) + "..." : message; - return make_error(status, ERR_PLUGIN_ERROR, std::move(msg), std::move(extra_params)); -} - -/// Map a `DataProviderErrorInfo` (from the typed plugin ABI) into the SOVD -/// `x-medkit-plugin-error` wire shape via `make_plugin_error`. -ErrorInfo make_provider_error(const DataProviderErrorInfo & info, const std::string & entity_id) { - return make_plugin_error(info.http_status, info.message, json{{"entity_id", entity_id}}); -} - /// Read the first positional capture group (entity_id) with the same "missing /// capture is treated as 400 invalid-request" semantics as the other migrated /// handlers (matches the legacy `req.matches.size() < N` guard). @@ -91,15 +77,6 @@ tl::expected read_topic_id(const http::TypedRequest & re return tl::make_unexpected(make_error(400, ERR_INVALID_REQUEST, "Invalid request")); } -/// A data item is addressed by its full ROS topic path; the leading slash is -/// optional on the wire because the route captures a percent-decoded segment. -std::string to_full_topic_path(const std::string & topic_name) { - if (topic_name.empty() || topic_name.front() == '/') { - return topic_name; - } - return "/" + topic_name; -} - /// What one addressed data item resolves to: the ROS topic to act on, the id to /// echo back to the caller, the member the id named, whether this gateway's own /// walk records that member as a provider of the topic, and the member the tree @@ -637,32 +614,7 @@ http::Result DataHandlers::get_data_item(const http::TypedReques // Delegate to plugin DataProvider for plugin-owned entities. if (entity_info.is_plugin) { - auto * pmgr = ctx_.services().plugin_manager(); - auto * data_prov = pmgr ? pmgr->get_data_provider_for_entity(entity_id) : nullptr; - if (data_prov == nullptr) { - // Same x-plc-data route fallback as list_data above, single-item form. - if (auto via_route = pmgr ? pmgr->fetch_entity_data_via_route(entity_id, topic_name) : std::nullopt) { - return dto::DataValue{std::move(*via_route)}; - } - return tl::make_unexpected( - make_error(404, ERR_RESOURCE_NOT_FOUND, "No data provider for plugin entity '" + entity_id + "'")); - } - try { - auto result = data_prov->read_data(entity_id, topic_name); - if (!result) { - return tl::make_unexpected(make_provider_error(result.error(), entity_id)); - } - return std::move(*result); - } catch (const std::exception & e) { - MEDKIT_LOG_ERROR(HandlerContext::logger(), "Plugin DataProvider threw for entity '%s': %s", entity_id.c_str(), - e.what()); - return tl::make_unexpected(make_plugin_error(500, "Plugin threw exception", json{{"entity_id", entity_id}})); - } catch (...) { - MEDKIT_LOG_ERROR(HandlerContext::logger(), "Plugin DataProvider threw unknown exception for entity '%s'", - entity_id.c_str()); - return tl::make_unexpected( - make_plugin_error(500, "Plugin threw unknown exception", json{{"entity_id", entity_id}})); - } + return read_plugin_data_item(ctx_.services().plugin_manager(), entity_id, topic_name); } try { @@ -677,16 +629,7 @@ http::Result DataHandlers::get_data_item(const http::TypedReques } const std::string & full_topic_path = addressed->full_topic_path; - // Sampling goes through the pool-backed TopicDataProvider (issue #375 race - // fix). The provider is configured in main() before serving traffic. - auto data_access_mgr = ctx_.services().data_access_manager(); - const auto timeout_sec = data_access_mgr->get_topic_sample_timeout(); - auto * provider = data_access_mgr->get_topic_data_provider(); - if (provider == nullptr) { - return tl::make_unexpected(make_error(503, ERR_SERVICE_UNAVAILABLE, "Topic sampling is not configured")); - } - const auto timeout_ms = std::chrono::milliseconds{static_cast(std::max(timeout_sec, 0.0) * 1000.0)}; - auto r = provider->sample(full_topic_path, timeout_ms); + auto r = sample_data_topic(*ctx_.services().data_access_manager(), full_topic_path); if (!r) { // Propagate the provider's ErrorInfo verbatim: http_status, code (SOVD // constant or x-medkit-*), message, and structured params. Collapsing From 20d55fa7e4c0ff5252de99deaf0b33260b111ad9 Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Sun, 25 Oct 2026 13:40:44 +0100 Subject: [PATCH 2/7] feat(gateway): the core builds the bulk-data store and reports only the categories the host serves BulkDataConfig and make_bulk_data_store() build the store from the bulk_data settings, and the node uses them. The rosbags category is listed only when the fault manager transport serves rosbags, and GET / reports bulk_data and cyclic_subscriptions only when the host provides a store or rosbags, and a transport and a sampler. --- docs/config/server.rst | 7 +- scripts/check_cores_without_ros/README.md | 7 + scripts/check_cores_without_ros/main.cpp | 120 +++++++++++++++++- src/ros2_medkit_gateway/README.md | 7 +- src/ros2_medkit_gateway/design/index.rst | 22 ++++ .../ros2_medkit_gateway/core/config.hpp | 10 ++ .../core/managers/bulk_data_store.hpp | 7 + .../core/managers/fault_manager.hpp | 3 + .../core/subscription_transport.hpp | 2 + .../transports/fault_service_transport.hpp | 7 + .../src/core/managers/bulk_data_store.cpp | 27 ++++ .../src/core/managers/fault_manager.cpp | 4 + .../src/core/subscription_transport.cpp | 5 + src/ros2_medkit_gateway/src/gateway_node.cpp | 41 ++---- .../src/http/handlers/bulkdata_handlers.cpp | 6 +- .../src/http/handlers/health_handlers.cpp | 14 +- 16 files changed, 249 insertions(+), 40 deletions(-) diff --git a/docs/config/server.rst b/docs/config/server.rst index 848dc55ae..1b1fc8c3b 100644 --- a/docs/config/server.rst +++ b/docs/config/server.rst @@ -696,7 +696,7 @@ Bulk Data Storage ----------------- Configure file-based bulk data storage for uploads (calibration files, firmware, etc.). -The ``rosbags`` category is always available via the Fault Manager and does not need +The ``rosbags`` category is served by the Fault Manager and does not need configuration. .. list-table:: @@ -733,8 +733,9 @@ Example: .. note:: - The ``rosbags`` category is always available through the Fault Manager and cannot be - used for uploads or deletes. It is automatically included in the category list. + The ``rosbags`` category is available through the Fault Manager and cannot be + used for uploads or deletes. It is included in the category list whenever the Fault + Manager serves rosbags, which is always the case on the ROS 2 node. .. note:: diff --git a/scripts/check_cores_without_ros/README.md b/scripts/check_cores_without_ros/README.md index 2469f83e3..360ca0097 100644 --- a/scripts/check_cores_without_ros/README.md +++ b/scripts/check_cores_without_ros/README.md @@ -17,6 +17,13 @@ not installed, so the PostgreSQL backend is not built. ## What `main.cpp` checks +- Cyclic subscriptions: with `register_builtin_samplers()` and the SSE transport, a subscription on a data resource (served by + a controlled `TopicDataProvider`), on a resource a plugin's `DataProvider` serves, and on faults each stream two events whose + values differ. +- Bulk data: a store built by `make_bulk_data_store()` takes an upload over HTTP, lists it, returns it byte-equal and deletes + it; an upload past `max_upload_size` is refused; the categories do not contain `rosbags` when the fault manager serves none. +- `GET /` capabilities: `bulk_data` and `cyclic_subscriptions` are false for services without a store, rosbags, transport or + sampler, and true as soon as a store or rosbags (bulk data), or a transport and a sampler (subscriptions), exist. - Gateway: with `server_name`, `vendor_name` and `server_version` set in `RestConfig`, `GET /api/v1/` returns the name and version, and `GET /api/v1/version-info` returns the vendor name and version. - Reporter core: a report below the local filter threshold is filtered, the next one reaches the sink, a PASSED report is diff --git a/scripts/check_cores_without_ros/main.cpp b/scripts/check_cores_without_ros/main.cpp index 1b2f45752..490bd1384 100644 --- a/scripts/check_cores_without_ros/main.cpp +++ b/scripts/check_cores_without_ros/main.cpp @@ -55,6 +55,7 @@ #include "ros2_medkit_gateway/core/http/rate_limiter.hpp" #include "ros2_medkit_gateway/core/http/rest_server.hpp" #include "ros2_medkit_gateway/core/http/sse_client_tracker.hpp" +#include "ros2_medkit_gateway/core/managers/bulk_data_store.hpp" #include "ros2_medkit_gateway/core/managers/configuration_manager.hpp" #include "ros2_medkit_gateway/core/managers/data_access_manager.hpp" #include "ros2_medkit_gateway/core/managers/fault_manager.hpp" @@ -167,7 +168,11 @@ class NoFaultTransport : public gw::FaultServiceTransport { bool is_available() const override { return false; } + bool serves_rosbags() const override { + return rosbags.load(); + } std::atomic reports_faults{false}; + std::atomic rosbags{false}; std::atomic list_calls{0}; private: @@ -466,7 +471,7 @@ class MinimalServices : public gw::GatewayServices { return nullptr; } gw::BulkDataStore * bulk_data_store() override { - return nullptr; + return bulk_store_.get(); } gw::FaultTriggerEngine * fault_trigger_engine() override { return nullptr; @@ -499,12 +504,16 @@ class MinimalServices : public gw::GatewayServices { plugins_->register_entity_ownership("check_plugin", owned_entities); return ref; } + void set_bulk_store(std::unique_ptr store) { + bulk_store_ = std::move(store); + } private: gw::ThreadSafeEntityCache cache_; EmptyDiscovery discovery_; NoTypeSchemas schemas_; std::shared_ptr fault_transport_; + std::unique_ptr bulk_store_; std::unique_ptr plugins_; gw::FaultManager fault_; gw::ConfigurationManager config_; @@ -623,6 +632,17 @@ bool values_change(const std::vector & events, return !first.is_null() && !second.is_null() && first != second; } +// `GET /` capabilities of a server over `services`. +nlohmann::json root_capabilities(MinimalServices & services) { + RunningServer running(services); + auto client = make_client(running.port()); + auto res = client.Get("/api/v1/"); + if (!res || res->status != 200) { + return nlohmann::json::object(); + } + return nlohmann::json::parse(res->body, nullptr, false).value("capabilities", nlohmann::json::object()); +} + gw::App hosted_app() { gw::App app; app.id = "hosted_app"; @@ -938,7 +958,7 @@ static int run() { server_thread.join(); } - // A host that registers the samplers and a transport gets cyclic subscriptions. + // A host that registers the samplers, a transport and a bulk-data store gets subscriptions and bulk data. { MinimalServices services; CountingTopicProvider topics; @@ -948,6 +968,13 @@ static int run() { plugin_app.id = "plugin_app"; services.mutable_cache().update_apps({hosted_app(), plugin_app}); services.add_data_plugin({"plugin_app"}); + const auto store_dir = std::filesystem::temp_directory_path() / "check_cores_without_ros_bulk"; + std::filesystem::remove_all(store_dir); + gw::BulkDataConfig bulk; + bulk.storage_dir = store_dir.string(); + bulk.max_upload_size = 1024; + bulk.categories = {"calibration", ""}; + services.set_bulk_store(gw::make_bulk_data_store(bulk)); gw::register_builtin_samplers(services); add_sse_transport(services); { @@ -1008,8 +1035,97 @@ static int run() { if (!fault_sub.first.empty()) { client.Delete("/api/v1/apps/hosted_app/cyclic-subscriptions/" + fault_sub.first); } + + // Bulk data: upload, list, download, delete. + const std::string payload("bulk\0payload\xff\x01", 15); + const std::string base = "/api/v1/apps/hosted_app/bulk-data"; + httplib::MultipartFormDataItems items = {{"file", payload, "cal.bin", "application/octet-stream"}}; + auto uploaded = client.Post(base + "/calibration", items); + ok &= check(uploaded && uploaded->status == 201, "an upload to a configured category answers 201"); + const std::string item_id = uploaded && uploaded->status == 201 + ? nlohmann::json::parse(uploaded->body, nullptr, false).value("id", "") + : ""; + auto listed = client.Get(base + "/calibration"); + const auto listed_json = listed ? nlohmann::json::parse(listed->body, nullptr, false) : nlohmann::json(); + ok &= + check(listed && listed->status == 200 && listed_json.contains("items") && listed_json["items"].size() == 1 && + listed_json["items"][0].value("id", "") == item_id && !item_id.empty(), + "the uploaded item is listed"); + auto downloaded = client.Get(base + "/calibration/" + item_id); + ok &= check(downloaded && downloaded->status == 200 && downloaded->body == payload, "the download is byte-equal"); + auto removed = client.Delete(base + "/calibration/" + item_id); + ok &= check(removed && removed->status == 204, "the delete answers 204"); + auto relisted = client.Get(base + "/calibration"); + const auto relisted_json = relisted ? nlohmann::json::parse(relisted->body, nullptr, false) : nlohmann::json(); + ok &= check(relisted && relisted_json.contains("items") && relisted_json["items"].empty(), + "the deleted item is gone"); + auto too_big = client.Post( + base + "/calibration", + httplib::MultipartFormDataItems{{"file", std::string(2048, 'x'), "big.bin", "application/octet-stream"}}); + ok &= check(too_big && too_big->status >= 400, "an upload past max_upload_size is refused"); + + // Categories: the store's own, and no rosbags when the fault manager serves none. + auto categories = client.Get(base); + const auto categories_json = + categories ? nlohmann::json::parse(categories->body, nullptr, false) : nlohmann::json(); + const auto cat_items = categories_json.value("items", nlohmann::json::array()); + ok &= check(categories && categories->status == 200 && + std::find(cat_items.begin(), cat_items.end(), nlohmann::json("calibration")) != cat_items.end(), + "the categories list the configured category"); + ok &= check(std::find(cat_items.begin(), cat_items.end(), nlohmann::json("rosbags")) == cat_items.end(), + "the categories do not list rosbags when the fault manager serves none"); + + auto root = client.Get("/api/v1/"); + const auto caps = + root ? nlohmann::json::parse(root->body, nullptr, false).value("capabilities", nlohmann::json::object()) + : nlohmann::json::object(); + ok &= check(caps.value("bulk_data", false), "GET / reports bulk_data true with a store"); + ok &= check(caps.value("cyclic_subscriptions", false), + "GET / reports cyclic_subscriptions true with samplers and a transport"); } services.transport_registry()->shutdown_all(*services.subscription_manager()); + std::filesystem::remove_all(store_dir); + } + + // Capabilities follow what the host provides. + { + MinimalServices bare; + const auto caps = root_capabilities(bare); + ok &= + check(caps.contains("bulk_data") && !caps.value("bulk_data", true), "no store and no rosbags: bulk_data false"); + ok &= check(caps.contains("cyclic_subscriptions") && !caps.value("cyclic_subscriptions", true), + "no transport and no sampler: cyclic_subscriptions false"); + + MinimalServices transport_only; + add_sse_transport(transport_only); + const auto transport_caps = root_capabilities(transport_only); + ok &= check(transport_caps.contains("cyclic_subscriptions") && !transport_caps.value("cyclic_subscriptions", true), + "a transport without a sampler: cyclic_subscriptions false"); + transport_only.transport_registry()->shutdown_all(*transport_only.subscription_manager()); + + MinimalServices sampler_only; + sampler_only.sampler_registry()->register_sampler( + "x-check", [](const std::string &, const std::string &) -> tl::expected { + return nlohmann::json::object(); + }); + const auto sampler_caps = root_capabilities(sampler_only); + ok &= check(sampler_caps.contains("cyclic_subscriptions") && !sampler_caps.value("cyclic_subscriptions", true), + "a sampler without a transport: cyclic_subscriptions false"); + + MinimalServices store_only; + gw::BulkDataConfig store_only_config; + store_only_config.storage_dir = + (std::filesystem::temp_directory_path() / "check_cores_without_ros_store_only").string(); + store_only.set_bulk_store(gw::make_bulk_data_store(store_only_config)); + store_only.fault_transport().rosbags = false; + const auto store_caps = root_capabilities(store_only); + ok &= check(store_caps.value("bulk_data", false), "a store without rosbags: bulk_data true"); + std::filesystem::remove_all(std::filesystem::temp_directory_path() / "check_cores_without_ros_store_only"); + + MinimalServices rosbags_only; + rosbags_only.fault_transport().rosbags = true; + const auto rosbag_caps = root_capabilities(rosbags_only); + ok &= check(rosbag_caps.value("bulk_data", false), "rosbags without a store: bulk_data true"); } return ok ? 0 : 1; diff --git a/src/ros2_medkit_gateway/README.md b/src/ros2_medkit_gateway/README.md index 04125c74c..2c2ddb672 100644 --- a/src/ros2_medkit_gateway/README.md +++ b/src/ros2_medkit_gateway/README.md @@ -66,7 +66,7 @@ All endpoints are prefixed with `/api/v1` for API versioning. ### Bulk Data Endpoints -- `GET /api/v1/{entity}/{id}/bulk-data` - List bulk-data categories (rosbags + configured) +- `GET /api/v1/{entity}/{id}/bulk-data` - List bulk-data categories (configured, plus `rosbags` when the fault manager serves them) - `GET /api/v1/{entity}/{id}/bulk-data/{category}` - List bulk-data items in a category - `GET /api/v1/{entity}/{id}/bulk-data/{category}/{item_id}` - Download a bulk-data file - `POST /api/v1/{entity}/{id}/bulk-data/{category}` - Upload bulk data (components/apps only) @@ -1761,6 +1761,11 @@ target_link_libraries(check_cores_without_ros PRIVATE gateway_core) The file takes a target the including build already defines, else `find_package`, else the copy vendored in `src/vendored/` (cpp-httplib, jwt-cpp, tl::expected). An existing `httplib::httplib` target is reused when it is older than 0.20 (or has no version), because 0.20 removed the multipart API the handlers use. `gateway_core` requires C++17 from its consumers (`cxx_std_17` is a public compile feature) and works with CMake 3.16. `-DENABLE_SWAGGER_UI=ON` works in both builds and needs network access and `xxd` at configure time. The check implements `GatewayServices` with minimal stand-ins and starts `RESTServer`. A program that hosts the core names its server through `RestConfig::server_name`, `vendor_name` and `server_version` (`name` and `version` on `GET /`, `vendor_info` on `GET /version-info`, and the OpenAPI title and version). The defaults are the ROS 2 node's values. `scripts/check_cores_without_ros/` builds and runs it on Ubuntu 20.04 with gcc 9, the oldest compiler the cores are checked with. +A program that hosts the core gets the cyclic-subscription samplers and the bulk-data store from the core, and passes the following: + +- **Samplers.** `register_builtin_samplers(services)` (`core/builtin_samplers.hpp`) registers `data`, `faults`, `configurations` and `logs` on `services.sampler_registry()`, plus `updates` when `services.update_manager()` is set. Each sampler reads through the same managers and helpers as its GET route. A `data` subscription on an entity a plugin owns reads through that plugin's `DataProvider` (or its `x-plc-data` route), and on any other entity through `services.data_access_manager()` and its `TopicDataProvider`. A collection with no manager behind it answers with the error text of its GET route, not a crash. The host passes `GatewayServices` (it must outlive the samplers) and registers a subscription transport itself, for example `SseTransportProvider`. `GET /` reports `cyclic_subscriptions` true only when a transport and at least one sampler are registered. +- **Bulk data.** `BulkDataConfig` (`core/config.hpp`: `storage_dir`, `max_upload_size`, `categories`, with the defaults of the `bulk_data.*` parameters) and `make_bulk_data_store(config)` (`core/managers/bulk_data_store.hpp`) build the store. The host returns it from `GatewayServices::bulk_data_store()` before it starts `RESTServer`, which sizes the upload limit from it. The `rosbags` category is listed only when `FaultServiceTransport::serves_rosbags()` is true. The default is true; a host whose fault manager records no rosbags overrides it. `GET /` reports `bulk_data` true only when a store exists or rosbags are served. + The `gateway_node` executable and existing test targets link `gateway_ros2`, so they transitively get both layers from a single dependency. The neutral contract is enforced by two CTest checks: - `gateway_core_purity` (linter label) - greps `core/` for any ROS-package include and fails on any match. diff --git a/src/ros2_medkit_gateway/design/index.rst b/src/ros2_medkit_gateway/design/index.rst index 09c3e0799..855db0262 100644 --- a/src/ros2_medkit_gateway/design/index.rst +++ b/src/ros2_medkit_gateway/design/index.rst @@ -35,6 +35,28 @@ The check implements ``GatewayServices`` with minimal stand-ins and starts defaults are the ROS 2 node's values. ``scripts/check_cores_without_ros/`` builds and runs it on Ubuntu 20.04 with gcc 9, the oldest compiler the cores are checked with. +A program that hosts the core also gets the cyclic-subscription samplers and the bulk-data +store from it: + +* ``register_builtin_samplers(services)`` (``core/builtin_samplers.hpp``) + registers ``data``, ``faults``, ``configurations`` and ``logs`` on + ``services.sampler_registry()``, plus ``updates`` when + ``services.update_manager()`` is set. Each sampler reads through the + managers and helpers of its GET route. ``data`` on a plugin-owned entity + goes through the plugin's ``DataProvider`` (``core/data/data_item_reader.hpp``, + shared with ``GET /{entity}/data/{item}``); on any other entity it goes + through ``DataAccessManager`` and its ``TopicDataProvider``. The host passes + ``GatewayServices``, which must outlive the samplers, and registers a + subscription transport itself. +* ``BulkDataConfig`` (``core/config.hpp``) and ``make_bulk_data_store()`` + (``core/managers/bulk_data_store.hpp``) build the store. The host returns it + from ``GatewayServices::bulk_data_store()`` before it starts ``RESTServer``. + The ``rosbags`` category is listed only when + ``FaultServiceTransport::serves_rosbags()`` is true (the default). +* ``GET /`` reports ``bulk_data`` true only when a store exists or rosbags are + served, and ``cyclic_subscriptions`` true only when a transport and at least + one sampler are registered. + The ``gateway_node`` executable and existing test targets link ``gateway_ros2``, so they transitively get both layers from a single dependency. The neutral contract is enforced by two CTest checks: diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/config.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/config.hpp index 9fbc5bdaa..2472b9aa3 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/config.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/config.hpp @@ -46,6 +46,16 @@ struct RestConfig : ServerIdentity { std::string logger_name{"rest_server"}; }; +/// Where uploaded bulk data goes and what a host accepts. The defaults are the node's `bulk_data.*` defaults. +struct BulkDataConfig { + /// Base directory of the stored files. + std::string storage_dir{"/tmp/ros2_medkit_bulk_data"}; + /// Largest accepted upload in bytes. 0 means unlimited. A negative value is refused and the default is used. + int64_t max_upload_size{104857600}; + /// Accepted upload categories. Empty names are dropped. `rosbags` is never listed here. + std::vector categories; +}; + /** * @brief TLS/HTTPS configuration settings * diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/bulk_data_store.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/bulk_data_store.hpp index 0c3982560..aeef86084 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/bulk_data_store.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/bulk_data_store.hpp @@ -15,6 +15,7 @@ #pragma once #include +#include #include #include #include @@ -23,6 +24,8 @@ #include #include +#include "ros2_medkit_gateway/core/config.hpp" + namespace ros2_medkit_gateway { /** @@ -148,4 +151,8 @@ class BulkDataStore { static tl::expected validate_path_component(const std::string & value, const std::string & name); }; +/// Builds the store a host serves bulk data from. Drops empty category names. A negative +/// `max_upload_size` is logged and replaced by the default of `BulkDataConfig`. +std::unique_ptr make_bulk_data_store(const BulkDataConfig & config); + } // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/fault_manager.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/fault_manager.hpp index 2b6627b7f..1386d1daf 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/fault_manager.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/fault_manager.hpp @@ -96,6 +96,9 @@ class FaultManager { /// Check if fault manager services are available. bool is_available() const; + /// Whether the fault manager can serve rosbag files (forwards to transport). + bool serves_rosbags() const; + /// Wait for fault manager services to become available (forwards to transport). bool wait_for_services(std::chrono::duration timeout); diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/subscription_transport.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/subscription_transport.hpp index 9e2fc0c82..2d9084b23 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/subscription_transport.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/subscription_transport.hpp @@ -90,6 +90,8 @@ class TransportRegistry { void register_transport(std::unique_ptr provider); SubscriptionTransportProvider * get_transport(const std::string & protocol) const; bool has_transport(const std::string & protocol) const; + /// True when no transport is registered. + bool empty() const; /// Shutdown all subscriptions by delegating to SubscriptionManager::shutdown(). /// Transport cleanup happens via the manager's on_removed callback, which diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/fault_service_transport.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/fault_service_transport.hpp index 8a4c1288c..97f908424 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/fault_service_transport.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/fault_service_transport.hpp @@ -63,6 +63,13 @@ class FaultServiceTransport { virtual bool wait_for_services(std::chrono::duration timeout) = 0; virtual bool is_available() const = 0; + + /// Whether this transport serves rosbag files (`get_rosbag`, `list_rosbags`). The bulk-data + /// `rosbags` category is listed only when it does. A host whose fault manager records no + /// rosbags overrides it to return false. + virtual bool serves_rosbags() const { + return true; + } }; } // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/core/managers/bulk_data_store.cpp b/src/ros2_medkit_gateway/src/core/managers/bulk_data_store.cpp index 27e8384b1..9763426ab 100644 --- a/src/ros2_medkit_gateway/src/core/managers/bulk_data_store.cpp +++ b/src/ros2_medkit_gateway/src/core/managers/bulk_data_store.cpp @@ -14,13 +14,16 @@ #include "ros2_medkit_gateway/core/managers/bulk_data_store.hpp" +#include #include +#include #include #include #include #include #include "ros2_medkit_gateway/core/http/http_utils.hpp" +#include "ros2_medkit_gateway/core/logging.hpp" namespace ros2_medkit_gateway { @@ -35,6 +38,30 @@ BulkDataStore::BulkDataStore(const std::string & storage_dir, size_t max_upload_ std::filesystem::create_directories(storage_dir_, ec); } +std::unique_ptr make_bulk_data_store(const BulkDataConfig & config) { + // A negative value would wrap to a huge size_t and be handed to set_payload_max_length as an + // effectively unlimited cap. 0 already means unlimited by design, so a negative is refused rather + // than folded into it. + size_t max_upload = 0; + if (config.max_upload_size < 0) { + const BulkDataConfig defaults; + MEDKIT_LOG_WARN(log::Logger("bulk_data"), + "bulk_data.max_upload_size %" PRId64 + " is negative. Must be >= 0 (0 = unlimited). Using default %" PRId64 ".", + config.max_upload_size, defaults.max_upload_size); + max_upload = static_cast(defaults.max_upload_size); + } else { + max_upload = static_cast(config.max_upload_size); + } + auto categories = config.categories; + categories.erase(std::remove_if(categories.begin(), categories.end(), + [](const std::string & item) { + return item.empty(); + }), + categories.end()); + return std::make_unique(config.storage_dir, max_upload, std::move(categories)); +} + // --- Validation --- tl::expected BulkDataStore::validate_path_component(const std::string & value, diff --git a/src/ros2_medkit_gateway/src/core/managers/fault_manager.cpp b/src/ros2_medkit_gateway/src/core/managers/fault_manager.cpp index cb02ef227..c406cbac6 100644 --- a/src/ros2_medkit_gateway/src/core/managers/fault_manager.cpp +++ b/src/ros2_medkit_gateway/src/core/managers/fault_manager.cpp @@ -61,6 +61,10 @@ bool FaultManager::is_available() const { return transport_->is_available(); } +bool FaultManager::serves_rosbags() const { + return transport_->serves_rosbags(); +} + bool FaultManager::wait_for_services(std::chrono::duration timeout) { return transport_->wait_for_services(timeout); } diff --git a/src/ros2_medkit_gateway/src/core/subscription_transport.cpp b/src/ros2_medkit_gateway/src/core/subscription_transport.cpp index 2144def53..71b2ee71a 100644 --- a/src/ros2_medkit_gateway/src/core/subscription_transport.cpp +++ b/src/ros2_medkit_gateway/src/core/subscription_transport.cpp @@ -41,6 +41,11 @@ bool TransportRegistry::has_transport(const std::string & protocol) const { return transports_.count(protocol) > 0; } +bool TransportRegistry::empty() const { + std::shared_lock lock(mutex_); + return transports_.empty(); +} + void TransportRegistry::shutdown_all(SubscriptionManager & sub_mgr) { // Delegate to sub_mgr.shutdown() which triggers on_removed callbacks // for each active subscription, causing transport->stop() for each. diff --git a/src/ros2_medkit_gateway/src/gateway_node.cpp b/src/ros2_medkit_gateway/src/gateway_node.cpp index 322d831aa..0fb4343fe 100644 --- a/src/ros2_medkit_gateway/src/gateway_node.cpp +++ b/src/ros2_medkit_gateway/src/gateway_node.cpp @@ -269,8 +269,9 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki declare_parameter("aggregation.peer_names", std::vector{""}); // Bulk data storage parameters - declare_parameter("bulk_data.storage_dir", "/tmp/ros2_medkit_bulk_data"); - declare_parameter("bulk_data.max_upload_size", 104857600); // 100MB + const BulkDataConfig bulk_data_defaults; + declare_parameter("bulk_data.storage_dir", bulk_data_defaults.storage_dir); + declare_parameter("bulk_data.max_upload_size", bulk_data_defaults.max_upload_size); declare_parameter("bulk_data.categories", std::vector{}); // Runtime (heuristic) discovery options @@ -691,34 +692,14 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki fault_mgr_ = std::make_unique(fault_service_transport_); // Initialize bulk data store - auto bd_storage_dir = get_parameter("bulk_data.storage_dir").as_string(); - // A negative value would wrap to a huge size_t and be handed to - // set_payload_max_length as an effectively unlimited cap. 0 already means - // unlimited by documented design, so a negative cannot be folded into it - // without turning a typo into that same silent "no limit". Reject it and say - // which value was refused. - const int64_t bd_max_upload_raw = get_parameter("bulk_data.max_upload_size").as_int(); - size_t bd_max_upload = 0; - if (bd_max_upload_raw < 0) { - RCLCPP_WARN(get_logger(), - "bulk_data.max_upload_size %" PRId64 - " is negative. Must be >= 0 (0 = unlimited). " - "Using default 104857600.", - bd_max_upload_raw); - bd_max_upload = 104857600; - } else { - bd_max_upload = static_cast(bd_max_upload_raw); - } - auto bd_categories = get_parameter("bulk_data.categories").as_string_array(); - bd_categories.erase(std::remove_if(bd_categories.begin(), bd_categories.end(), - [](const auto & item) { - return item.empty(); - }), - bd_categories.end()); - - bulk_data_store_ = std::make_unique(bd_storage_dir, bd_max_upload, bd_categories); - RCLCPP_INFO(get_logger(), "Bulk data store: dir=%s, max_upload=%zuB, categories=%zu", bd_storage_dir.c_str(), - bd_max_upload, bd_categories.size()); + BulkDataConfig bulk_data_config; + bulk_data_config.storage_dir = get_parameter("bulk_data.storage_dir").as_string(); + bulk_data_config.max_upload_size = get_parameter("bulk_data.max_upload_size").as_int(); + bulk_data_config.categories = get_parameter("bulk_data.categories").as_string_array(); + bulk_data_store_ = make_bulk_data_store(bulk_data_config); + RCLCPP_INFO(get_logger(), "Bulk data store: dir=%s, max_upload=%zuB, categories=%zu", + bulk_data_config.storage_dir.c_str(), bulk_data_store_->max_upload_bytes(), + bulk_data_store_->list_categories().size()); // Initialize subscription manager for cyclic subscriptions // Same wrap hazard as bulk_data.max_upload_size: a negative here becomes a diff --git a/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp index 1c8063f89..818bdebfa 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp @@ -289,9 +289,11 @@ http::Result BulkDataHandlers::list_categories(const return tl::unexpected(make_error(400, ERR_COLLECTION_NOT_SUPPORTED, access.error().message)); } - // Build categories list: "rosbags" always available + BulkDataStore categories. + // Build categories list: "rosbags" when the fault manager serves them + BulkDataStore categories. dto::BulkDataCategoryList response; - response.items.push_back("rosbags"); // Always available via FaultManager + if (auto * fault_mgr = ctx_.services().fault_manager(); fault_mgr != nullptr && fault_mgr->serves_rosbags()) { + response.items.push_back("rosbags"); + } auto * store = ctx_.bulk_data_store(); if (store) { diff --git a/src/ros2_medkit_gateway/src/http/handlers/health_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/health_handlers.cpp index 5708b5ce4..591cac254 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/health_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/health_handlers.cpp @@ -27,12 +27,16 @@ #include "ros2_medkit_gateway/core/http/fan_out_helpers.hpp" #include "ros2_medkit_gateway/core/http/http_utils.hpp" #include "ros2_medkit_gateway/core/http/warning_codes.hpp" +#include "ros2_medkit_gateway/core/managers/bulk_data_store.hpp" +#include "ros2_medkit_gateway/core/managers/fault_manager.hpp" #include "ros2_medkit_gateway/core/managers/lock_manager.hpp" #include "ros2_medkit_gateway/core/managers/script_manager.hpp" #include "ros2_medkit_gateway/core/managers/trigger_manager.hpp" #include "ros2_medkit_gateway/core/managers/update_manager.hpp" #include "ros2_medkit_gateway/core/models/thread_safe_entity_cache.hpp" #include "ros2_medkit_gateway/core/plugins/plugin_manager.hpp" +#include "ros2_medkit_gateway/core/resource_sampler.hpp" +#include "ros2_medkit_gateway/core/subscription_transport.hpp" #include "ros2_medkit_gateway/core/version.hpp" #include "ros2_medkit_gateway/dto/health.hpp" @@ -285,8 +289,14 @@ http::Result HealthHandlers::get_root(const http::TypedReques capabilities.configurations = true; capabilities.faults = true; capabilities.logs = true; - capabilities.bulk_data = true; - capabilities.cyclic_subscriptions = true; + const auto * fault_mgr = ctx_.services().fault_manager(); + const auto * transports = ctx_.services().transport_registry(); + const auto * samplers = ctx_.services().sampler_registry(); + capabilities.bulk_data = + ctx_.services().bulk_data_store() != nullptr || (fault_mgr != nullptr && fault_mgr->serves_rosbags()); + capabilities.cyclic_subscriptions = ctx_.services().subscription_manager() != nullptr && transports != nullptr && + !transports->empty() && samplers != nullptr && + !samplers->collection_names().empty(); capabilities.locking = ctx_.services().lock_manager() != nullptr; capabilities.triggers = ctx_.services().trigger_manager() != nullptr; capabilities.updates = ctx_.services().update_manager() != nullptr; From 7f9c8237de85de527dc3e6a9225d412021bad90f Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Sun, 25 Oct 2026 13:43:53 +0100 Subject: [PATCH 3/7] fix(gateway): a host without rosbags answers the rosbags category as an unknown category The bulk-data list and download routes take the rosbags path only when the fault manager serves rosbags. Otherwise the category falls through to the store check and answers 404 Unknown category. --- scripts/check_cores_without_ros/main.cpp | 18 ++++++++++++++++++ src/ros2_medkit_gateway/README.md | 2 +- .../src/http/handlers/bulkdata_handlers.cpp | 14 +++++++++++--- 3 files changed, 30 insertions(+), 4 deletions(-) diff --git a/scripts/check_cores_without_ros/main.cpp b/scripts/check_cores_without_ros/main.cpp index 490bd1384..ff817f24d 100644 --- a/scripts/check_cores_without_ros/main.cpp +++ b/scripts/check_cores_without_ros/main.cpp @@ -1075,6 +1075,24 @@ static int run() { ok &= check(std::find(cat_items.begin(), cat_items.end(), nlohmann::json("rosbags")) == cat_items.end(), "the categories do not list rosbags when the fault manager serves none"); + // A host without rosbags answers rosbags like any category that does not exist. + const auto error_code_of = [](const httplib::Result & r) { + return r ? nlohmann::json::parse(r->body, nullptr, false).value("error_code", "") : std::string{}; + }; + auto no_rosbag_list = client.Get(base + "/rosbags"); + auto no_such_list = client.Get(base + "/no_such_category"); + ok &= check(no_such_list && no_such_list->status == 404, "an unknown category list answers 404"); + ok &= check(no_rosbag_list && no_such_list && no_rosbag_list->status == no_such_list->status && + error_code_of(no_rosbag_list) == error_code_of(no_such_list), + "listing rosbags without rosbags answers like an unknown category"); + auto no_rosbag_item = client.Get(base + "/rosbags/some-recording"); + auto no_such_item = client.Get(base + "/no_such_category/some-recording"); + ok &= check(no_rosbag_item && no_rosbag_item->body.find("Unknown category: rosbags") != std::string::npos, + "the rosbag download error names the unknown category"); + ok &= check(no_rosbag_item && no_such_item && no_rosbag_item->status == no_such_item->status && + error_code_of(no_rosbag_item) == error_code_of(no_such_item), + "downloading a rosbag without rosbags answers like an unknown category"); + auto root = client.Get("/api/v1/"); const auto caps = root ? nlohmann::json::parse(root->body, nullptr, false).value("capabilities", nlohmann::json::object()) diff --git a/src/ros2_medkit_gateway/README.md b/src/ros2_medkit_gateway/README.md index 2c2ddb672..b24178ed9 100644 --- a/src/ros2_medkit_gateway/README.md +++ b/src/ros2_medkit_gateway/README.md @@ -1764,7 +1764,7 @@ The file takes a target the including build already defines, else `find_package` A program that hosts the core gets the cyclic-subscription samplers and the bulk-data store from the core, and passes the following: - **Samplers.** `register_builtin_samplers(services)` (`core/builtin_samplers.hpp`) registers `data`, `faults`, `configurations` and `logs` on `services.sampler_registry()`, plus `updates` when `services.update_manager()` is set. Each sampler reads through the same managers and helpers as its GET route. A `data` subscription on an entity a plugin owns reads through that plugin's `DataProvider` (or its `x-plc-data` route), and on any other entity through `services.data_access_manager()` and its `TopicDataProvider`. A collection with no manager behind it answers with the error text of its GET route, not a crash. The host passes `GatewayServices` (it must outlive the samplers) and registers a subscription transport itself, for example `SseTransportProvider`. `GET /` reports `cyclic_subscriptions` true only when a transport and at least one sampler are registered. -- **Bulk data.** `BulkDataConfig` (`core/config.hpp`: `storage_dir`, `max_upload_size`, `categories`, with the defaults of the `bulk_data.*` parameters) and `make_bulk_data_store(config)` (`core/managers/bulk_data_store.hpp`) build the store. The host returns it from `GatewayServices::bulk_data_store()` before it starts `RESTServer`, which sizes the upload limit from it. The `rosbags` category is listed only when `FaultServiceTransport::serves_rosbags()` is true. The default is true; a host whose fault manager records no rosbags overrides it. `GET /` reports `bulk_data` true only when a store exists or rosbags are served. +- **Bulk data.** `BulkDataConfig` (`core/config.hpp`: `storage_dir`, `max_upload_size`, `categories`, with the defaults of the `bulk_data.*` parameters) and `make_bulk_data_store(config)` (`core/managers/bulk_data_store.hpp`) build the store. The host returns it from `GatewayServices::bulk_data_store()` before it starts `RESTServer`, which sizes the upload limit from it. The `rosbags` category is listed only when `FaultServiceTransport::serves_rosbags()` is true; otherwise a request for it answers 404 like any unknown category. The default is true; a host whose fault manager records no rosbags overrides it. `GET /` reports `bulk_data` true only when a store exists or rosbags are served. The `gateway_node` executable and existing test targets link `gateway_ros2`, so they transitively get both layers from a single dependency. The neutral contract is enforced by two CTest checks: diff --git a/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp index 818bdebfa..35f99d713 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp @@ -269,6 +269,14 @@ std::vector compute_bulkdata_source_filters(const ThreadSafeEntityC } // namespace detail +namespace { +/// Whether the fault manager serves the `rosbags` category. When it does not, the category is unknown. +bool rosbags_served(const HandlerContext & ctx) { + auto * fault_mgr = ctx.services().fault_manager(); + return fault_mgr != nullptr && fault_mgr->serves_rosbags(); +} +} // namespace + // --------------------------------------------------------------------------- // GET /{entity}/bulk-data - list categories // --------------------------------------------------------------------------- @@ -291,7 +299,7 @@ http::Result BulkDataHandlers::list_categories(const // Build categories list: "rosbags" when the fault manager serves them + BulkDataStore categories. dto::BulkDataCategoryList response; - if (auto * fault_mgr = ctx_.services().fault_manager(); fault_mgr != nullptr && fault_mgr->serves_rosbags()) { + if (rosbags_served(ctx_)) { response.items.push_back("rosbags"); } @@ -330,7 +338,7 @@ BulkDataHandlers::list_descriptors(const http::TypedRequest & req) { return tl::unexpected(make_error(400, ERR_INVALID_REQUEST, "Missing category")); } - if (category == "rosbags") { + if (category == "rosbags" && rosbags_served(ctx_)) { // === Rosbags: served via FaultManager === auto fault_mgr = ctx_.services().fault_manager(); @@ -433,7 +441,7 @@ http::Result BulkDataHandlers::download(const http::TypedR std::string mimetype; std::string filename; - if (category == "rosbags") { + if (category == "rosbags" && rosbags_served(ctx_)) { // === Rosbags: served via FaultManager === // // The id is a recording id. A pre-#620 URL carrying a fault code still resolves: From 530d51ccdc39e9ab42356b03560452550f42ba47 Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Thu, 5 Nov 2026 12:39:08 +0100 Subject: [PATCH 4/7] fix(gateway): decode the resource path of a cyclic subscription like a GET path A data subscription on a plugin item named with characters a client percent-encodes (a space, a quote) looked the item up by its encoded name and never found it. The resource path is now decoded with the function that decodes the GET path, before the '..' segment check. --- .../handlers/cyclic_subscription_handlers.cpp | 3 +- .../demo_nodes/test_data_provider_plugin.cpp | 3 + .../test_plugin_item_subscription.test.py | 146 ++++++++++++++++++ 3 files changed, 151 insertions(+), 1 deletion(-) create mode 100644 src/ros2_medkit_integration_tests/test/features/test_plugin_item_subscription.test.py diff --git a/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp index ef44efffb..64e8472a6 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp @@ -456,7 +456,8 @@ CyclicSubscriptionHandlers::parse_resource_uri(const std::string & resource) { parsed.entity_type = match[1].str(); parsed.entity_id = match[2].str(); parsed.collection = match[3].str(); - parsed.resource_path = match[4].matched ? match[4].str() : ""; + // Decoded as the GET path is, so a subscription reads the item GET reads. Before the '..' check. + parsed.resource_path = match[4].matched ? httplib::detail::decode_url(match[4].str(), false) : ""; // Security: reject '..' as a path segment (not as substring of e.g. '/..foo') if (!parsed.resource_path.empty()) { diff --git a/src/ros2_medkit_gateway/test/demo_nodes/test_data_provider_plugin.cpp b/src/ros2_medkit_gateway/test/demo_nodes/test_data_provider_plugin.cpp index ef8c841b2..c1a92d8ce 100644 --- a/src/ros2_medkit_gateway/test/demo_nodes/test_data_provider_plugin.cpp +++ b/src/ros2_medkit_gateway/test/demo_nodes/test_data_provider_plugin.cpp @@ -97,6 +97,9 @@ class TestDataProviderPlugin : public GatewayPlugin, {"unit", "degC"}}); items.push_back( {{"id", "pressure"}, {"name", "Pressure"}, {"category", "currentData"}, {"value", 3.2}, {"unit", "bar"}}); + // Names that need percent-encoding in a URI, like PLC symbols. + items.push_back({{"id", "\"DB1\".temp"}, {"name", "DB1 temp"}, {"category", "currentData"}, {"value", 7.5}}); + items.push_back({{"id", "Motor Speed"}, {"name", "Motor Speed"}, {"category", "currentData"}, {"value", 1450}}); return dto::DataListResult{nlohmann::json{{"items", std::move(items)}}}; } diff --git a/src/ros2_medkit_integration_tests/test/features/test_plugin_item_subscription.test.py b/src/ros2_medkit_integration_tests/test/features/test_plugin_item_subscription.test.py new file mode 100644 index 000000000..0d246d0fb --- /dev/null +++ b/src/ros2_medkit_integration_tests/test/features/test_plugin_item_subscription.test.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +# Copyright 2026 bburda +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""A data subscription on a plugin item reads the item GET of the same URI reads. + +The item names carry characters that a client percent-encodes in a URI path +(a space, a quote), like PLC symbols such as ``"DB1".temp``. +""" + +import json +import os +import unittest +from urllib.parse import quote + +from ament_index_python.packages import get_package_prefix +import launch_testing +import requests + +from ros2_medkit_test_utils.constants import ALLOWED_EXIT_CODES, API_BASE_PATH +from ros2_medkit_test_utils.gateway_test_case import GatewayTestCase +from ros2_medkit_test_utils.launch_helpers import create_test_launch + +PLUGIN_APP = 'test_plc_app' +PLAIN_ITEM = 'temperature' +QUOTED_ITEM = '"DB1".temp' +SPACED_ITEM = 'Motor Speed' + + +def _get_plugin_path(so_name): + pkg_prefix = get_package_prefix('ros2_medkit_gateway') + return os.path.join(pkg_prefix, 'lib', 'ros2_medkit_gateway', so_name) + + +def generate_test_description(): + return create_test_launch( + demo_nodes=[], + fault_manager=False, + gateway_params={ + 'plugins': ['test_data_provider'], + 'plugins.test_data_provider.path': + _get_plugin_path('libtest_data_provider_plugin.so'), + 'sse.max_clients': 10, + 'sse.max_subscriptions': 50, + }, + ) + + +class TestPluginItemSubscription(GatewayTestCase): + + MIN_EXPECTED_APPS = 1 + REQUIRED_APPS = {PLUGIN_APP} + + def _item_uri(self, item): + return f'{API_BASE_PATH}/apps/{PLUGIN_APP}/data/{quote(item, safe="")}' + + def _first_event(self, item, expect_status=201): + resource = self._item_uri(item) + r = requests.post( + f'{self.BASE_URL}/apps/{PLUGIN_APP}/cyclic-subscriptions', + json={'resource': resource, 'interval': 'fast', 'duration': 30}, + timeout=5, + ) + self.assertEqual(r.status_code, expect_status, r.text) + if expect_status != 201: + return r + body = r.json() + # The resource is echoed as the client sent it. + self.assertEqual(body['observed_resource'], resource) + sub_id = body['id'] + try: + with requests.get(self.BASE_URL.removesuffix(API_BASE_PATH) + + body['event_source'], stream=True, + timeout=15) as stream: + for line in stream.iter_lines(decode_unicode=True): + if line and line.startswith('data:'): + return json.loads(line[len('data:'):].strip()) + finally: + requests.delete( + f'{self.BASE_URL}/apps/{PLUGIN_APP}/cyclic-subscriptions/' + f'{sub_id}', timeout=5) + self.fail('SSE stream ended without an event') + + def _assert_event_matches_get(self, item): + get = requests.get( + f'{self.BASE_URL}/apps/{PLUGIN_APP}/data/{quote(item, safe="")}', + timeout=5) + self.assertEqual(get.status_code, 200, get.text) + event = self._first_event(item) + self.assertEqual( + event.get('payload', event), get.json(), + f'event {event} differs from GET {get.json()} for {item!r}') + + def test_plain_item_matches_get(self): + """Check one case. + + @verifies REQ_INTEROP_089 + """ + self._assert_event_matches_get(PLAIN_ITEM) + + def test_quoted_item_matches_get(self): + """Check one case. + + @verifies REQ_INTEROP_089 + """ + self._assert_event_matches_get(QUOTED_ITEM) + + def test_spaced_item_matches_get(self): + """Check one case. + + @verifies REQ_INTEROP_089 + """ + self._assert_event_matches_get(SPACED_ITEM) + + def test_encoded_dot_dot_segment_is_refused(self): + """Check one case. + + @verifies REQ_INTEROP_089 + """ + r = requests.post( + f'{self.BASE_URL}/apps/{PLUGIN_APP}/cyclic-subscriptions', + json={'resource': + f'{API_BASE_PATH}/apps/{PLUGIN_APP}/data/%2E%2E', + 'interval': 'fast', 'duration': 30}, + timeout=5, + ) + self.assertEqual(r.status_code, 400, r.text) + + +@launch_testing.post_shutdown_test() +class TestShutdown(unittest.TestCase): + + def test_exit_codes(self, proc_info): + for info in proc_info: + self.assertIn(info.returncode, ALLOWED_EXIT_CODES) From cf128c586fecf8cf71ae798a8f6f6919d8008287 Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Thu, 5 Nov 2026 12:40:40 +0100 Subject: [PATCH 5/7] docs(gateway): state that a subscription resource path is percent-decoded --- docs/api/rest.rst | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/api/rest.rst b/docs/api/rest.rst index 200795504..b854572bd 100644 --- a/docs/api/rest.rst +++ b/docs/api/rest.rst @@ -2158,7 +2158,8 @@ collection. - ``resource`` (string, required): Full SOVD resource URI to observe (e.g. ``/api/v1/apps/{id}/data/{topic}``, ``/api/v1/apps/{id}/faults``, ``/api/v1/functions/{id}/x-medkit-graph``). The URI ends at the collection - unless that collection's sampler narrows its payload to a named resource + unless that collection's sampler narrows its payload to a named resource. + The path is percent-decoded as a GET path is, so the subscription reads the item that GET of the same URI returns. - ``protocol`` (string, optional): Transport protocol. Only ``"sse"`` supported. Default: ``"sse"`` - ``interval`` (string, required): One of ``fast``, ``normal``, ``slow`` - ``duration`` (integer, required): Subscription lifetime in seconds. From 0454163c28d38ed0259925614151980d61f7e5a3 Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Thu, 5 Nov 2026 15:47:14 +0100 Subject: [PATCH 6/7] test(gateway): cover a percent-encoded topic URI in a cyclic subscription The documented encoded form of a topic path (powertrain%2Fengine%2Ftemperature) must deliver the sample GET of the same URI returns. --- ...est_multi_collection_subscriptions.test.py | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py b/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py index f5e09f946..26c20b002 100644 --- a/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py +++ b/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py @@ -27,6 +27,7 @@ import threading import time import unittest +from urllib.parse import quote import launch_testing import launch_testing.actions @@ -159,6 +160,37 @@ def test_data_subscription_create(self): self.assertIn('event_source', data) self.assertTrue(data['event_source'].endswith('/events')) + def test_encoded_topic_subscription_matches_get(self): + """A percent-encoded topic URI delivers the data GET of that URI returns. + + @verifies REQ_INTEROP_089 + """ + encoded = quote(self.topic_id.lstrip('/'), safe='') + resource = f'/api/v1/apps/{self.app_id}/data/{encoded}' + r = self._create_subscription(resource, interval='fast') + self.assertEqual(r.status_code, 201, f'Create failed: {r.text}') + body = r.json() + self.addCleanup(self._delete_subscription, body['id']) + self.assertEqual(body['observed_resource'], resource) + + event = None + with requests.get( + self.BASE_URL.removesuffix(API_BASE_PATH) + body['event_source'], + stream=True, timeout=15) as stream: + for line in stream.iter_lines(decode_unicode=True): + if line and line.startswith('data:'): + event = json.loads(line[len('data:'):].strip()) + break + self.assertIsNotNone(event, 'SSE stream ended without an event') + + get = requests.get(f'{self.BASE_URL}/apps/{self.app_id}/data/{encoded}', + timeout=5) + self.assertEqual(get.status_code, 200, get.text) + self.assertNotIn('error', event, f'Sample failed: {event}') + sample = event['payload'] + self.assertEqual(sample['id'], get.json()['id']) + self.assertEqual(set(sample['data']), set(get.json()['data'])) + def test_faults_subscription_create(self): """Create a subscription on faults collection. From b9995456ff98d4565c11bf2e329dbefd0991b212 Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Thu, 5 Nov 2026 17:02:51 +0100 Subject: [PATCH 7/7] fix(gateway): read subscription items the way GET reads them An update id in a status subscription is decoded and checked with the entity-id rules, as GET does. The data item is what follows the separator slash, so a topic encoded with its leading slash reads the topic GET reads. The resource field docs say a member-qualified id is not resolved. --- docs/api/rest.rst | 5 ++- .../src/core/builtin_samplers.cpp | 12 ++--- .../handlers/cyclic_subscription_handlers.cpp | 10 ++++- ...est_multi_collection_subscriptions.test.py | 23 +++++++--- .../test_update_status_subscription.test.py | 45 +++++++++++++++++++ 5 files changed, 82 insertions(+), 13 deletions(-) diff --git a/docs/api/rest.rst b/docs/api/rest.rst index b854572bd..1c8e0336f 100644 --- a/docs/api/rest.rst +++ b/docs/api/rest.rst @@ -2159,7 +2159,10 @@ collection. (e.g. ``/api/v1/apps/{id}/data/{topic}``, ``/api/v1/apps/{id}/faults``, ``/api/v1/functions/{id}/x-medkit-graph``). The URI ends at the collection unless that collection's sampler narrows its payload to a named resource. - The path is percent-decoded as a GET path is, so the subscription reads the item that GET of the same URI returns. + The item path is percent-decoded as a GET path is, so an encoded topic or + plugin item name addresses the item GET of the same URI reads. A + member-qualified id (``member:/topic``) is not resolved by a subscription; + use the bare item id. - ``protocol`` (string, optional): Transport protocol. Only ``"sse"`` supported. Default: ``"sse"`` - ``interval`` (string, required): One of ``fast``, ``normal``, ``slow`` - ``duration`` (integer, required): Subscription lifetime in seconds. diff --git a/src/ros2_medkit_gateway/src/core/builtin_samplers.cpp b/src/ros2_medkit_gateway/src/core/builtin_samplers.cpp index 71163b93e..5812c999c 100644 --- a/src/ros2_medkit_gateway/src/core/builtin_samplers.cpp +++ b/src/ros2_medkit_gateway/src/core/builtin_samplers.cpp @@ -41,28 +41,30 @@ namespace { using Sampled = tl::expected; Sampled sample_data(GatewayServices & services, const std::string & entity_id, const std::string & resource_path) { + // The item is what follows the separator slash, as in GET /{entity}/data/{item}. + const std::string item_name = + (!resource_path.empty() && resource_path.front() == '/') ? resource_path.substr(1) : resource_path; // A plugin-owned entity is read through its provider, as GET /{entity}/data/{item} does. auto * plugins = services.plugin_manager(); if (plugins != nullptr && plugins->get_entity_owner(entity_id)) { - const std::string item_name = - (!resource_path.empty() && resource_path.front() == '/') ? resource_path.substr(1) : resource_path; auto item = read_plugin_data_item(plugins, entity_id, item_name); if (!item) { return tl::make_unexpected(item.error().message); } return std::move(item->content); } - auto sample = sample_data_topic(*services.data_access_manager(), to_full_topic_path(resource_path)); + const std::string topic_path = to_full_topic_path(item_name); + auto sample = sample_data_topic(*services.data_access_manager(), topic_path); if (!sample) { return tl::make_unexpected(std::string{"Topic sample failed: "} + sample.error().message); } if (sample->has_data && sample->data.has_value()) { nlohmann::json payload; - payload["id"] = resource_path; + payload["id"] = topic_path; payload["data"] = *sample->data; return payload; } - return tl::make_unexpected("Topic data not available: " + resource_path); + return tl::make_unexpected("Topic data not available: " + topic_path); } Sampled sample_faults(GatewayServices & services, const std::string & entity_id) { diff --git a/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp index 64e8472a6..98905de6d 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp @@ -177,6 +177,14 @@ CyclicSubscriptionHandlers::post_subscription(const http::TypedRequest & req, } } + // GET refuses an update id that fails the entity-id rules; so does a subscription. + if (parsed->collection == "updates") { + if (auto id_ok = ctx_.validate_entity_id(parsed->resource_path); !id_ok) { + return tl::unexpected(make_error(400, ERR_X_MEDKIT_INVALID_RESOURCE_URI, id_ok.error(), + json{{"parameter", "resource"}, {"value", resource}})); + } + } + // Check sampler is registered if (!sampler_registry_.has_sampler(parsed->collection)) { return tl::unexpected(make_error(400, ERR_X_MEDKIT_COLLECTION_NOT_AVAILABLE, @@ -483,7 +491,7 @@ CyclicSubscriptionHandlers::parse_resource_uri(const std::string & resource) { parsed.entity_type = ""; // server-level, no entity parsed.entity_id = ""; parsed.collection = "updates"; - parsed.resource_path = match[1].str(); // update package ID + parsed.resource_path = httplib::detail::decode_url(match[1].str(), false); // update package ID return parsed; } diff --git a/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py b/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py index 26c20b002..befb3ffce 100644 --- a/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py +++ b/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py @@ -160,12 +160,8 @@ def test_data_subscription_create(self): self.assertIn('event_source', data) self.assertTrue(data['event_source'].endswith('/events')) - def test_encoded_topic_subscription_matches_get(self): - """A percent-encoded topic URI delivers the data GET of that URI returns. - - @verifies REQ_INTEROP_089 - """ - encoded = quote(self.topic_id.lstrip('/'), safe='') + def _assert_encoded_topic_matches_get(self, encoded): + """Check that a subscription on an encoded topic URI delivers what GET returns.""" resource = f'/api/v1/apps/{self.app_id}/data/{encoded}' r = self._create_subscription(resource, interval='fast') self.assertEqual(r.status_code, 201, f'Create failed: {r.text}') @@ -191,6 +187,21 @@ def test_encoded_topic_subscription_matches_get(self): self.assertEqual(sample['id'], get.json()['id']) self.assertEqual(set(sample['data']), set(get.json()['data'])) + def test_encoded_topic_subscription_matches_get(self): + """Topic path encoded without its leading slash. + + @verifies REQ_INTEROP_089 + """ + self._assert_encoded_topic_matches_get( + quote(self.topic_id.lstrip('/'), safe='')) + + def test_encoded_topic_with_leading_slash_matches_get(self): + """Topic id encoded as listed, leading slash included. + + @verifies REQ_INTEROP_089 + """ + self._assert_encoded_topic_matches_get(quote(self.topic_id, safe='')) + def test_faults_subscription_create(self): """Create a subscription on faults collection. diff --git a/src/ros2_medkit_integration_tests/test/features/test_update_status_subscription.test.py b/src/ros2_medkit_integration_tests/test/features/test_update_status_subscription.test.py index 3251bcd6f..2bb918890 100644 --- a/src/ros2_medkit_integration_tests/test/features/test_update_status_subscription.test.py +++ b/src/ros2_medkit_integration_tests/test/features/test_update_status_subscription.test.py @@ -195,6 +195,51 @@ def test_02_subscription_listed_for_entity(self): sub_ids = [s['id'] for s in items] self.assertIn(sub_id, sub_ids) + def _create_update_subscription(self, update_id_segment): + return requests.post( + f'{self.BASE_URL}/apps/{self.app_id}/cyclic-subscriptions', + json={ + 'resource': f'/api/v1/updates/{update_id_segment}/status', + 'interval': 'fast', + 'duration': 60, + }, + timeout=5, + ) + + def test_02b_encoded_update_id_matches_get(self): + """A percent-encoded update id delivers the status GET returns.""" + self._register_package('sub-enc-pkg') + encoded = '%73ub-enc-pkg' + r = self._create_update_subscription(encoded) + self.assertEqual(r.status_code, 201, f'Create failed: {r.text}') + body = r.json() + self.addCleanup(self._cleanup_subscription, body['id']) + self.assertEqual( + body['observed_resource'], f'/api/v1/updates/{encoded}/status') + + event = None + with requests.get(f'http://127.0.0.1:{PORT}{body["event_source"]}', + stream=True, timeout=15) as stream: + for line in stream.iter_lines(decode_unicode=True): + if line and line.startswith('data:'): + event = json.loads(line[len('data:'):].strip()) + break + self.assertIsNotNone(event, 'SSE stream ended without an event') + self.assertNotIn('error', event, f'Sample failed: {event}') + + get = requests.get( + f'{self.BASE_URL}/updates/{encoded}/status', timeout=5) + self.assertEqual(get.status_code, 200, get.text) + self.assertEqual(event['payload'], get.json()) + + def test_02c_update_id_refused_by_get_is_refused(self): + """An id GET refuses (here '..' hidden as %2E%2E) is refused at create.""" + get = requests.get( + f'{self.BASE_URL}/updates/%2E%2E/status', timeout=5) + self.assertEqual(get.status_code, 400, get.text) + r = self._create_update_subscription('%2E%2E') + self.assertEqual(r.status_code, 400, r.text) + def test_03_sse_stream_delivers_update_progress(self): """SSE stream delivers status snapshots during update prepare.""" self._register_package('sub-sse-pkg')