fix(graph): make graph format explicit

Author: schmidma

crates/ros-z-protocol/src/format/ros2dds.rs

@@ -131,7 +131,11 @@ impl KeyExprFormatter for Ros2DdsFormatter {
 
         // Topic key expression (escaped)
         let topic_escaped = iter.next().ok_or(MissingTopicName)?;
-        let topic = Self::demangle_name(topic_escaped);
+        let topic = match Self::demangle_name(topic_escaped) {
+            topic if topic.is_empty() => "/".to_string(),
+            topic if topic.starts_with('/') => topic,
+            topic => format!("/{}", topic),
+        };
 
         // Type name (escaped)
         let type_escaped = iter.next().ok_or(MissingTopicType)?;

crates/ros-z/src/context.rs

@@ -95,7 +95,7 @@ impl ZContextBuilder {
         self
     }
 
-    /// Set the key expression format for ROS 2 entity mapping.
+    /// Set the key expression format for ROS 2 entity mapping and graph discovery.
     ///
     /// # Example
     /// ```ignore
@@ -533,7 +533,7 @@ impl Builder for ZContextBuilder {
         }
 
         let domain_id = builder.domain_id;
-        let graph = Arc::new(Graph::new(&session, domain_id)?);
+        let graph = Arc::new(Graph::new(&session, domain_id, builder.keyexpr_format)?);
 
         Ok(ZContext {
             session: Arc::new(session),

crates/ros-z/src/dynamic/tests/pubsub_tests.rs

@@ -164,7 +164,9 @@ fn test_zpub_builder_with_dyn_schema() {
 
     // Create a mock builder to test with_dyn_schema
     let session = zenoh::Wait::wait(zenoh::open(zenoh::Config::default())).unwrap();
-    let graph = std::sync::Arc::new(crate::graph::Graph::new(&session, 0).unwrap());
+    let graph = std::sync::Arc::new(
+        crate::graph::Graph::new(&session, 0, ros_z_protocol::KeyExprFormat::default()).unwrap(),
+    );
     let builder: ZPubBuilder<DynamicMessage> = ZPubBuilder {
         entity: crate::entity::EndpointEntity {
             id: 0,
@@ -204,7 +206,9 @@ fn test_zpub_builder_with_serdes_preserves_schema() {
 
     // Create builder with schema
     let session = zenoh::Wait::wait(zenoh::open(zenoh::Config::default())).unwrap();
-    let graph = std::sync::Arc::new(crate::graph::Graph::new(&session, 0).unwrap());
+    let graph = std::sync::Arc::new(
+        crate::graph::Graph::new(&session, 0, ros_z_protocol::KeyExprFormat::default()).unwrap(),
+    );
     let builder: ZPubBuilder<DynamicMessage> = ZPubBuilder {
         entity: crate::entity::EndpointEntity {
             id: 0,

crates/ros-z/src/graph.rs

@@ -431,6 +431,33 @@ impl std::fmt::Debug for Graph {
 }
 
 impl Graph {
+    /// Create a new Graph using an explicit key expression format.
+    ///
+    /// The format determines both the liveliness subscription pattern and the
+    /// parser used to turn liveliness keys back into ROS entities.
+    pub fn new(
+        session: &Session,
+        domain_id: usize,
+        format: ros_z_protocol::KeyExprFormat,
+    ) -> Result<Self> {
+        let liveliness_pattern = match format {
+            ros_z_protocol::KeyExprFormat::RmwZenoh => {
+                format!("{ADMIN_SPACE}/{domain_id}/**")
+            }
+            ros_z_protocol::KeyExprFormat::Ros2Dds => "@/*/@ros2_lv/**".to_string(),
+            _ => {
+                return Err(zenoh::Error::from(format!(
+                    "unsupported key expression format for graph construction: {:?}",
+                    format
+                )));
+            }
+        };
+
+        Self::new_with_pattern(session, domain_id, liveliness_pattern, move |ke| {
+            format.parse_liveliness(ke)
+        })
+    }
+
     async fn wait_until<F>(&self, timeout: Duration, predicate: F) -> bool
     where
         F: Fn(&Self) -> bool,
@@ -458,17 +485,6 @@ impl Graph {
         }
     }
 
-    pub fn new(session: &Session, domain_id: usize) -> Result<Self> {
-        // Default to RmwZenoh format
-        let format = ros_z_protocol::KeyExprFormat::default();
-        Self::new_with_pattern(
-            session,
-            domain_id,
-            format!("{ADMIN_SPACE}/{domain_id}/**"),
-            move |ke| format.parse_liveliness(ke),
-        )
-    }
-
     /// Create a new Graph with a custom liveliness subscription pattern and parser
     ///
     /// # Arguments

crates/ros-z/tests/graph.rs

@@ -677,6 +677,44 @@ mod tests {
         Ok(())
     }
 
+    /// Tests that a Ros2Dds context uses a Ros2Dds graph for introspection and matching.
+    #[cfg(feature = "ros2dds")]
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_ros2dds_context_graph_tracks_local_entities() -> Result<()> {
+        let ctx = ZContextBuilder::default()
+            .keyexpr_format(ros_z_protocol::KeyExprFormat::Ros2Dds)
+            .build()?;
+        let pub_node = ctx.create_node("test_graph_pub_dds").build()?;
+        let sub_node = ctx.create_node("test_graph_sub_dds").build()?;
+        let topic_name = "/test_ros2dds_context_graph";
+
+        let publisher = pub_node.create_pub::<RosString>(topic_name).build()?;
+        let subscriber = sub_node.create_sub::<RosString>(topic_name).build()?;
+
+        assert!(
+            publisher
+                .wait_for_subscription(1, Duration::from_secs(2))
+                .await
+        );
+        assert!(
+            subscriber
+                .wait_for_publisher(1, Duration::from_secs(2))
+                .await
+        );
+
+        let graph = ctx.graph();
+        assert!(
+            graph.count(EntityKind::Publisher, topic_name) >= 1,
+            "Expected Ros2Dds graph to discover local publisher"
+        );
+        assert!(
+            graph.count(EntityKind::Subscription, topic_name) >= 1,
+            "Expected Ros2Dds graph to discover local subscriber"
+        );
+
+        Ok(())
+    }
+
     /// Tests getting action names and types from the graph
     #[tokio::test(flavor = "multi_thread")]
     async fn test_action_names_and_types() -> Result<()> {