diff --git a/docs/config/server.rst b/docs/config/server.rst index 42d11c3ab..3b56e4a4c 100644 --- a/docs/config/server.rst +++ b/docs/config/server.rst @@ -846,6 +846,13 @@ Example: storage: path: "/var/lib/ros2_medkit/triggers.db" +A program that hosts the gateway core instead of the ROS 2 node sets the same +values in ``TriggerHostConfig`` (``enabled``, ``max_triggers``, +``on_restart_behavior``, ``storage_path``). It has one more setting, which the +node does not read: ``data_poll_interval_ms`` (default ``1000``, clamped to +``100`` - ``60000`` with a warning) is the period at which a data trigger on an +item a plugin serves reads that item. A trigger has no interval of its own. + Fault-trigger rules ^^^^^^^^^^^^^^^^^^^ diff --git a/scripts/check_cores_without_ros/README.md b/scripts/check_cores_without_ros/README.md index bec6f33b1..9b96aa21e 100644 --- a/scripts/check_cores_without_ros/README.md +++ b/scripts/check_cores_without_ros/README.md @@ -26,6 +26,11 @@ not installed, so the PostgreSQL backend is not built. evicted past `max_faults`, a re-confirmation replaces the frame, and a disabled configuration builds no capture. - Fault event fan-out: one upstream source shared through `FaultEventFanOut` by the fault SSE stream (read over HTTP) and the capture: a confirmation shows up as an SSE event and as a freeze frame. Without the fan-out one source serves one consumer only. +- Triggers: a host built by `make_trigger_host()` over the fan-out of the single fault event source. A fault trigger created over + HTTP fires on a confirmation, its event is read from the trigger's event stream, and the fault SSE stream still gets the + confirmation. A data trigger on a plugin-served item does not fire while the value is below its range, fires once when the value + enters it, and the item is read again and again by the core's polling source. A fourth trigger with `max_triggers = 3` answers + 503. A limit below 1 or past `INT_MAX` and a polling interval outside 100-60000 ms fall back with a notice. - 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 diff --git a/scripts/check_cores_without_ros/main.cpp b/scripts/check_cores_without_ros/main.cpp index 193c803a1..842d80f1e 100644 --- a/scripts/check_cores_without_ros/main.cpp +++ b/scripts/check_cores_without_ros/main.cpp @@ -69,10 +69,12 @@ #include "ros2_medkit_gateway/core/managers/operation_manager.hpp" #include "ros2_medkit_gateway/core/managers/script_manager.hpp" #include "ros2_medkit_gateway/core/managers/subscription_manager.hpp" +#include "ros2_medkit_gateway/core/managers/trigger_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_change_notifier.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" @@ -80,6 +82,7 @@ #include "ros2_medkit_gateway/core/transports/parameter_transport.hpp" #include "ros2_medkit_gateway/core/transports/service_transport.hpp" #include "ros2_medkit_gateway/core/transports/topic_transport.hpp" +#include "ros2_medkit_gateway/core/trigger_host.hpp" #include "ros2_medkit_gateway/core/types/type_schema_provider.hpp" namespace core = ros2_medkit_fault_reporter::core; @@ -281,7 +284,14 @@ class CheckDataPlugin : public gw::GatewayPlugin, public gw::DataProvider { } 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}}}; + ++read_calls; + const int pinned_value = pinned.load(); + const int value = pinned_value >= 0 ? pinned_value : ++reads; + if (wrapped) { + return gw::dto::DataValue{ + {{"id", resource_name}, {"data", {{"value", value}}}, {"x-medkit", {{"source", "plugin"}}}}}; + } + return gw::dto::DataValue{{{"source", "plugin"}, {"name", resource_name}, {"value", value}}}; } tl::expected write_data(const std::string & /*entity_id*/, const std::string & /*resource_name*/, @@ -290,6 +300,11 @@ class CheckDataPlugin : public gw::GatewayPlugin, public gw::DataProvider { } std::atomic reads{0}; std::atomic speed{10}; + // At or above zero every read answers this value instead of the counter. + std::atomic pinned{-1}; + std::atomic read_calls{0}; + // True: reads answer the body of a topic item, {id, data, x-medkit}, with the value inside data. + std::atomic wrapped{false}; }; // A fault event feed the check drives by hand. @@ -540,7 +555,7 @@ class MinimalServices : public gw::GatewayServices { return nullptr; } gw::TriggerManager * trigger_manager() override { - return nullptr; + return triggers_; } gw::ThreadSafeEntityCache & mutable_cache() { @@ -573,6 +588,9 @@ class MinimalServices : public gw::GatewayServices { void set_script_manager(gw::ScriptManager * scripts) { scripts_ = scripts; } + void set_trigger_manager(gw::TriggerManager * triggers) { + triggers_ = triggers; + } private: gw::ThreadSafeEntityCache cache_; @@ -585,6 +603,7 @@ class MinimalServices : public gw::GatewayServices { gw::FaultEventSource * fault_events_{nullptr}; gw::LockManager * locks_{nullptr}; gw::ScriptManager * scripts_{nullptr}; + gw::TriggerManager * triggers_{nullptr}; gw::FaultManager fault_; gw::ConfigurationManager config_; gw::DataAccessManager data_; @@ -627,6 +646,9 @@ class RunningServer { if (aggregation != nullptr) { server_->set_aggregation_manager(aggregation); } + if (auto * triggers = services.trigger_manager()) { + server_->set_trigger_handlers(*triggers); + } thread_ = std::thread([this] { server_->start(); }); @@ -988,6 +1010,210 @@ bool check_lifecycle_without_reader() { return ok; } +// Collects the events of one trigger's SSE stream until the stream ends. +class TriggerEventReader { + public: + TriggerEventReader(int port, const std::string & path) { + thread_ = std::thread([this, port, path] { + httplib::Client client("127.0.0.1", port); + client.set_connection_timeout(5); + client.set_read_timeout(60); + 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); + const auto at = frame.find("data: "); + if (at != std::string::npos) { + std::lock_guard lock(mutex_); + events_.push_back(nlohmann::json::parse(frame.substr(at + 6), nullptr, false)); + } + } + return true; + }); + }); + } + // The stream ends when the trigger is deleted. + void join() { + if (thread_.joinable()) { + thread_.join(); + } + } + ~TriggerEventReader() { + join(); + } + TriggerEventReader(const TriggerEventReader &) = delete; + TriggerEventReader & operator=(const TriggerEventReader &) = delete; + TriggerEventReader(TriggerEventReader &&) = delete; + TriggerEventReader & operator=(TriggerEventReader &&) = delete; + std::vector events() { + std::lock_guard lock(mutex_); + return events_; + } + + private: + std::mutex mutex_; + std::vector events_; + std::thread thread_; +}; + +httplib::Result post_trigger(httplib::Client & client, const std::string & resource, const nlohmann::json & condition, + const std::string & path = "") { + nlohmann::json body = {{"resource", resource}, {"trigger_condition", condition}, {"multishot", true}}; + if (!path.empty()) { + body["path"] = path; + } + return client.Post("/api/v1/apps/plugin_app/triggers", body.dump(), "application/json"); +} + +// Triggers on a host with no ROS node: a fault confirmation from the host's single event source (shared with the fault +// SSE stream through the fan-out) fires a fault trigger, a plugin-served data item fires a data trigger when it crosses +// a threshold, and the trigger count is bounded. +bool check_triggers() { + bool ok = true; + MinimalServices services; + auto plugin_app = hosted_app(); + plugin_app.id = "plugin_app"; + services.mutable_cache().update_apps({plugin_app}); + auto & plugin = services.add_data_plugin({"plugin_app"}); + plugin.pinned = 10; + plugin.wrapped = true; + ControlledFaultEvents upstream; + gw::FaultEventFanOut fan_out(upstream); + auto sse_branch = fan_out.make_branch(); + services.set_fault_events(sse_branch.get()); + gw::ResourceChangeNotifier notifier; + gw::ConditionRegistry conditions; + gw::register_builtin_conditions(conditions); + gw::TriggerHostConfig config; + config.max_triggers = 3; + config.data_poll_interval_ms = 100; + gw::TriggerHostWiring wiring; + wiring.cache = &services.entity_cache(); + wiring.polled_plugins = services.plugin_manager(); + auto trigger_branch = fan_out.make_branch(); + wiring.fault_events = trigger_branch.get(); + auto host = gw::make_trigger_host(config, notifier, conditions, wiring); + ok &= check(host.max_triggers == 3 && host.data_poll_interval_ms == 100, + "the host builds with the configured trigger limit and polling interval"); + { + // Settings outside their documented range fall back with a notice; a disabled host builds nothing. + std::vector notices; + auto wiring_with_notices = wiring; + wiring_with_notices.fault_events = nullptr; + wiring_with_notices.warn = [¬ices](const std::string & message) { + notices.push_back(message); + }; + gw::ResourceChangeNotifier other_notifier; + gw::ConditionRegistry other_conditions; + gw::TriggerHostConfig out_of_range; + out_of_range.max_triggers = 0; + out_of_range.data_poll_interval_ms = 5; + auto low = gw::make_trigger_host(out_of_range, other_notifier, other_conditions, wiring_with_notices); + ok &= check(low.max_triggers == 1000 && low.data_poll_interval_ms == 100 && notices.size() == 2, + "a limit below 1 uses 1000 and an interval below 100 ms is raised to 100 ms, each with a notice"); + out_of_range.max_triggers = 5000000000LL; + out_of_range.data_poll_interval_ms = 1000000000LL; + auto high = gw::make_trigger_host(out_of_range, other_notifier, other_conditions, wiring_with_notices); + ok &= check(high.max_triggers == 1000 && high.data_poll_interval_ms == 60000 && notices.size() == 4, + "a limit past what an int holds uses 1000 and an interval past 60 s is lowered to 60 s"); + out_of_range.enabled = false; + ok &= check(!gw::make_trigger_host(out_of_range, other_notifier, other_conditions, wiring_with_notices).manager, + "a disabled configuration builds no trigger manager"); + } + services.set_trigger_manager(host.manager.get()); + RunningServer running(services); + auto client = make_client(running.port()); + + // Fault trigger. + auto fault_trigger = post_trigger(client, "/api/v1/apps/plugin_app/faults", {{"condition_type", "OnChange"}}); + ok &= check(fault_trigger && fault_trigger->status == 201, "a fault trigger is created over HTTP"); + const auto fault_json = fault_trigger ? nlohmann::json::parse(fault_trigger->body, nullptr, false) : nlohmann::json(); + const std::string fault_id = fault_json.is_object() ? fault_json.value("id", "") : ""; + const std::string fault_events_path = fault_json.is_object() ? fault_json.value("event_source", "") : ""; + TriggerEventReader fault_reader(running.port(), fault_events_path); + std::atomic sse_seen{false}; + std::atomic stream_done{false}; + std::thread fault_stream([&] { + auto stream_client = make_client(running.port()); + std::string buffer; + stream_client.Get( + "/api/v1/faults/stream", httplib::Headers{}, + [](const httplib::Response &) { + return true; + }, + [&](const char * data, size_t len) { + buffer.append(data, len); + if (buffer.find("TRIGGER_FAULT") != std::string::npos) { + sse_seen = true; + } + return !sse_seen && !stream_done; + }); + }); + for (int i = 0; i < 60 && !sse_seen; ++i) { + upstream.confirm("TRIGGER_FAULT", "plugin_app"); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + stream_done = true; + fault_stream.join(); + ok &= check(sse_seen.load(), "the fault SSE stream still gets the confirmation"); + const bool fault_fired = wait_until([&] { + return !fault_reader.events().empty(); + }); + ok &= check(fault_fired, "a confirmation from the host's event source fires the fault trigger"); + if (fault_fired) { + const auto event = fault_reader.events().front(); + ok &= check(event.contains("payload") && event["payload"].value("fault_code", "") == "TRIGGER_FAULT" && + event["payload"].value("event_type", "") == "fault_confirmed", + "the trigger event carries the confirmed fault"); + } + if (!fault_id.empty()) { + client.Delete("/api/v1/apps/plugin_app/triggers/" + fault_id); + } + fault_reader.join(); + + // Data trigger on a plugin-served item. + auto data_trigger = + post_trigger(client, "/api/v1/apps/plugin_app/data/speed", + {{"condition_type", "EnterRange"}, {"lower_bound", 50}, {"upper_bound", 100}}, "/value"); + ok &= check(data_trigger && data_trigger->status == 201, "a data trigger on a plugin-served item is created"); + const auto data_json = data_trigger ? nlohmann::json::parse(data_trigger->body, nullptr, false) : nlohmann::json(); + const std::string data_id = data_json.is_object() ? data_json.value("id", "") : ""; + TriggerEventReader data_reader(running.port(), data_json.is_object() ? data_json.value("event_source", "") : ""); + const bool polled = wait_until([&] { + return plugin.read_calls.load() >= 5; + }); + ok &= check(polled, "the host reads the data item again and again"); + ok &= check(data_reader.events().empty(), "the data trigger does not fire while the value is below the range"); + plugin.pinned = 60; + const bool data_fired = wait_until([&] { + return !data_reader.events().empty(); + }); + ok &= check(data_fired, "the data trigger fires when the value enters the range"); + std::this_thread::sleep_for(std::chrono::milliseconds(1500)); + ok &= check(data_reader.events().size() == 1, "the data trigger fires once, not on every sample"); + if (!data_id.empty()) { + client.Delete("/api/v1/apps/plugin_app/triggers/" + data_id); + } + data_reader.join(); + + // The count limit. + std::vector ids; + for (int i = 0; i < 3; ++i) { + auto created = post_trigger(client, "/api/v1/apps/plugin_app/faults", {{"condition_type", "OnChange"}}); + ok &= check(created && created->status == 201, "a trigger below the limit is created"); + } + auto refused = post_trigger(client, "/api/v1/apps/plugin_app/faults", {{"condition_type", "OnChange"}}); + ok &= check(refused && refused->status == 503, "a trigger past triggers.max_triggers is refused with 503"); + return ok; +} + } // namespace static int run() { @@ -1571,6 +1797,7 @@ static int run() { ok &= check_script_host(); ok &= check_aggregation_host(); ok &= check_lifecycle_without_reader(); + ok &= check_triggers(); return ok ? 0 : 1; } diff --git a/scripts/check_no_naked_subscriptions.sh b/scripts/check_no_naked_subscriptions.sh index 0215dabeb..835f4c094 100755 --- a/scripts/check_no_naked_subscriptions.sh +++ b/scripts/check_no_naked_subscriptions.sh @@ -46,7 +46,6 @@ ALLOWED_DIRS_PATTERN="(${GATEWAY_ROOT}/src/ros2_common/|${GATEWAY_ROOT}/include/ # Keep this list SHORT and add a TODO in the file referencing the follow-up. ALLOWED_LEGACY_FILES=( "${GATEWAY_ROOT}/src/ros2/ros2_fault_event_source.cpp" # faults provider follow-up - "${GATEWAY_ROOT}/src/trigger_fault_subscriber.cpp" # faults provider follow-up "${GATEWAY_ROOT}/src/ros2/transports/ros2_action_transport.cpp" # operations provider follow-up "${GATEWAY_ROOT}/src/ros2/transports/ros2_log_source.cpp" # /rosout adapter (LogSource port) "${FAULT_MANAGER_ROOT}/src/snapshot_capture.cpp" # uses LockedSubscriptionGuard (in-place serialisation) diff --git a/src/ros2_medkit_gateway/README.md b/src/ros2_medkit_gateway/README.md index c4f496c9c..c039c170e 100644 --- a/src/ros2_medkit_gateway/README.md +++ b/src/ros2_medkit_gateway/README.md @@ -1769,6 +1769,7 @@ A program that hosts the core gets the cyclic-subscription samplers and the bulk - **Lock expiry.** `LockReaper(locks, subscriptions, interval_sec, logger)` (`core/managers/lock_reaper.hpp`) calls `LockManager::cleanup_expired()` every `interval_sec` (`LockConfig::cleanup_interval`, default 30, raised to 1 when lower) on a thread it owns. For each expired lock whose scope is empty or includes `cyclic-subscriptions` it removes the cyclic subscriptions of the locked entity. `subscriptions` may be null. The destructor and `stop()` join the thread. Without a reaper a host keeps expired locks until it calls `cleanup_expired()` itself. The ROS 2 node runs the same class. - **Scripts.** `make_script_host(config, plugin_backend)` (`core/managers/script_manager.hpp`) returns a `ScriptManager` and, when no plugin backend is given and `ScriptsConfig::scripts_dir` is not empty, the `DefaultScriptProvider` (fork/execvp) behind it. `ScriptsConfig` (`core/script_types.hpp`) has the defaults of the `scripts.*` parameters, and `entries` carries predefined scripts. The host returns `host.manager` from `GatewayServices::script_manager()` and keeps the `ScriptHost` alive as long as the services. With neither a backend nor a directory the script routes answer 501. - **Aggregation.** `AggregationHost(config, listen_port, mdns_name, logger)` (`core/aggregation/aggregation_host.hpp`) builds the `AggregationManager` from an `AggregationConfig` (the defaults of the `aggregation.*` parameters; the host range-checks values it reads from its own configuration) and, when `announce` or `discover` is set, the mDNS discovery that adds and removes peers and ignores this gateway's own announcements. The host passes `host.manager()` to `RESTServer::set_aggregation_manager()` and calls `host.merge_peers(areas, components, apps, functions)` on every entity refresh, before it fills the entity cache. The call merges the peers' entities into the four vectors and publishes the routing table, the leaf warnings and the per-entity peer contributors. It does nothing while there is no peer. Call `host.shutdown()` before `RESTServer::stop()`. +- **Triggers.** `make_trigger_host(config, notifier, conditions, wiring)` (`core/trigger_host.hpp`) builds the `SqliteTriggerStore` and the `TriggerManager`, wires the entity hierarchy, entity existence and topic lookups to `wiring.cache`, and restores persistent triggers. `TriggerHostConfig` has the names and defaults of the `triggers.*` parameters (`enabled`, `max_triggers`, `on_restart_behavior`, `storage_path`), and `data_poll_interval_ms` (default 1000, clamped to 100-60000 ms), which the node does not read. `register_builtin_conditions()` fills the `ConditionRegistry`. The host passes the manager to `RESTServer::set_trigger_handlers()` and returns it from `GatewayServices::trigger_manager()`. A trigger past `max_triggers` answers 503. Where its events come from is chosen by the wiring: `fault_events` (a `FaultEventSource`, for a host with one upstream source a `FaultEventFanOut` branch of its own) starts a `TriggerFaultFeed` that turns each fault event into a `faults` notification, and a fault trigger fires on it. A data trigger reads from `topic_transport` (a `TopicSubscriptionTransport`, the ROS 2 node's choice) when its topic is known, else from `polled_plugins`: a `PollingDataItemSource` that reads the item through `read_plugin_data_item`, as `GET /{entity}/data/{item}` does once per `data_poll_interval_ms`, starting at once. It delivers the `data` field of the body (the whole body when there is none), so a trigger `path` is relative to the value, as it is for a topic. A host without topics sets only `polled_plugins`; a host with neither gets data triggers that never fire. The trigger's condition runs on every sample, so `EnterRange` fires once at the crossing and `OnChange` at each difference. An item that cannot be read in a round delivers nothing in that round. Keep the host alive as long as the services. The ROS 2 node uses the same builder with its topic transport and feeds fault triggers through `TriggerFaultSubscriber`, a `Ros2FaultEventSource` under a `TriggerFaultFeed`. - **Lifecycle.** `RESTServer` takes a `LifecycleStateReader` or null. A host without one passes null. `GET /{entity}/status` then reports `ready` for an online app and `notReady` for an offline one, also when the app exposes the lifecycle services, because graph presence is the only signal left. A component is ready unless every app it hosts is offline. `PUT /{entity}/status/{action}` answers 501 unless a plugin `LifecycleProvider` serves the entity. 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/design/index.rst b/src/ros2_medkit_gateway/design/index.rst index 2e555a358..236471afd 100644 --- a/src/ros2_medkit_gateway/design/index.rst +++ b/src/ros2_medkit_gateway/design/index.rst @@ -87,6 +87,19 @@ store from it: The host passes the manager to ``RESTServer::set_aggregation_manager()`` and calls ``merge_peers()`` on every entity refresh. The ROS 2 node reads the ``aggregation.*`` parameters into the config and uses the same class. +* ``make_trigger_host`` (``core/trigger_host.hpp``) builds the trigger store and + the ``TriggerManager`` from a ``TriggerHostConfig`` (the ``triggers.*`` + names and defaults, plus ``data_poll_interval_ms``), wires the entity lookups + to the entity cache and restores persistent triggers. Fault triggers are fed + by a ``TriggerFaultFeed`` over a ``FaultEventSource`` (a ``FaultEventFanOut`` + branch when the source is shared). A data trigger with a known topic reads + from the ``TopicSubscriptionTransport``. Any other data trigger reads from a + ``DataItemSource``; ``PollingDataItemSource`` reads the item through + ``read_plugin_data_item`` once per ``data_poll_interval_ms`` and delivers the + ``data`` field of the item body, as a topic trigger sees the message, so a host whose + data a plugin serves needs no topics. The ROS 2 node passes the topic + transport and feeds fault triggers through ``TriggerFaultSubscriber``, a ROS + 2 event source under the same feed. * ``RESTServer`` takes a null ``LifecycleStateReader`` for a host without lifecycle nodes. ``GET /{entity}/status`` then derives readiness from presence and ``PUT /{entity}/status/{action}`` answers 501 without a diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/data/polling_data_item_source.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/data/polling_data_item_source.hpp new file mode 100644 index 000000000..4a95e1e70 --- /dev/null +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/data/polling_data_item_source.hpp @@ -0,0 +1,58 @@ +// 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 + +#include +#include + +#include "ros2_medkit_gateway/core/transports/data_item_source.hpp" + +namespace ros2_medkit_gateway { + +class PluginManager; + +/// A `DataItemSource` that reads each watched item from one thread, once per interval. +/// +/// A new watch is read at once. A read that fails delivers nothing and is tried again at the next interval. +class PollingDataItemSource : public DataItemSource { + public: + using ReadFn = std::function(const std::string & entity_id, + const std::string & item_name)>; + + PollingDataItemSource(ReadFn read, std::chrono::milliseconds interval); + ~PollingDataItemSource() override; + + std::unique_ptr watch(const std::string & entity_id, const std::string & item_name, + SampleCallback callback) override; + + private: + struct Shared; + class Handle; + + std::shared_ptr shared_; +}; + +/// A polling source over `read_plugin_data_item`. The sample is the `data` field of what +/// `GET /{entity}/data/{item}` answers for an entity a plugin serves, which is what a topic trigger sees: the message. +/// A body with no `data` field is the sample as it is. `plugins` must outlive the source. +std::unique_ptr make_plugin_data_item_source(PluginManager * plugins, + std::chrono::milliseconds interval); + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/faults/trigger_fault_feed.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/faults/trigger_fault_feed.hpp new file mode 100644 index 000000000..6c4e7823d --- /dev/null +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/faults/trigger_fault_feed.hpp @@ -0,0 +1,54 @@ +// 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 "ros2_medkit_gateway/core/faults/fault_event_source.hpp" +#include "ros2_medkit_gateway/core/resource_change_notifier.hpp" + +namespace ros2_medkit_gateway { + +/// Turns the events of a `FaultEventSource` into "faults" notifications for the trigger manager. +/// +/// The entity of an event is the first reporting source, mapped by the resolver when it knows the source and by its +/// last path segment otherwise. The resource path is "/" plus the fault code; the value is the fault JSON plus +/// `event_type`. A host gives it its own branch of a `FaultEventFanOut` when other consumers share the source. +class TriggerFaultFeed { + public: + /// Maps a reporting source to an entity id. Returns an empty string for a source it does not know. + using NodeToEntityFn = std::function; + + /// Subscribes to `source` at once. `source` and `notifier` must outlive the feed. + TriggerFaultFeed(FaultEventSource & source, ResourceChangeNotifier & notifier, NodeToEntityFn resolver = {}); + ~TriggerFaultFeed(); + TriggerFaultFeed(const TriggerFaultFeed &) = delete; + TriggerFaultFeed & operator=(const TriggerFaultFeed &) = delete; + TriggerFaultFeed(TriggerFaultFeed &&) = delete; + TriggerFaultFeed & operator=(TriggerFaultFeed &&) = delete; + + /// Replaces the resolver. Call it before events can arrive: it is not synchronised with delivery. + void set_node_to_entity_resolver(NodeToEntityFn resolver); + + private: + void on_event(const ros2_medkit::fault_detection::FaultEventRecord & event); + + FaultEventSource & source_; + ResourceChangeNotifier & notifier_; + NodeToEntityFn resolver_; +}; + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/trigger_manager.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/trigger_manager.hpp index 82218ab61..f54a6f1bb 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/trigger_manager.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/trigger_manager.hpp @@ -32,6 +32,7 @@ #include "ros2_medkit_gateway/core/condition_evaluator.hpp" #include "ros2_medkit_gateway/core/resource_change_notifier.hpp" +#include "ros2_medkit_gateway/core/transports/data_item_source.hpp" #include "ros2_medkit_gateway/core/transports/topic_subscription_transport.hpp" #include "ros2_medkit_gateway/core/trigger_store.hpp" @@ -91,8 +92,12 @@ class TriggerManager { /// topic subscriptions. When null the manager retains /// its CRUD/dispatch behaviour but never subscribes /// (useful in tests that don't need topic routing). + /// @param data_source Optional source for data triggers that have no topic. With a transport and a source, a + /// trigger whose topic is known uses the transport and every other data trigger uses the + /// source. With a source alone, every data trigger uses it. TriggerManager(ResourceChangeNotifier & notifier, ConditionRegistry & conditions, TriggerStore & store, - const TriggerConfig & config, std::shared_ptr topic_transport = nullptr); + const TriggerConfig & config, std::shared_ptr topic_transport = nullptr, + std::shared_ptr data_source = nullptr); ~TriggerManager(); // Non-copyable, non-movable (owns trigger state and notifier subscription) @@ -290,6 +295,9 @@ class TriggerManager { // the rclcpp resources; the manager only stores per-trigger handles. std::shared_ptr topic_transport_; + // Data trigger item source (shared, optional). Handles live in topic_handles_ with the topic ones. + std::shared_ptr data_source_; + // Per-trigger live subscription handles. Destruction of a handle // unsubscribes the underlying topic; the manager removes entries from // this map on remove()/cleanup_expired_trigger() so the dtor side-effect diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/data_item_source.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/data_item_source.hpp new file mode 100644 index 000000000..92d717adc --- /dev/null +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/data_item_source.hpp @@ -0,0 +1,45 @@ +// 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 "ros2_medkit_gateway/core/transports/topic_subscription_transport.hpp" + +namespace ros2_medkit_gateway { + +/// Delivers the value of one data item of an entity to a callback, repeatedly. +/// +/// For a host whose data is served by a plugin and not by topics. The +/// `TriggerManager` uses it for data triggers that have no topic. +class DataItemSource { + public: + using SampleCallback = TopicSubscriptionTransport::SampleCallback; + + DataItemSource() = default; + DataItemSource(const DataItemSource &) = delete; + DataItemSource & operator=(const DataItemSource &) = delete; + DataItemSource(DataItemSource &&) = delete; + DataItemSource & operator=(DataItemSource &&) = delete; + virtual ~DataItemSource() = default; + + /// Start delivering `item_name` of `entity_id` to `callback`. Returns nullptr when the source cannot watch it. + /// Destroying the handle stops the delivery; no callback runs once the destructor has returned. + virtual std::unique_ptr watch(const std::string & entity_id, const std::string & item_name, + SampleCallback callback) = 0; +}; + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/trigger_host.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/trigger_host.hpp new file mode 100644 index 000000000..7d3134f88 --- /dev/null +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/trigger_host.hpp @@ -0,0 +1,88 @@ +// 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 +#include +#include + +#include "ros2_medkit_gateway/core/condition_evaluator.hpp" +#include "ros2_medkit_gateway/core/faults/trigger_fault_feed.hpp" +#include "ros2_medkit_gateway/core/managers/trigger_manager.hpp" +#include "ros2_medkit_gateway/core/models/thread_safe_entity_cache.hpp" +#include "ros2_medkit_gateway/core/resource_change_notifier.hpp" +#include "ros2_medkit_gateway/core/transports/data_item_source.hpp" +#include "ros2_medkit_gateway/core/trigger_store.hpp" + +namespace ros2_medkit_gateway { + +class LogManager; +class PluginManager; + +/// Settings of the trigger subsystem. The names and defaults are those of the `triggers.*` parameters. +struct TriggerHostConfig { + bool enabled{true}; ///< `triggers.enabled` + int64_t max_triggers{1000}; ///< `triggers.max_triggers`; outside [1, INT_MAX] the default is used + std::string on_restart_behavior{"reset"}; ///< `triggers.on_restart_behavior` + std::string storage_path; ///< `triggers.storage.path`; empty keeps triggers in memory + int64_t data_poll_interval_ms{1000}; ///< `triggers.data_poll_interval_ms`; clamped to [100, 60000] +}; + +/// What the host supplies around the trigger manager. +struct TriggerHostWiring { + /// Entity hierarchy, existence and data-item lookup. Required, and must outlive the host. + const ThreadSafeEntityCache * cache{nullptr}; + /// Subscribes data triggers to topics. Null for a host without topics. + std::shared_ptr topic_transport; + /// Non-null: a data trigger with no topic reads the item through this plugin manager, once per + /// `data_poll_interval_ms`. Null: no polling. Must outlive the host. + PluginManager * polled_plugins{nullptr}; + /// Non-null: fault confirmations from this source fire fault triggers. Must outlive the host. + FaultEventSource * fault_events{nullptr}; + /// Receives the trigger log settings, when set. Must outlive the host. + LogManager * log_manager{nullptr}; + /// Operator-facing notices (clamped settings, a trigger that gave up resolving its topic). + std::function warn; + /// How long a data trigger with an unknown topic keeps resolving it. + std::chrono::seconds unresolved_timeout{60}; +}; + +/// The trigger manager and the objects it works with. Empty when the configuration is disabled. +struct TriggerHost { + // Declaration order is destruction order in reverse: the feed and the manager go before what they use. + std::unique_ptr store; + std::shared_ptr data_source; + std::unique_ptr manager; + std::unique_ptr fault_feed; + size_t restored{0}; ///< persistent triggers put back from the store + int max_triggers{0}; ///< the limit in use + int64_t data_poll_interval_ms{0}; ///< the polling interval in use +}; + +/// Registers OnChange, OnChangeTo, EnterRange and LeaveRange. +void register_builtin_conditions(ConditionRegistry & conditions); + +/// The entity a reporting source (a node FQN) belongs to, or an empty string. A leading slash is optional. +std::string resolve_trigger_entity(const ThreadSafeEntityCache & cache, const std::string & node_fqn); + +/// Builds the store and the manager, wires them to `cache`, restores persistent triggers and, when the wiring has a +/// fault event source, starts the fault feed. Throws `std::invalid_argument` when `wiring.cache` is null. +TriggerHost make_trigger_host(const TriggerHostConfig & config, ResourceChangeNotifier & notifier, + ConditionRegistry & conditions, const TriggerHostWiring & wiring); + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp index 1d4e8810f..0c0db1b71 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp @@ -55,6 +55,7 @@ #include "ros2_medkit_gateway/core/resource_change_notifier.hpp" #include "ros2_medkit_gateway/core/resource_sampler.hpp" #include "ros2_medkit_gateway/core/subscription_transport.hpp" +#include "ros2_medkit_gateway/core/trigger_host.hpp" #include "ros2_medkit_gateway/core/trigger_store.hpp" #include "ros2_medkit_gateway/discovery/discovery_manager.hpp" #include "ros2_medkit_gateway/entity_freeze_frame_capture.hpp" @@ -497,8 +498,7 @@ class GatewayNode : public rclcpp::Node, public GatewayServices { // Trigger infrastructure (destroyed after rest_server_) std::unique_ptr resource_change_notifier_; std::unique_ptr condition_registry_; - std::unique_ptr trigger_store_; - std::unique_ptr trigger_mgr_; + TriggerHost trigger_host_; std::unique_ptr trigger_fault_subscriber_; // Zero-config freeze-frames for plugin-backed entities (nullptr when disabled) std::unique_ptr entity_freeze_frame_capture_; diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/trigger_fault_subscriber.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/trigger_fault_subscriber.hpp index 1fc5ba4dc..21e4b3d13 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/trigger_fault_subscriber.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/trigger_fault_subscriber.hpp @@ -15,26 +15,22 @@ #pragma once #include -#include #include #include "rclcpp/rclcpp.hpp" +#include "ros2_medkit_gateway/core/faults/trigger_fault_feed.hpp" #include "ros2_medkit_gateway/core/resource_change_notifier.hpp" -#include "ros2_medkit_msgs/msg/fault_event.hpp" +#include "ros2_medkit_gateway/ros2/ros2_fault_event_source.hpp" namespace ros2_medkit_gateway { /** - * @brief Subscribes to the configured fault manager events topic and forwards fault events - * to ResourceChangeNotifier for trigger evaluation. + * @brief Feeds the fault events of the configured fault manager events topic to + * ResourceChangeNotifier for trigger evaluation. * - * Follows the same ROS 2 subscription pattern as SSEFaultHandler - subscribes - * to the fault event topic and converts each FaultEvent into a ResourceChange - * notification on the "faults" collection. - * - * Entity ID is derived from the first reporting source of the fault. The - * fault_code is used as the resource_path. The full fault JSON (via - * ros2::conversions::fault_to_json) is passed as the change value. + * The ROS adapter of TriggerFaultFeed: it owns a Ros2FaultEventSource on the + * events topic and a feed on top of it. The conversion and the entity mapping + * are the feed's. */ class TriggerFaultSubscriber { public: @@ -45,31 +41,23 @@ class TriggerFaultSubscriber { */ TriggerFaultSubscriber(rclcpp::Node * node, ResourceChangeNotifier & notifier); - ~TriggerFaultSubscriber(); - /// Callback to resolve ROS node FQN to manifest entity ID. /// Returns manifest entity ID, or empty string if not resolvable. - using NodeToEntityFn = std::function; + using NodeToEntityFn = TriggerFaultFeed::NodeToEntityFn; /// Set the node-to-entity resolver for manifest entity ID lookup. + /// Write-once: call it before the ROS executor starts spinning. void set_node_to_entity_resolver(NodeToEntityFn resolver); - // Non-copyable, non-movable + ~TriggerFaultSubscriber() = default; TriggerFaultSubscriber(const TriggerFaultSubscriber &) = delete; TriggerFaultSubscriber & operator=(const TriggerFaultSubscriber &) = delete; TriggerFaultSubscriber(TriggerFaultSubscriber &&) = delete; TriggerFaultSubscriber & operator=(TriggerFaultSubscriber &&) = delete; private: - /// Callback for fault events from ROS 2 topic - void on_fault_event(const ros2_medkit_msgs::msg::FaultEvent::ConstSharedPtr & msg); - - rclcpp::Subscription::SharedPtr subscription_; - ResourceChangeNotifier & notifier_; - rclcpp::Logger logger_; - /// Write-once: must be set before ROS executor starts spinning. - /// After that, only read from subscription callbacks (no mutex needed). - NodeToEntityFn node_to_entity_resolver_; + Ros2FaultEventSource source_; + TriggerFaultFeed feed_; ///< after source_: destroyed first, so it unsubscribes while the source lives }; } // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/core/data/polling_data_item_source.cpp b/src/ros2_medkit_gateway/src/core/data/polling_data_item_source.cpp new file mode 100644 index 000000000..b52d5d90c --- /dev/null +++ b/src/ros2_medkit_gateway/src/core/data/polling_data_item_source.cpp @@ -0,0 +1,191 @@ +// 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/polling_data_item_source.hpp" + +#include +#include +#include +#include +#include +#include + +#include "ros2_medkit_gateway/core/data/data_item_reader.hpp" + +namespace ros2_medkit_gateway { + +namespace { +using Clock = std::chrono::steady_clock; + +struct Watch { + std::string entity_id; + std::string item_name; + std::mutex mutex; ///< held while the item is read and the callback runs + bool alive{true}; + DataItemSource::SampleCallback callback; + Clock::time_point next_due; ///< guarded by Shared::mutex +}; +} // namespace + +struct PollingDataItemSource::Shared { + Shared(ReadFn read_fn, std::chrono::milliseconds poll_interval) + : read(std::move(read_fn)), interval(poll_interval), thread([this] { + run(); + }) { + } + + Shared(const Shared &) = delete; + Shared & operator=(const Shared &) = delete; + Shared(Shared &&) = delete; + Shared & operator=(Shared &&) = delete; + + ~Shared() { + { + std::lock_guard lock(mutex); + stop = true; + } + wake.notify_all(); + thread.join(); + } + + void run() { + std::unique_lock lock(mutex); + while (!stop) { + const auto now = Clock::now(); + std::vector> due; + auto next = Clock::time_point::max(); + for (const auto & watch : watches) { + if (watch->next_due <= now) { + due.push_back(watch); + } else { + next = std::min(next, watch->next_due); + } + } + if (due.empty()) { + if (next == Clock::time_point::max()) { + wake.wait(lock); + } else { + wake.wait_until(lock, next); + } + continue; + } + lock.unlock(); + for (const auto & watch : due) { + poll(*watch); + } + lock.lock(); + const auto after = Clock::now(); + for (const auto & watch : due) { + watch->next_due = after + interval; + } + } + } + + void poll(Watch & watch) const { + std::lock_guard lock(watch.mutex); + if (!watch.alive) { + return; + } + try { + auto value = read(watch.entity_id, watch.item_name); + if (value) { + watch.callback(*value); + } + } catch (...) { + // A throwing reader delivers nothing this round. + } + } + + ReadFn read; + std::chrono::milliseconds interval; + std::mutex mutex; ///< guards watches, stop and every Watch::next_due + std::condition_variable wake; + std::vector> watches; + bool stop{false}; + std::thread thread; ///< last member: starts after the others are built +}; + +class PollingDataItemSource::Handle : public TopicSubscriptionHandle { + public: + Handle(std::weak_ptr shared, std::shared_ptr watch) + : shared_(std::move(shared)), watch_(std::move(watch)) { + } + ~Handle() override { + { + std::lock_guard lock(watch_->mutex); + watch_->alive = false; + watch_->callback = nullptr; + } + if (auto shared = shared_.lock()) { + std::lock_guard lock(shared->mutex); + shared->watches.erase(std::remove(shared->watches.begin(), shared->watches.end(), watch_), shared->watches.end()); + } + } + Handle(const Handle &) = delete; + Handle & operator=(const Handle &) = delete; + Handle(Handle &&) = delete; + Handle & operator=(Handle &&) = delete; + + private: + std::weak_ptr shared_; + std::shared_ptr watch_; +}; + +PollingDataItemSource::PollingDataItemSource(ReadFn read, std::chrono::milliseconds interval) + : shared_(std::make_shared(std::move(read), interval)) { +} + +PollingDataItemSource::~PollingDataItemSource() = default; + +std::unique_ptr +PollingDataItemSource::watch(const std::string & entity_id, const std::string & item_name, SampleCallback callback) { + if (!callback) { + return nullptr; + } + auto watch = std::make_shared(); + watch->entity_id = entity_id; + watch->item_name = item_name; + watch->callback = std::move(callback); + watch->next_due = Clock::now(); + { + std::lock_guard lock(shared_->mutex); + shared_->watches.push_back(watch); + } + shared_->wake.notify_all(); + return std::make_unique(shared_, std::move(watch)); +} + +std::unique_ptr make_plugin_data_item_source(PluginManager * plugins, + std::chrono::milliseconds interval) { + return std::make_unique( + [plugins](const std::string & entity_id, + const std::string & item_name) -> tl::expected { + auto item = read_plugin_data_item(plugins, entity_id, item_name); + if (!item) { + return tl::make_unexpected(item.error().message); + } + // A topic trigger sees the message, which is the `data` field of the item body. The same trigger path must + // work here. A body without `data` (a plugin's own shape) is delivered whole. + if (item->content.is_object()) { + const auto data = item->content.find("data"); + if (data != item->content.end()) { + return std::move(*data); + } + } + return std::move(item->content); + }, + interval); +} + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/core/faults/trigger_fault_feed.cpp b/src/ros2_medkit_gateway/src/core/faults/trigger_fault_feed.cpp new file mode 100644 index 000000000..4bd5c4104 --- /dev/null +++ b/src/ros2_medkit_gateway/src/core/faults/trigger_fault_feed.cpp @@ -0,0 +1,74 @@ +// 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/faults/trigger_fault_feed.hpp" + +#include + +#include "ros2_medkit_gateway/core/faults/fault_json.hpp" + +namespace ros2_medkit_gateway { + +namespace fd = ros2_medkit::fault_detection; + +TriggerFaultFeed::TriggerFaultFeed(FaultEventSource & source, ResourceChangeNotifier & notifier, + NodeToEntityFn resolver) + : source_(source), notifier_(notifier), resolver_(std::move(resolver)) { + source_.subscribe([this](const fd::FaultEventRecord & event) { + on_event(event); + }); +} + +TriggerFaultFeed::~TriggerFaultFeed() { + source_.unsubscribe(); +} + +void TriggerFaultFeed::set_node_to_entity_resolver(NodeToEntityFn resolver) { + resolver_ = std::move(resolver); +} + +void TriggerFaultFeed::on_event(const fd::FaultEventRecord & event) { + nlohmann::json fault_json = fault_record_to_json(event.fault); + fault_json["event_type"] = event.event_type; + + // The resolver maps a source to a manifest entity id. Without a mapping the last path segment is the entity id. + std::string entity_id; + if (!event.fault.reporting_sources.empty()) { + const auto & source = event.fault.reporting_sources[0]; + if (resolver_) { + entity_id = resolver_(source); + } + if (entity_id.empty()) { + const auto last_slash = source.rfind('/'); + entity_id = (last_slash != std::string::npos) ? source.substr(last_slash + 1) : source; + } + } + // No entity: a catch-all match would be unintended. + if (entity_id.empty()) { + return; + } + + // Both fault_cleared and a heal remove the fault from the active set. The payload "status" tells them apart. + ChangeType change_type = ChangeType::UPDATED; + if (event.event_type == fd::fault_event_type::kConfirmed) { + change_type = ChangeType::CREATED; + } else if (event.event_type == fd::fault_event_type::kCleared) { + change_type = ChangeType::DELETED; + } + + // The leading slash matches the resource_path that parse_resource_uri() gives a trigger. + notifier_.notify("faults", entity_id, "/" + event.fault.fault_code, fault_json, change_type); +} + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/core/managers/trigger_manager.cpp b/src/ros2_medkit_gateway/src/core/managers/trigger_manager.cpp index d43a7b26b..7b4e72e07 100644 --- a/src/ros2_medkit_gateway/src/core/managers/trigger_manager.cpp +++ b/src/ros2_medkit_gateway/src/core/managers/trigger_manager.cpp @@ -24,18 +24,27 @@ namespace ros2_medkit_gateway { +namespace { +/// The item name of a data resource path: "/speed" is "speed". +std::string data_item_name(const std::string & resource_path) { + return (!resource_path.empty() && resource_path.front() == '/') ? resource_path.substr(1) : resource_path; +} +} // namespace + // --------------------------------------------------------------------------- // Lifecycle // --------------------------------------------------------------------------- TriggerManager::TriggerManager(ResourceChangeNotifier & notifier, ConditionRegistry & conditions, TriggerStore & store, const TriggerConfig & config, - std::shared_ptr topic_transport) + std::shared_ptr topic_transport, + std::shared_ptr data_source) : notifier_(notifier) , conditions_(conditions) , store_(store) , config_(config) - , topic_transport_(std::move(topic_transport)) { + , topic_transport_(std::move(topic_transport)) + , data_source_(std::move(data_source)) { // Subscribe to all resource changes (empty filter = catch-all) NotifierFilter filter; notifier_sub_id_ = notifier_.subscribe(filter, [this](const ResourceChange & change) { @@ -417,39 +426,43 @@ tl::expected TriggerManager::create(const Trigg triggers_[state->info.id] = std::move(state); } - // Subscribe to topic for data triggers. - // When resolved_topic_name is empty (topic not yet discoverable at creation time), - // queue for deferred resolution - retry_unresolved_triggers() will pick it up. - if (topic_transport_ && req.collection == "data") { - if (!req.resolved_topic_name.empty()) { - const std::string entity_id = req.entity_id; - const std::string resource_path = req.resource_path; - auto handle = topic_transport_->subscribe( - req.resolved_topic_name, /*msg_type=*/"", [this, entity_id, resource_path](const nlohmann::json & sample) { - notifier_.notify("data", entity_id, resource_path, sample, ChangeType::UPDATED); - }); - if (handle) { + // Subscribe data triggers. A known topic goes to the transport. Without one, the item source reads the item, or, with + // no item source, deferred resolution waits for the topic (retry_unresolved_triggers() picks it up). + if (req.collection == "data") { + const std::string entity_id = req.entity_id; + const std::string resource_path = req.resource_path; + const auto notify_sample = [this, entity_id, resource_path](const nlohmann::json & sample) { + notifier_.notify("data", entity_id, resource_path, sample, ChangeType::UPDATED); + }; + // A failed subscribe rolls the trigger back, so the caller does not get a 201 for a trigger that will never + // deliver. A persisted record goes too, so a restart does not bring a doomed trigger back. + const auto roll_back = [&](const std::string & what) -> tl::expected { + { std::lock_guard lock(triggers_mutex_); - topic_handles_[info_copy.id] = std::move(handle); - } else { - // Subscribe failed at trigger creation. Roll back the trigger we - // just added so the caller does not get a 201 for a trigger that - // will never deliver. The underlying TriggerTopicSubscriber has - // already logged the rclcpp failure. Persisted records are also - // rolled back so a restart does not resurrect a doomed trigger. - { - std::lock_guard lock(triggers_mutex_); - remove_from_dispatch_index(info_copy.id, info_copy.collection, info_copy.entity_id); - triggers_.erase(info_copy.id); - } - if (req.persistent) { - (void)store_.remove(info_copy.id); - } - return tl::make_unexpected(TriggerCreateError{ - TriggerError::SubscribeFailed, "Failed to subscribe to topic '" + req.resolved_topic_name + - "' for data trigger; see gateway log for the rclcpp root cause"}); + remove_from_dispatch_index(info_copy.id, info_copy.collection, info_copy.entity_id); + triggers_.erase(info_copy.id); + } + if (req.persistent) { + (void)store_.remove(info_copy.id); + } + return tl::make_unexpected(TriggerCreateError{TriggerError::SubscribeFailed, what}); + }; + if (topic_transport_ && !req.resolved_topic_name.empty()) { + auto handle = topic_transport_->subscribe(req.resolved_topic_name, /*msg_type=*/"", notify_sample); + if (!handle) { + return roll_back("Failed to subscribe to topic '" + req.resolved_topic_name + + "' for data trigger; see gateway log for the rclcpp root cause"); } - } else if (!req.resource_path.empty()) { + std::lock_guard lock(triggers_mutex_); + topic_handles_[info_copy.id] = std::move(handle); + } else if (data_source_ && !req.resource_path.empty()) { + auto handle = data_source_->watch(req.entity_id, data_item_name(req.resource_path), notify_sample); + if (!handle) { + return roll_back("Failed to watch data item '" + req.resource_path + "' of entity '" + req.entity_id + "'"); + } + std::lock_guard lock(triggers_mutex_); + topic_handles_[info_copy.id] = std::move(handle); + } else if (topic_transport_ && !req.resource_path.empty()) { std::lock_guard lock(triggers_mutex_); // The entity was validated to exist before this request was accepted, so // it counts as seen and the give-up budget runs from now. @@ -801,19 +814,39 @@ size_t TriggerManager::load_persistent_triggers() { add_to_dispatch_index(state->info.id, state->info.collection, state->info.entity_id); - // Restored data triggers go through deferred resolution rather than - // straight onto the persisted topic name. The entity belongs to a process - // this gateway has not met yet, so the subscription attempt has to stay - // open for as long as the record does, and deferred resolution is the - // only path whose budget is tied to entity_seen. Subscribing here could - // not honour that in any case: restore runs inside the constructor, the - // transport's executor is wired after it returns, so the attempt would be - // parked on a clock that starts now and runs out long before a late - // entity arrives. - if (topic_transport_ && state->info.collection == "data") { - const std::string trigger_id = state->info.id; - const std::string entity_id = state->info.entity_id; - const std::string resource_path = state->info.resource_path; + const std::string trigger_id = state->info.id; + const std::string entity_id = state->info.entity_id; + const std::string resource_path = state->info.resource_path; + // With an item source, a stored trigger that had no topic is read from its item. A reader that cannot answer yet + // delivers nothing and is asked again, so it needs no deferred resolution. + if (data_source_ && state->info.collection == "data" && !resource_path.empty() && + (!topic_transport_ || state->info.resolved_topic_name.empty())) { + auto handle = data_source_->watch( + entity_id, data_item_name(resource_path), [this, entity_id, resource_path](const nlohmann::json & sample) { + notifier_.notify("data", entity_id, resource_path, sample, ChangeType::UPDATED); + }); + if (handle) { + topic_handles_[trigger_id] = std::move(handle); + } else { + std::string notice = "persistent trigger restore: trigger '"; + notice += trigger_id; + notice += "' could not watch data item '"; + notice += resource_path; + notice += "' of entity '"; + notice += entity_id; + notice += "'. It stays active but cannot fire; delete it and recreate it."; + warnings.push_back(std::move(notice)); + } + } else if (topic_transport_ && state->info.collection == "data") { + // Restored data triggers go through deferred resolution rather than + // straight onto the persisted topic name. The entity belongs to a process + // this gateway has not met yet, so the subscription attempt has to stay + // open for as long as the record does, and deferred resolution is the + // only path whose budget is tied to entity_seen. Subscribing here could + // not honour that in any case: restore runs inside the constructor, the + // transport's executor is wired after it returns, so the attempt would be + // parked on a clock that starts now and runs out long before a late + // entity arrives. if (!resource_path.empty()) { unresolved_data_triggers_.push_back({trigger_id, entity_id, resource_path, std::chrono::steady_clock::now(), std::nullopt, diff --git a/src/ros2_medkit_gateway/src/core/trigger_host.cpp b/src/ros2_medkit_gateway/src/core/trigger_host.cpp new file mode 100644 index 000000000..26bf33ba0 --- /dev/null +++ b/src/ros2_medkit_gateway/src/core/trigger_host.cpp @@ -0,0 +1,149 @@ +// 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/trigger_host.hpp" + +#include +#include +#include +#include +#include +#include + +#include "ros2_medkit_gateway/core/data/polling_data_item_source.hpp" +#include "ros2_medkit_gateway/core/sqlite_trigger_store.hpp" + +namespace ros2_medkit_gateway { + +namespace { +constexpr int kDefaultMaxTriggers = 1000; +constexpr int64_t kMinPollIntervalMs = 100; +constexpr int64_t kMaxPollIntervalMs = 60000; + +std::vector apps_under(const ThreadSafeEntityCache & cache, const std::string & entity_id, + const std::string & entity_type) { + if (entity_type == "components") { + return cache.get_apps_for_component(entity_id); + } + if (entity_type == "areas") { + std::vector all_apps; + for (const auto & component_id : cache.get_components_for_area(entity_id)) { + auto apps = cache.get_apps_for_component(component_id); + all_apps.insert(all_apps.end(), apps.begin(), apps.end()); + } + return all_apps; + } + if (entity_type == "functions") { + return cache.get_apps_for_function(entity_id); + } + return {}; +} + +std::string topic_of(const ThreadSafeEntityCache & cache, const std::string & entity_id, + const std::string & resource_path) { + for (const auto & topic : cache.get_entity_data(entity_id).topics) { + if (topic.name == resource_path || + (topic.name.size() > resource_path.size() && + topic.name.compare(topic.name.size() - resource_path.size(), resource_path.size(), resource_path) == 0)) { + return topic.name; + } + } + return ""; +} +} // namespace + +void register_builtin_conditions(ConditionRegistry & conditions) { + conditions.register_condition("OnChange", std::make_shared()); + conditions.register_condition("OnChangeTo", std::make_shared()); + conditions.register_condition("EnterRange", std::make_shared()); + conditions.register_condition("LeaveRange", std::make_shared()); +} + +std::string resolve_trigger_entity(const ThreadSafeEntityCache & cache, const std::string & node_fqn) { + auto entity_id = cache.resolve_node_to_app(node_fqn); + if (entity_id.empty() && !node_fqn.empty() && node_fqn[0] == '/') { + entity_id = cache.resolve_node_to_app(node_fqn.substr(1)); + } + return entity_id; +} + +TriggerHost make_trigger_host(const TriggerHostConfig & config, ResourceChangeNotifier & notifier, + ConditionRegistry & conditions, const TriggerHostWiring & wiring) { + TriggerHost host; + if (!config.enabled) { + return host; + } + if (wiring.cache == nullptr) { + throw std::invalid_argument("make_trigger_host needs an entity cache"); + } + const ThreadSafeEntityCache * cache = wiring.cache; + const auto warn = [notice = wiring.warn](const std::string & message) { + if (notice) { + notice(message); + } + }; + + host.store = std::make_unique(config.storage_path.empty() ? ":memory:" : config.storage_path); + + // The limit is compared with a size_t in TriggerManager, so a negative would become a huge bound and the + // documented 503 at the limit would never come. Checked as int64 before it is narrowed. + TriggerConfig trigger_config; + if (config.max_triggers < 1 || config.max_triggers > std::numeric_limits::max()) { + warn("triggers.max_triggers " + std::to_string(config.max_triggers) + " out of range. Using default " + + std::to_string(kDefaultMaxTriggers) + "."); + trigger_config.max_triggers = kDefaultMaxTriggers; + } else { + trigger_config.max_triggers = static_cast(config.max_triggers); + } + trigger_config.on_restart_behavior = config.on_restart_behavior; + host.max_triggers = trigger_config.max_triggers; + + host.data_poll_interval_ms = std::clamp(config.data_poll_interval_ms, kMinPollIntervalMs, kMaxPollIntervalMs); + if (host.data_poll_interval_ms != config.data_poll_interval_ms) { + warn("triggers.data_poll_interval_ms " + std::to_string(config.data_poll_interval_ms) + " clamped to " + + std::to_string(host.data_poll_interval_ms)); + } + if (wiring.polled_plugins != nullptr) { + host.data_source = + make_plugin_data_item_source(wiring.polled_plugins, std::chrono::milliseconds(host.data_poll_interval_ms)); + } + + host.manager = std::make_unique(notifier, conditions, *host.store, trigger_config, + wiring.topic_transport, host.data_source); + host.manager->set_entity_children_fn([cache](const std::string & entity_id, const std::string & entity_type) { + return apps_under(*cache, entity_id, entity_type); + }); + host.manager->set_entity_exists_fn([cache](const std::string & entity_id, const std::string & /*entity_type*/) { + return cache->find_entity(entity_id).has_value(); + }); + host.manager->set_resolve_topic_fn([cache](const std::string & entity_id, const std::string & resource_path) { + return topic_of(*cache, entity_id, resource_path); + }); + host.manager->set_warn_log_fn(warn); + host.manager->set_unresolved_timeout(wiring.unresolved_timeout); + if (wiring.log_manager != nullptr) { + host.manager->set_log_manager(wiring.log_manager); + } + host.restored = host.manager->load_persistent_triggers(); + + if (wiring.fault_events != nullptr) { + host.fault_feed = + std::make_unique(*wiring.fault_events, notifier, [cache](const std::string & node_fqn) { + return resolve_trigger_entity(*cache, node_fqn); + }); + } + return host; +} + +} // 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 97fd2c044..fba5a4e7b 100644 --- a/src/ros2_medkit_gateway/src/gateway_node.cpp +++ b/src/ros2_medkit_gateway/src/gateway_node.cpp @@ -42,7 +42,6 @@ #include "ros2_medkit_gateway/ros2/status/ros2_lifecycle_state_reader.hpp" #include "ros2_medkit_gateway/core/http/handlers/sse_transport_provider.hpp" -#include "ros2_medkit_gateway/core/sqlite_trigger_store.hpp" using namespace std::chrono_literals; @@ -956,34 +955,14 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki }); condition_registry_ = std::make_unique(); - condition_registry_->register_condition("OnChange", std::make_shared()); - condition_registry_->register_condition("OnChangeTo", std::make_shared()); - condition_registry_->register_condition("EnterRange", std::make_shared()); - condition_registry_->register_condition("LeaveRange", std::make_shared()); + register_builtin_conditions(*condition_registry_); auto triggers_enabled = get_parameter("triggers.enabled").as_bool(); if (triggers_enabled) { - auto storage_path = get_parameter("triggers.storage.path").as_string(); - if (storage_path.empty()) { - trigger_store_ = std::make_unique(":memory:"); - } else { - trigger_store_ = std::make_unique(storage_path); - } - - TriggerConfig trigger_config; - // The cap is stored as int here and compared against a size_t in - // TriggerManager, so a negative would become a huge unsigned bound and the - // documented "HTTP 503 when this limit is reached" would never fire - the - // trigger list would grow without one. Checked as int64 before narrowing. - const int64_t max_triggers_raw = get_parameter("triggers.max_triggers").as_int(); - if (max_triggers_raw < 1 || max_triggers_raw > std::numeric_limits::max()) { - RCLCPP_WARN(get_logger(), "triggers.max_triggers %" PRId64 " out of range. Using default 1000.", - max_triggers_raw); - trigger_config.max_triggers = 1000; - } else { - trigger_config.max_triggers = static_cast(max_triggers_raw); - } + TriggerHostConfig trigger_config; + trigger_config.max_triggers = get_parameter("triggers.max_triggers").as_int(); trigger_config.on_restart_behavior = get_parameter("triggers.on_restart_behavior").as_string(); + trigger_config.storage_path = get_parameter("triggers.storage.path").as_string(); // Subscribe to data topics for data triggers. The adapter is created // before the trigger manager because the manager takes the transport @@ -991,46 +970,27 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki trigger_topic_subscriber_ = std::make_unique(this); trigger_topic_transport_ = std::make_shared(trigger_topic_subscriber_.get()); - trigger_mgr_ = std::make_unique(*resource_change_notifier_, *condition_registry_, *trigger_store_, - trigger_config, trigger_topic_transport_); - - // Set entity hierarchy resolver using thread-safe cache - trigger_mgr_->set_entity_children_fn( - [this](const std::string & entity_id, const std::string & entity_type) -> std::vector { - const auto & cache = get_thread_safe_cache(); - if (entity_type == "components") { - return cache.get_apps_for_component(entity_id); - } else if (entity_type == "areas") { - auto comps = cache.get_components_for_area(entity_id); - std::vector all_apps; - for (const auto & comp_id : comps) { - auto apps = cache.get_apps_for_component(comp_id); - all_apps.insert(all_apps.end(), apps.begin(), apps.end()); - } - return all_apps; - } else if (entity_type == "functions") { - return cache.get_apps_for_function(entity_id); - } - return {}; - }); - - // Set entity existence checker for orphan sweep - trigger_mgr_->set_entity_exists_fn( - [this](const std::string & entity_id, const std::string & /*entity_type*/) -> bool { - const auto & cache = get_thread_safe_cache(); - return cache.find_entity(entity_id).has_value(); - }); - - trigger_mgr_->set_warn_log_fn([this](const std::string & message) { + // The deferred-resolution budget must outlive at least one full discovery + // refresh: plugin-declared topics only reach the entity cache on a refresh + // pass, and refresh_interval_ms is configurable up to 60 s - equal to the + // old fixed budget, which made "resolves on a later tick" false at the + // extreme. Two refresh cycles keeps the reasoning true for every allowed + // configuration. + TriggerHostWiring trigger_wiring; + trigger_wiring.cache = &get_thread_safe_cache(); + trigger_wiring.topic_transport = trigger_topic_transport_; + trigger_wiring.log_manager = log_mgr_.get(); + trigger_wiring.warn = [this](const std::string & message) { RCLCPP_WARN(get_logger(), "%s", message.c_str()); - }); - // Load persistent triggers. This is the only pass: a trigger the store - // holds but this call does not put back stays absent until the next - // restart, so the count is worth having in the log of every start that - // asked for a restore. - const size_t restored_triggers = trigger_mgr_->load_persistent_triggers(); + }; + trigger_wiring.unresolved_timeout = + std::max(std::chrono::seconds(60), std::chrono::seconds(2 * refresh_interval_ms_ / 1000)); + trigger_host_ = make_trigger_host(trigger_config, *resource_change_notifier_, *condition_registry_, trigger_wiring); + // Persistent triggers are restored once, inside make_trigger_host. A trigger the store holds but that pass does not + // put back stays absent until the next restart, so the count is worth having in the log of every start that asked + // for a restore. if (trigger_config.on_restart_behavior == "restore") { - RCLCPP_INFO(get_logger(), "Restored %zu persistent trigger(s) from the trigger store", restored_triggers); + RCLCPP_INFO(get_logger(), "Restored %zu persistent trigger(s) from the trigger store", trigger_host_.restored); } // Wire notifier to managers so they emit events for trigger evaluation @@ -1041,39 +1001,14 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki // Wire LogManager <-> TriggerManager for log_settings integration if (log_mgr_) { - trigger_mgr_->set_log_manager(log_mgr_.get()); log_mgr_->set_notifier(resource_change_notifier_.get()); } // Subscribe to fault events and forward to notifier trigger_fault_subscriber_ = std::make_unique(this, *resource_change_notifier_); - // Wire deferred topic resolution: when a trigger's resource_path couldn't - // be resolved to a ROS 2 topic at creation time, retry periodically using - // the entity cache to find matching topics. - trigger_mgr_->set_resolve_topic_fn([this](const std::string & entity_id, - const std::string & resource_path) -> std::string { - const auto & cache = get_thread_safe_cache(); - auto aggregated = cache.get_entity_data(entity_id); - for (const auto & topic : aggregated.topics) { - if (topic.name == resource_path || - (topic.name.size() > resource_path.size() && - topic.name.compare(topic.name.size() - resource_path.size(), resource_path.size(), resource_path) == 0)) { - return topic.name; - } - } - return ""; - }); - // The deferred-resolution budget must outlive at least one full discovery - // refresh: plugin-declared topics only reach the entity cache on a refresh - // pass, and refresh_interval_ms is configurable up to 60 s - equal to the - // old fixed budget, which made "resolves on a later tick" false at the - // extreme. Two refresh cycles keeps the reasoning true for every allowed - // configuration. - trigger_mgr_->set_unresolved_timeout( - std::max(std::chrono::seconds(60), std::chrono::seconds(2 * refresh_interval_ms_ / 1000))); trigger_topic_subscriber_->set_retry_callback([this]() { - trigger_mgr_->retry_unresolved_triggers(); + trigger_host_.manager->retry_unresolved_triggers(); }); // Wire node-to-entity resolver for trigger notifications (#305). @@ -1081,18 +1016,7 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki // mapping (populated from linking result). Captures 'this' - the lambda is // called from ROS subscription callbacks on the same node's executor. auto node_resolver = [this](const std::string & ros_fqn) -> std::string { - auto result = thread_safe_cache_.resolve_node_to_app(ros_fqn); - if (!result.empty()) { - return result; - } - // Try with normalized FQN (strip leading /) - if (!ros_fqn.empty() && ros_fqn[0] == '/') { - result = thread_safe_cache_.resolve_node_to_app(ros_fqn.substr(1)); - if (!result.empty()) { - return result; - } - } - return ""; + return resolve_trigger_entity(thread_safe_cache_, ros_fqn); }; if (trigger_fault_subscriber_) { @@ -1102,8 +1026,8 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki log_mgr_->set_node_to_entity_resolver(node_resolver); } - RCLCPP_INFO(get_logger(), "Trigger subsystem: enabled (max=%d, storage=%s)", trigger_config.max_triggers, - storage_path.empty() ? ":memory:" : storage_path.c_str()); + RCLCPP_INFO(get_logger(), "Trigger subsystem: enabled (max=%d, storage=%s)", trigger_host_.max_triggers, + trigger_config.storage_path.empty() ? ":memory:" : trigger_config.storage_path.c_str()); } else { RCLCPP_INFO(get_logger(), "Trigger subsystem: disabled"); } @@ -1271,8 +1195,8 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki refresh_cache(); // Sweep triggers whose entities disappeared from discovery - if (trigger_mgr_) { - trigger_mgr_->sweep_orphaned_triggers(); + if (trigger_host_.manager) { + trigger_host_.manager->sweep_orphaned_triggers(); } }); @@ -1306,8 +1230,8 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki cors_config_, auth_config_, rate_limit_config_, rest_config, tls_config_); // Wire trigger handlers into REST server - if (trigger_mgr_) { - rest_server_->set_trigger_handlers(*trigger_mgr_); + if (trigger_host_.manager) { + rest_server_->set_trigger_handlers(*trigger_host_.manager); } // Wire aggregation manager into REST server handler context @@ -1436,8 +1360,8 @@ GatewayNode::~GatewayNode() { trigger_topic_subscriber_->shutdown(); } // 4. Shutdown trigger subsystem (wakes SSE trigger streams) - if (trigger_mgr_) { - trigger_mgr_->shutdown(); + if (trigger_host_.manager) { + trigger_host_.manager->shutdown(); } // 4b. Drain in-flight update tasks BEFORE the notifier is shut down and // freed. Update tasks run on their own std::async threads (not the @@ -1838,7 +1762,7 @@ ResourceChangeNotifier * GatewayNode::get_resource_change_notifier() const { } TriggerManager * GatewayNode::get_trigger_manager() const { - return trigger_mgr_.get(); + return trigger_host_.manager.get(); } ConditionRegistry * GatewayNode::get_condition_registry() const { diff --git a/src/ros2_medkit_gateway/src/trigger_fault_subscriber.cpp b/src/ros2_medkit_gateway/src/trigger_fault_subscriber.cpp index f839eaecb..dca012715 100644 --- a/src/ros2_medkit_gateway/src/trigger_fault_subscriber.cpp +++ b/src/ros2_medkit_gateway/src/trigger_fault_subscriber.cpp @@ -14,81 +14,20 @@ #include "ros2_medkit_gateway/trigger_fault_subscriber.hpp" +#include + #include "ros2_medkit_gateway/fault_manager_paths.hpp" -#include "ros2_medkit_gateway/ros2/conversions/fault_msg_conversions.hpp" namespace ros2_medkit_gateway { TriggerFaultSubscriber::TriggerFaultSubscriber(rclcpp::Node * node, ResourceChangeNotifier & notifier) - : notifier_(notifier), logger_(node->get_logger()) { - const auto fault_events_topic = build_fault_manager_events_topic(node); - subscription_ = node->create_subscription( - fault_events_topic, rclcpp::QoS(100).reliable(), - [this](const ros2_medkit_msgs::msg::FaultEvent::ConstSharedPtr & msg) { - on_fault_event(msg); - }); - - RCLCPP_INFO(logger_, "TriggerFaultSubscriber initialized, subscribed to %s", fault_events_topic.c_str()); -} - -TriggerFaultSubscriber::~TriggerFaultSubscriber() { - subscription_.reset(); + : source_(*node, build_fault_manager_events_topic(node)), feed_(source_, notifier) { + RCLCPP_INFO(node->get_logger(), "TriggerFaultSubscriber initialized, subscribed to %s", + build_fault_manager_events_topic(node).c_str()); } void TriggerFaultSubscriber::set_node_to_entity_resolver(NodeToEntityFn resolver) { - node_to_entity_resolver_ = std::move(resolver); -} - -void TriggerFaultSubscriber::on_fault_event(const ros2_medkit_msgs::msg::FaultEvent::ConstSharedPtr & msg) { - // Convert fault to JSON using the same helper as SSEFaultHandler - nlohmann::json fault_json = ros2::conversions::fault_to_json(msg->fault); - fault_json["event_type"] = msg->event_type; - - // Derive entity_id from reporting_sources (first source, if available). - // In hybrid/manifest mode, the resolver maps ROS FQNs to manifest entity IDs. - // Falls back to last-segment extraction for runtime_only mode or when the - // resolver cannot map the FQN. - std::string entity_id; - if (!msg->fault.reporting_sources.empty()) { - auto raw_fqn = msg->fault.reporting_sources[0]; - - // Try resolving ROS FQN to manifest entity ID - if (node_to_entity_resolver_) { - entity_id = node_to_entity_resolver_(raw_fqn); - } - - // Fallback: extract last path segment (backward compat for runtime_only) - if (entity_id.empty()) { - auto last_slash = raw_fqn.rfind('/'); - entity_id = (last_slash != std::string::npos) ? raw_fqn.substr(last_slash + 1) : raw_fqn; - } - } - - // Map event_type to ChangeType. fault_cleared covers the manual clear AND - // the auto-heal; both remove the fault from the active set, so both are - // DELETED. Consumers needing the distinction read the payload's "status" - // (CLEARED vs HEALED) - change_type itself has no reader today. - ChangeType change_type = ChangeType::UPDATED; - if (msg->event_type == "fault_confirmed") { - change_type = ChangeType::CREATED; - } else if (msg->event_type == "fault_cleared") { - change_type = ChangeType::DELETED; - } - - // Skip dispatch if entity_id could not be determined (no reporting sources - // or FQN was just "/"). Prevents unintended catch-all matching. - if (entity_id.empty()) { - return; - } - - // Prefix fault_code with '/' to match the resource_path format produced by - // parse_resource_uri() (e.g. "/fault_001"), so fault triggers with a specific - // resource_path can match. - auto resource_path = "/" + msg->fault.fault_code; - notifier_.notify("faults", entity_id, resource_path, fault_json, change_type); - - RCLCPP_DEBUG(logger_, "Forwarded fault event to notifier: %s for %s (entity=%s)", msg->event_type.c_str(), - msg->fault.fault_code.c_str(), entity_id.c_str()); + feed_.set_node_to_entity_resolver(std::move(resolver)); } } // namespace ros2_medkit_gateway