Create DDS Reader on Zenoh Subscription (#7)

eclipse-zenoh · Oct 10, 2023 · b087607 · b087607
1 parent fb6f5bd
commit b087607
Show file tree

Hide file tree

Showing 13 changed files with 286 additions and 1,239 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -48,13 +48,13 @@ regex = "1.7.1"
 rustc_version = "0.4"
 serde = "1.0.154"
 serde_json = "1.0.94"
-zenoh = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "master", features = ["unstable"] }
-zenoh-collections = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "master" }
-zenoh-core = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "master" }
-zenoh-ext = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "master", features = ["unstable"] }
-zenoh-plugin-rest = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "master", default-features = false }
-zenoh-plugin-trait = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "master", default-features = false }
-zenoh-util = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "master", default-features = false }
+zenoh = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "matching_state", features = ["unstable"] }
+zenoh-collections = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "matching_state" }
+zenoh-core = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "matching_state" }
+zenoh-ext = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "matching_state", features = ["unstable"] }
+zenoh-plugin-rest = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "matching_state", default-features = false }
+zenoh-plugin-trait = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "matching_state", default-features = false }
+zenoh-util = { git = "https://github.com/eclipse-zenoh/zenoh", branch = "matching_state", default-features = false }
 
 [profile.release]
 debug = false

diff --git a/zenoh-plugin-ros2dds/src/lib.rs b/zenoh-plugin-ros2dds/src/lib.rs
@@ -267,7 +267,7 @@ pub async fn run(runtime: Runtime, config: Config) {
 
     let mut ros2_plugin = ROS2PluginRuntime {
         config: Arc::new(config),
-        zsession: &zsession,
+        zsession,
         participant,
         _member: member,
         plugin_id,
@@ -281,7 +281,7 @@ pub struct ROS2PluginRuntime<'a> {
     config: Arc<Config>,
     // Note: &'a Arc<Session> here to keep the ownership of Session outside this struct
     // and be able to store the publishers/subscribers it creates in this same struct.
-    zsession: &'a Arc<Session>,
+    zsession: Arc<Session>,
     participant: dds_entity_t,
     _member: LivelinessToken<'a>,
     plugin_id: OwnedKeyExpr,
@@ -323,7 +323,7 @@ impl<'a> ROS2PluginRuntime<'a> {
             .liveliness()
             .declare_subscriber(ke_liveliness_all)
             .querying()
-            .with(zenoh::handlers::DefaultHandler {})
+            .with(flume::unbounded())
             .res_async()
             .await
             .expect("Failed to create Liveliness Subscriber");
@@ -367,7 +367,7 @@ impl<'a> ROS2PluginRuntime<'a> {
         let mut routes_mgr = RoutesMgr::new(
             self.plugin_id.clone(),
             self.config.clone(),
-            self.zsession,
+            self.zsession.clone(),
             self.participant,
             discovery_mgr.discovered_entities.clone(),
             ros_discovery_mgr,

diff --git a/zenoh-plugin-ros2dds/src/ros_discovery.rs b/zenoh-plugin-ros2dds/src/ros_discovery.rs
@@ -189,7 +189,11 @@ impl RosDiscoveryInfoMgr {
                     _ = ros_disco_timer_rcv.recv_async() => {
                         let (ref msg, ref mut has_changed) = *zwrite!(participant_entities_state);
                         if *has_changed {
-                            log::debug!("Publish update on 'ros_discovery_info': {msg:?}");
+                            log::debug!("Publish update on 'ros_discovery_info' with {} writers and {} readers",
+                                msg.node_entities_info_seq.values().next().map_or(0, |n| n.writer_gid_seq.len()),
+                                msg.node_entities_info_seq.values().next().map_or(0, |n| n.reader_gid_seq.len())
+                            );
+                            log::trace!("Publish update on 'ros_discovery_info': {msg:?}");
                             Self::write(writer, msg).unwrap_or_else(|e|
                                 log::error!("Failed to publish update on 'ros_discovery_info' topic: {e}")
                             );