RouteServiceCli: create DDS Reader/Writer only if a remote Server is …

…announced (fix #62) (#63)
eclipse-zenoh · Jan 23, 2024 · ca44eb4 · ca44eb4
1 parent 65a4d94
commit ca44eb4
Show file tree

Hide file tree

Showing 8 changed files with 252 additions and 126 deletions.
diff --git a/zenoh-plugin-ros2dds/src/lib.rs b/zenoh-plugin-ros2dds/src/lib.rs
@@ -548,13 +548,24 @@ impl<'a> ROS2PluginRuntime<'a> {
         if let Some(allowance) = &self.config.allowance {
             use ROS2DiscoveryEvent::*;
             match evt {
-                DiscoveredMsgPub(_, iface) => allowance.is_publisher_allowed(&iface.name),
-                DiscoveredMsgSub(_, iface) => allowance.is_subscriber_allowed(&iface.name),
-                DiscoveredServiceSrv(_, iface) => allowance.is_service_srv_allowed(&iface.name),
-                DiscoveredServiceCli(_, iface) => allowance.is_service_cli_allowed(&iface.name),
-                DiscoveredActionSrv(_, iface) => allowance.is_action_srv_allowed(&iface.name),
-                DiscoveredActionCli(_, iface) => allowance.is_action_cli_allowed(&iface.name),
-                _ => true, // only Undiscovered events remain - always allow them (in case dynamic change of config is supported)
+                DiscoveredMsgPub(_, iface) | UndiscoveredMsgPub(_, iface) => {
+                    allowance.is_publisher_allowed(&iface.name)
+                }
+                DiscoveredMsgSub(_, iface) | UndiscoveredMsgSub(_, iface) => {
+                    allowance.is_subscriber_allowed(&iface.name)
+                }
+                DiscoveredServiceSrv(_, iface) | UndiscoveredServiceSrv(_, iface) => {
+                    allowance.is_service_srv_allowed(&iface.name)
+                }
+                DiscoveredServiceCli(_, iface) | UndiscoveredServiceCli(_, iface) => {
+                    allowance.is_service_cli_allowed(&iface.name)
+                }
+                DiscoveredActionSrv(_, iface) | UndiscoveredActionSrv(_, iface) => {
+                    allowance.is_action_srv_allowed(&iface.name)
+                }
+                DiscoveredActionCli(_, iface) | UndiscoveredActionCli(_, iface) => {
+                    allowance.is_action_cli_allowed(&iface.name)
+                }
             }
         } else {
             // no allow/deny configured => allow all

diff --git a/zenoh-plugin-ros2dds/src/route_action_cli.rs b/zenoh-plugin-ros2dds/src/route_action_cli.rs
@@ -83,7 +83,7 @@ impl RouteActionCli<'_> {
             format!("{ros2_name}/{}", *KE_SUFFIX_ACTION_SEND_GOAL),
             format!("{ros2_type}_SendGoal"),
             &zenoh_key_expr_prefix / *KE_SUFFIX_ACTION_SEND_GOAL,
-            &None,
+            None,
             send_goal_queries_timeout,
             context.clone(),
         )
@@ -97,7 +97,7 @@ impl RouteActionCli<'_> {
             format!("{ros2_name}/{}", *KE_SUFFIX_ACTION_CANCEL_GOAL),
             ROS2_ACTION_CANCEL_GOAL_SRV_TYPE.to_string(),
             &zenoh_key_expr_prefix / *KE_SUFFIX_ACTION_CANCEL_GOAL,
-            &None,
+            None,
             cancel_goal_queries_timeout,
             context.clone(),
         )
@@ -111,7 +111,7 @@ impl RouteActionCli<'_> {
             format!("{ros2_name}/{}", *KE_SUFFIX_ACTION_GET_RESULT),
             format!("{ros2_type}_GetResult"),
             &zenoh_key_expr_prefix / *KE_SUFFIX_ACTION_GET_RESULT,
-            &None,
+            None,
             get_result_queries_timeout,
             context.clone(),
         )
@@ -154,7 +154,8 @@ impl RouteActionCli<'_> {
         })
     }
 
