refacto(zenoh builtin): deserialize via a HashMap

This change allows having subscribers and publishers that include special
characters such as '*'.

The previous behavior was deriving the output and input port id from the key
expression which forbade the use of these special characters. It defeated the
point of using Zenoh as our communication middleware.

Signed-off-by: Julien Loudet <julien.loudet@zettascale.tech>

zenoh-flow-descriptors/src/nodes/builtin/zenoh.rs

@@ -39,19 +39,19 @@ use zenoh_keyexpr::OwnedKeyExpr;
 /// let yaml_description = r#"
 /// description: My zenoh source
 /// zenoh-subscribers:
-///   - rt/*/cmd_vel
-///   - rt/*/status
+///   "cmd_vel": "rt/*/cmd_vel"
+///   "status": "rt/*/status"
 /// "#;
 ///
 /// let z_source_yaml = serde_yaml::from_str::<ZenohSourceDescriptor>(yaml_description).unwrap();
 ///
 /// let json_description = r#"
 /// {
 ///    "description": "My zenoh source",
-///    "zenoh-subscribers": [
-///      "rt/*/cmd_vel",
-///      "rt/*/status"
-///    ]
+///    "zenoh-subscribers": {
+///      "cmd_vel": "rt/*/cmd_vel",
+///      "status": "rt/*/status"
+///    }
 /// }
 /// "#;
 ///
@@ -85,19 +85,19 @@ pub struct ZenohSourceDescriptor {
 /// let yaml_description = r#"
 /// description: My zenoh sink
 /// zenoh-publishers:
-///   - rt/cmd_vel
-///   - rt/status
+///   cmd_vel: rt/cmd_vel
+///   status: rt/status
 /// "#;
 ///
 /// let z_sink_yaml = serde_yaml::from_str::<ZenohSinkDescriptor>(yaml_description).unwrap();
 ///
 /// let json_description = r#"
 /// {
 ///    "description": "My zenoh sink",
-///    "zenoh-publishers": [
-///      "rt/cmd_vel",
-///      "rt/status"
-///    ]
+///    "zenoh-publishers": {
+///      "cmd_vel": "rt/cmd_vel",
+///      "status": "rt/status"
+///    }
 /// }
 /// "#;
 ///
@@ -112,20 +112,19 @@ pub struct ZenohSinkDescriptor {
     pub publishers: HashMap<PortId, OwnedKeyExpr>,
 }
 
-// Transforms a Vec<String> into a HashMap<PortId, OwnedKeyExpr>.
-//
-//
+// Transforms a HashMap<String, String> into a HashMap<PortId, OwnedKeyExpr>.
 fn deserialize_canon<'de, D>(
     deserializer: D,
 ) -> std::result::Result<HashMap<PortId, OwnedKeyExpr>, D::Error>
 where
     D: Deserializer<'de>,
 {
-    let key_expressions: Vec<String> = serde::de::Deserialize::deserialize(deserializer)?;
+    let key_expressions: HashMap<String, String> =
+        serde::de::Deserialize::deserialize(deserializer)?;
     let mut h_map = HashMap::with_capacity(key_expressions.len());
     let mut h_set = HashSet::with_capacity(key_expressions.len());
 
-    for key_expr in key_expressions {
+    for (port_id, key_expr) in key_expressions {
         let owned_canon_ke = OwnedKeyExpr::autocanonize(key_expr.clone()).map_err(|e| {
             serde::de::Error::custom(format!(
                 "Failed to autocanonize key expression < {} >:\n{:?}",
@@ -157,7 +156,7 @@ https://github.com/eclipse-zenoh/roadmap/blob/main/rfcs/ALL/Key%20Expressions.md
             );
         }
 
-        h_map.insert(key_expr.into(), owned_canon_ke);
+        h_map.insert(port_id.into(), owned_canon_ke);
     }
 
     Ok(h_map)

zenoh-flow-descriptors/src/nodes/sink.rs

@@ -76,8 +76,8 @@ use zenoh_flow_commons::{Configuration, NodeId, PortId};
 /// id: my-sink-0
 /// description: My zenoh sink
 /// zenoh-publishers:
-///   - key/expr/0
-///   - key/expr/1
+///   key_0: key/expr/0
+///   key_1: key/expr/1
 /// "#;
 ///
 /// assert!(serde_yaml::from_str::<SinkDescriptor>(sink_desc_zenoh).is_ok());

zenoh-flow-descriptors/src/nodes/source.rs

@@ -76,13 +76,12 @@ use zenoh_flow_commons::{Configuration, NodeId, PortId};
 /// id: my-source-0
 /// description: My zenoh source
 /// zenoh-subscribers:
-///   - key/expr/0
-///   - key/expr/1
+///   ke-0: key/expr/0
+///   ke-1: key/expr/1
 /// "#;
 ///
 /// assert!(serde_yaml::from_str::<SourceDescriptor>(source_desc_zenoh).is_ok());
 /// ```
-///
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
 pub struct SourceDescriptor {
     pub id: NodeId,