Update up-client-zenoh-rust with up-spec 1.5.7

Cargo.toml

@@ -20,5 +20,6 @@ log = "0.4.17"
 prost = "0.12"
 prost-types = "0.12"
 protobuf = { version = "3.3" }
-up-rust = { git = "https://github.com/eclipse-uprotocol/up-rust", rev = "68c8a1d94f0006daf4ba135c9cbbfddcd793108d" }
+rand = "0.8.5"
+up-rust = { git = "https://github.com/eclipse-uprotocol/up-rust", rev = "953ac0e8da3607330d6df183437114b916fd2879" }
 zenoh = { version = "0.10.1-rc", features = ["unstable"]}

README.md

@@ -2,7 +2,7 @@
 
 Rust UPClient implementation for the Zenoh transport
 
-# Build
+## Build
 
 ```shell
 # Check clippy
@@ -13,6 +13,6 @@ cargo build
 cargo test
 ```
 
-# Note
+## Note
 
 The implementation follows the spec defined in [up-l1/zenoh](https://github.com/eclipse-uprotocol/up-spec/blob/main/up-l1/zenoh.adoc).

src/lib.rs

@@ -19,7 +19,10 @@ use std::{
     collections::HashMap,
     sync::{atomic::AtomicU64, Arc, Mutex},
 };
-use up_rust::uprotocol::{UAttributes, UCode, UMessage, UPayloadFormat, UPriority, UStatus, UUri};
+use up_rust::{
+    UAttributes, UCode, UEntity, UMessage, UPayloadFormat, UPriority, UResourceBuilder, UStatus,
+    UUri,
+};
 use zenoh::{
     config::Config,
     prelude::r#async::*,
@@ -43,18 +46,86 @@ pub struct UPClientZenoh {
     query_map: Arc<Mutex<HashMap<String, Query>>>,
     // Save the callback for RPC response
     rpc_callback_map: Arc<Mutex<HashMap<String, Arc<UtransportListener>>>>,
+    // Used to identify different callback
     callback_counter: AtomicU64,
+    // Source UUri in RPC
+    source_uuri: UUri,
 }
 
 impl UPClientZenoh {
+    /// Create `UPClientZenoh` by applying the Zenoh configuration.
+    /// The `UUri` will be generated randomly.
+    ///
+    /// # Arguments
+    ///
+    /// * `config` - Zenoh configuration. You can refer to [here](https://github.com/eclipse-zenoh/zenoh/blob/0.10.1-rc/DEFAULT_CONFIG.json5) for more configuration details.
+    ///
     /// # Errors
-    /// Will return `Err` if unable to create Zenoh session
+    /// Will return `Err` if unable to create `UPClientZenoh`
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # async_std::task::block_on(async {
+    ///     use up_client_zenoh::UPClientZenoh;
+    ///     use zenoh::config::Config;
+    ///     let upclient = UPClientZenoh::new(Config::default()).await.unwrap();
+    /// # });
+    /// ```
     pub async fn new(config: Config) -> Result<UPClientZenoh, UStatus> {
+        let uuri = UUri {
+            entity: Some(UEntity {
+                name: "default.entity".to_string(),
+                id: Some(u32::from(rand::random::<u16>())),
+                version_major: Some(1),
+                version_minor: None,
+                ..Default::default()
+            })
+            .into(),
+            ..Default::default()
+        };
+        UPClientZenoh::new_with_uuri(config, uuri).await
+    }
+
+    /// Create `UPClientZenoh` by applying the Zenoh configuration and self-defined `UUri`.
+    ///
+    /// # Arguments
+    ///
+    /// * `config` - Zenoh configuration. You can refer to [here](https://github.com/eclipse-zenoh/zenoh/blob/0.10.1-rc/DEFAULT_CONFIG.json5) for more configuration details.
+    /// * `source_uuri` - The `UUri` which is put in source while sending RPC request.
+    ///
+    /// # Errors
+    /// Will return `Err` if unable to create `UPClientZenoh`
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # async_std::task::block_on(async {
+    ///     use up_client_zenoh::UPClientZenoh;
+    ///     use up_rust::{UEntity, UUri};
+    ///     use zenoh::config::Config;
+    ///     let uuri = UUri {
+    ///         entity: Some(UEntity {
+    ///             name: "default.entity".to_string(),
+    ///             id: Some(u32::from(rand::random::<u16>())),
+    ///             version_major: Some(1),
+    ///             version_minor: None,
+    ///             ..Default::default()
+    ///         })
+    ///         .into(),
+    ///         ..Default::default()
+    ///     };
+    ///     let upclient = UPClientZenoh::new_with_uuri(Config::default(), uuri).await.unwrap();
+    /// # });
+    /// ```
+    pub async fn new_with_uuri(
+        config: Config,
+        source_uuri: UUri,
+    ) -> Result<UPClientZenoh, UStatus> {
         let Ok(session) = zenoh::open(config).res().await else {
-            return Err(UStatus::fail_with_code(
-                UCode::INTERNAL,
-                "Unable to open Zenoh session",
-            ));
+            let msg = "Unable to open Zenoh session".to_string();
+            log::error!("{msg}");
+            return Err(UStatus::fail_with_code(UCode::INTERNAL, msg));
         };
         Ok(UPClientZenoh {
             session: Arc::new(session),
@@ -63,25 +134,46 @@ impl UPClientZenoh {
             query_map: Arc::new(Mutex::new(HashMap::new())),
             rpc_callback_map: Arc::new(Mutex::new(HashMap::new())),
             callback_counter: AtomicU64::new(0),
+            source_uuri,
         })
     }
 
+    /// Get the `UUri` of `UPClientZenoh` in for RPC response
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # async_std::task::block_on(async {
+    ///     use up_client_zenoh::UPClientZenoh;
+    ///     use up_rust::{UUri, UriValidator};
+    ///     use zenoh::config::Config;
+    ///     let upclient = UPClientZenoh::new(Config::default()).await.unwrap();
+    ///     let uuri = upclient.get_response_uuri();
+    ///     assert!(uuri.authority.is_none());
+    ///     assert!(UriValidator::is_rpc_response(&uuri));
+    ///     assert_eq!(uuri.entity.unwrap().name, "default.entity");
+    /// # });
+    /// ```
+    pub fn get_response_uuri(&self) -> UUri {
+        let mut source = self.source_uuri.clone();
+        source.resource = Some(UResourceBuilder::for_rpc_response()).into();
+        source
+    }
+
     fn get_uauth_from_uuri(uri: &UUri) -> Result<String, UStatus> {
         if let Some(authority) = uri.authority.as_ref() {
             let buf: Vec<u8> = authority.try_into().map_err(|_| {
-                UStatus::fail_with_code(
-                    UCode::INVALID_ARGUMENT,
-                    "Unable to transform UAuthority into micro form",
-                )
+                let msg = "Unable to transform UAuthority into micro form".to_string();
+                log::error!("{msg}");
+                UStatus::fail_with_code(UCode::INVALID_ARGUMENT, msg)
             })?;
             Ok(buf
                 .iter()
                 .fold(String::new(), |s, c| s + &format!("{c:02x}")))
         } else {
-            Err(UStatus::fail_with_code(
-                UCode::INVALID_ARGUMENT,
-                "Empty UAuthority",
-            ))
+            let msg = "UAuthority is empty".to_string();
+            log::error!("{msg}");
+            Err(UStatus::fail_with_code(UCode::INVALID_ARGUMENT, msg))
         }
     }
 
@@ -90,11 +182,10 @@ impl UPClientZenoh {
         if uri.authority.is_some() && uri.entity.is_none() && uri.resource.is_none() {
             Ok(String::from("upr/") + &UPClientZenoh::get_uauth_from_uuri(uri)? + "/**")
         } else {
-            let micro_uuri: Vec<u8> = uri.try_into().map_err(|_| {
-                UStatus::fail_with_code(
-                    UCode::INVALID_ARGUMENT,
-                    "Unable to serialize into micro format",
-                )
+            let micro_uuri: Vec<u8> = uri.try_into().map_err(|e| {
+                let msg = format!("Unable to serialize into micro format: {e}");
+                log::error!("{msg}");
+                UStatus::fail_with_code(UCode::INVALID_ARGUMENT, msg)
             })?;
             // If the UUri is larger than 8 bytes, then it should be remote UUri with UAuthority
             // We should prepend it to the Zenoh key.
@@ -148,30 +239,28 @@ impl UPClientZenoh {
     fn attachment_to_uattributes(attachment: &Attachment) -> anyhow::Result<UAttributes> {
         let mut attachment_iter = attachment.iter();
         if let Some((_, value)) = attachment_iter.next() {
-            let version = *value.as_slice().first().ok_or(UStatus::fail_with_code(
-                UCode::INTERNAL,
-                "uAttributes version is empty",
-            ))?;
-            if version != 1 {
-                return Err(UStatus::fail_with_code(
-                    UCode::INTERNAL,
-                    "uAttributes version should be 1",
-                )
-                .into());
+            let version = *value.as_slice().first().ok_or_else(|| {
+                let msg = format!("UAttributes version is empty (should be {UATTRIBUTE_VERSION})");
+                log::error!("{msg}");
+                UStatus::fail_with_code(UCode::INVALID_ARGUMENT, msg)
+            })?;
+            if version != UATTRIBUTE_VERSION {
+                let msg =
+                    format!("UAttributes version is {version} (should be {UATTRIBUTE_VERSION})");
+                log::error!("{msg}");
+                return Err(UStatus::fail_with_code(UCode::INVALID_ARGUMENT, msg).into());
             }
         } else {
-            return Err(UStatus::fail_with_code(
-                UCode::INTERNAL,
-                "Unable to get the uAttributes version",
-            )
-            .into());
+            let msg = "Unable to get the UAttributes version".to_string();
+            log::error!("{msg}");
+            return Err(UStatus::fail_with_code(UCode::INVALID_ARGUMENT, msg).into());
         }
         let uattributes = if let Some((_, value)) = attachment_iter.next() {
             UAttributes::parse_from_bytes(value.as_slice())?
         } else {
-            return Err(
-                UStatus::fail_with_code(UCode::INTERNAL, "Unable to get the uAttributes").into(),
-            );
+            let msg = "Unable to get the UAttributes".to_string();
+            log::error!("{msg}");
+            return Err(UStatus::fail_with_code(UCode::INVALID_ARGUMENT, msg).into());
         };
         Ok(uattributes)
     }
@@ -180,7 +269,7 @@ impl UPClientZenoh {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use up_rust::uprotocol::{uri::uauthority::Number, UAuthority, UEntity, UResource, UUri};
+    use up_rust::{Number, UAuthority, UEntity, UResource, UUri};
 
     #[test]
     fn test_to_zenoh_key_string() {