-    async fn activate(&mut self) -> Result<(), String> {
+    // Announce the route over Zenoh via a LivelinessToken
+    async fn announce_route(&mut self) -> Result<(), String> {
         self.is_active = true;
 
         // create associated LivelinessToken
@@ -163,6 +164,7 @@ impl RouteActionCli<'_> {
             &self.zenoh_key_expr_prefix,
             &self.ros2_type,
         )?;
+        log::debug!("{self} announce via token {liveliness_ke}");
         let ros2_name = self.ros2_name.clone();
         self.liveliness_token = Some(self.context.zsession
             .liveliness()
@@ -178,8 +180,9 @@ impl RouteActionCli<'_> {
         Ok(())
     }
 
-    fn deactivate(&mut self) {
-        log::debug!("{self} deactivate");
+    // Retire the route over Zenoh removing the LivelinessToken
+    fn retire_route(&mut self) {
+        log::debug!("{self} retire");
         // Drop Zenoh Publisher and Liveliness token
         // The DDS Writer remains to be discovered by local ROS nodes
         self.is_active = false;
@@ -256,7 +259,7 @@ impl RouteActionCli<'_> {
         log::debug!("{self} now serving local nodes {:?}", self.local_nodes);
         // if 1st local node added, activate the route
         if self.local_nodes.len() == 1 {
-            if let Err(e) = self.activate().await {
+            if let Err(e) = self.announce_route().await {
                 log::error!("{self} activation failed: {e}");
             }
         }
@@ -274,7 +277,7 @@ impl RouteActionCli<'_> {
         log::debug!("{self} now serving local nodes {:?}", self.local_nodes);
         // if last local node removed, deactivate the route
         if self.local_nodes.is_empty() {
-            self.deactivate();
+            self.retire_route();
         }
     }
 

diff --git a/zenoh-plugin-ros2dds/src/route_action_srv.rs b/zenoh-plugin-ros2dds/src/route_action_srv.rs
@@ -140,7 +140,8 @@ impl RouteActionSrv<'_> {
         })
     }
 
-    async fn activate(&mut self) -> Result<(), String> {
+    // Announce the route over Zenoh via a LivelinessToken
+    async fn announce_route(&mut self) -> Result<(), String> {
         self.is_active = true;
 
         // create associated LivelinessToken
@@ -149,6 +150,7 @@ impl RouteActionSrv<'_> {
             &self.zenoh_key_expr_prefix,
             &self.ros2_type,
         )?;
+        log::debug!("{self} announce via token {liveliness_ke}");
         let ros2_name = self.ros2_name.clone();
         self.liveliness_token = Some(self.context.zsession
             .liveliness()
@@ -164,8 +166,9 @@ impl RouteActionSrv<'_> {
         Ok(())
     }
 
-    fn deactivate(&mut self) {
-        log::debug!("{self} deactivate");
+    // Retire the route over Zenoh removing the LivelinessToken
+    fn retire_route(&mut self) {
+        log::debug!("{self} retire");
         // Drop Zenoh Publisher and Liveliness token
         // The DDS Writer remains to be discovered by local ROS nodes
         self.is_active = false;
@@ -242,7 +245,7 @@ impl RouteActionSrv<'_> {
         log::debug!("{self} now serving local nodes {:?}", self.local_nodes);
         // if 1st local node added, activate the route
         if self.local_nodes.len() == 1 {
-            if let Err(e) = self.activate().await {
+            if let Err(e) = self.announce_route().await {
                 log::error!("{self} activation failed: {e}");
             }
         }
@@ -260,7 +263,7 @@ impl RouteActionSrv<'_> {
         log::debug!("{self} now serving local nodes {:?}", self.local_nodes);
         // if last local node removed, deactivate the route
         if self.local_nodes.is_empty() {
-            self.deactivate();
+            self.retire_route();
         }
     }
 

diff --git a/zenoh-plugin-ros2dds/src/route_publisher.rs b/zenoh-plugin-ros2dds/src/route_publisher.rs
@@ -248,8 +248,6 @@ impl RoutePublisher<'_> {
                 .map_err(|e| format!("Failed to lisetn of matchibng status changes: {e}",))?
         };
 
-        // Ok(route)
-
         Ok(RoutePublisher {
             ros2_name,
             ros2_type,
@@ -427,6 +425,7 @@ fn activate_dds_reader(
     context.ros_discovery_mgr.add_dds_reader(get_guid(&reader)?);
 
     if old != DDS_ENTITY_NULL {
+        log::warn!("{route_id}: on activation their was already a DDS Reader - overwrite it");
         if let Err(e) = delete_dds_entity(old) {
             log::warn!("{route_id}: failed to delete overwritten DDS Reader: {e}");
         }