[ZEN-587] Make Reply::replier_id() return an EntityGlobalId (#2052)

* Make replied_id an EntityGlobalId

* Use EntityGlobalIdProto internally

* Change replier_id of timeout errors to None

Author: oteffahi

zenoh/src/api/query.rs

@@ -17,11 +17,11 @@ use std::{collections::HashMap, error::Error, fmt::Display};
 #[cfg(feature = "unstable")]
 use serde::Deserialize;
 #[cfg(feature = "unstable")]
-use zenoh_config::ZenohId;
+use zenoh_config::wrappers::EntityGlobalId;
 use zenoh_keyexpr::OwnedKeyExpr;
-use zenoh_protocol::core::Parameters;
 #[cfg(feature = "unstable")]
-use zenoh_protocol::core::ZenohIdProto;
+use zenoh_protocol::core::EntityGlobalIdProto;
+use zenoh_protocol::core::Parameters;
 /// The [`Queryable`](crate::query::Queryable)s that should be target of a [`get`](crate::Session::get).
 pub use zenoh_protocol::network::request::ext::QueryTarget;
 #[doc(inline)]
@@ -129,7 +129,7 @@ impl Error for ReplyError {}
 pub struct Reply {
     pub(crate) result: Result<Sample, ReplyError>,
     #[cfg(feature = "unstable")]
-    pub(crate) replier_id: Option<ZenohIdProto>,
+    pub(crate) replier_id: Option<EntityGlobalIdProto>,
 }
 
 impl Reply {
@@ -149,10 +149,8 @@ impl Reply {
     }
 
     #[zenoh_macros::unstable]
-    // @TODO: maybe return an `Option<EntityGlobalId>`?
-    //
     /// Gets the id of the zenoh instance that answered this Reply.
-    pub fn replier_id(&self) -> Option<ZenohId> {
+    pub fn replier_id(&self) -> Option<EntityGlobalId> {
         self.replier_id.map(Into::into)
     }
 

zenoh/src/api/session.rs

@@ -2275,7 +2275,7 @@ impl SessionInner {
                                 query.callback.call(Reply {
                                     result: Err(ReplyError::new("Timeout", Encoding::ZENOH_STRING)),
                                     #[cfg(feature = "unstable")]
-                                    replier_id: Some(session.zid().into()),
+                                    replier_id: None
                                 });
                             }
                         }
@@ -2374,7 +2374,7 @@ impl SessionInner {
                                 query.callback.call(Reply {
                                     result: Err(ReplyError::new("Timeout", Encoding::ZENOH_STRING)),
                                     #[cfg(feature = "unstable")]
-                                    replier_id: Some(session.zid().into()),
+                                    replier_id: None
                                 });
                             }
                         }
@@ -2814,7 +2814,12 @@ impl Primitives for WeakSession {
                                 encoding: mem::take(&mut e.encoding).into(),
                             }),
                             #[cfg(feature = "unstable")]
-                            replier_id: mem::take(&mut msg.ext_respid).map(|rid| rid.zid),
+                            replier_id: mem::take(&mut msg.ext_respid).map(|rid| {
+                                zenoh_protocol::core::EntityGlobalIdProto {
+                                    zid: rid.zid,
+                                    eid: rid.eid,
+                                }
+                            }),
                         };
                         callback.call(new_reply);
                     }
@@ -2906,7 +2911,12 @@ impl Primitives for WeakSession {
                         let new_reply = Reply {
                             result: Ok(sample),
                             #[cfg(feature = "unstable")]
-                            replier_id: mem::take(&mut msg.ext_respid).map(|rid| rid.zid),
+                            replier_id: mem::take(&mut msg.ext_respid).map(|rid| {
+                                zenoh_protocol::core::EntityGlobalIdProto {
+                                    zid: rid.zid,
+                                    eid: rid.eid,
+                                }
+                            }),
                         };
                         let callback =
                             match query.reception_mode {

zenoh/tests/session.rs

@@ -249,7 +249,7 @@ async fn test_session_query_reply_internal<Getter: HasGet>(
             let rs = getter.get("ok_put").await;
             while let Ok(s) = ztimeout!(rs.recv_async()) {
                 #[cfg(feature = "unstable")]
-                assert_eq!(s.replier_id(), Some(qbl.id().zid()));
+                assert_eq!(s.replier_id(), Some(qbl.id()));
                 let s = s.result().unwrap();
                 assert_eq!(s.kind(), SampleKind::Put);
                 assert_eq!(s.payload().len(), size);
@@ -268,7 +268,7 @@ async fn test_session_query_reply_internal<Getter: HasGet>(
             let rs = getter.get("ok_del").await;
             while let Ok(s) = ztimeout!(rs.recv_async()) {
                 #[cfg(feature = "unstable")]
-                assert_eq!(s.replier_id(), Some(qbl.id().zid()));
+                assert_eq!(s.replier_id(), Some(qbl.id()));
                 let s = s.result().unwrap();
                 assert_eq!(s.kind(), SampleKind::Delete);
                 assert_eq!(s.payload().len(), 0);
@@ -287,7 +287,7 @@ async fn test_session_query_reply_internal<Getter: HasGet>(
             let rs = getter.get("err").await;
             while let Ok(s) = ztimeout!(rs.recv_async()) {
                 #[cfg(feature = "unstable")]
-                assert_eq!(s.replier_id(), Some(qbl.id().zid()));
+                assert_eq!(s.replier_id(), Some(qbl.id()));
                 let e = s.result().unwrap_err();
                 assert_eq!(e.payload().len(), size);
                 cnt += 1;