Merge branch 'master' into simplified_load_plugin

commons/zenoh-config/src/lib.rs

@@ -629,7 +629,9 @@ impl Config {
 
 impl std::fmt::Display for Config {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", serde_json::to_string(self).unwrap())
+        let mut json = serde_json::to_value(self).unwrap();
+        sift_privates(&mut json);
+        write!(f, "{json}")
     }
 }
 

examples/examples/z_queryable.rs

@@ -16,6 +16,7 @@ use clap::{App, Arg};
 use futures::prelude::*;
 use futures::select;
 use std::convert::TryFrom;
+use std::sync::atomic::Ordering::Relaxed;
 use std::time::Duration;
 use zenoh::config::Config;
 use zenoh::prelude::r#async::*;
@@ -26,6 +27,7 @@ async fn main() {
     env_logger::init();
 
     let (config, key_expr, value, complete) = parse_args();
+    let send_errors = std::sync::atomic::AtomicBool::new(false);
 
     let key_expr = KeyExpr::try_from(key_expr).unwrap();
     println!("Opening session...");
@@ -39,7 +41,7 @@ async fn main() {
         .await
         .unwrap();
 
-    println!("Enter 'q' to quit...");
+    println!("Enter 'q' to quit, 'e' to reply an error to next query...");
     let mut stdin = async_std::io::stdin();
     let mut input = [0_u8];
     loop {
@@ -50,8 +52,22 @@ async fn main() {
                     None => println!(">> [Queryable ] Received Query '{}'", query.selector()),
                     Some(value) => println!(">> [Queryable ] Received Query '{}' with value '{}'", query.selector(), value),
                 }
+                let reply = if send_errors.swap(false, Relaxed) {
+                    println!(
+                        ">> [Queryable ] Replying (ERROR: '{}')",
+                        value,
+                    );
+                    Err(value.clone().into())
+                } else {
+                    println!(
+                        ">> [Queryable ] Responding ('{}': '{}')",
+                        key_expr.as_str(),
+                        value,
+                    );
+                    Ok(Sample::new(key_expr.clone(), value.clone()))
+                };
                 query
-                    .reply(Ok(Sample::new(key_expr.clone(), value.clone())))
+                    .reply(reply)
                     .res()
                     .await
                     .unwrap_or_else(|e| println!(">> [Queryable ] Error sending reply: {e}"));
@@ -61,6 +77,7 @@ async fn main() {
                 match input[0] {
                     b'q' => break,
                     0 => sleep(Duration::from_secs(1)).await,
+                    b'e' => send_errors.store(true, Relaxed),
                     _ => (),
                 }
             }

examples/examples/z_scout.rs

@@ -36,5 +36,5 @@ async fn main() {
     .await;
 
     // stop scouting
-    drop(receiver);
+    receiver.stop();
 }

zenoh-ext/src/session_ext.rs

@@ -19,16 +19,20 @@ use zenoh::{Session, SessionRef};
 
 /// Some extensions to the [`zenoh::Session`](zenoh::Session)
 pub trait SessionExt {
+    type PublicationCacheBuilder<'a, 'b, 'c>
+    where
+        Self: 'a;
     fn declare_publication_cache<'a, 'b, 'c, TryIntoKeyExpr>(
         &'a self,
         pub_key_expr: TryIntoKeyExpr,
-    ) -> PublicationCacheBuilder<'a, 'b, 'c>
+    ) -> Self::PublicationCacheBuilder<'a, 'b, 'c>
     where
         TryIntoKeyExpr: TryInto<KeyExpr<'b>>,
         <TryIntoKeyExpr as TryInto<KeyExpr<'b>>>::Error: Into<zenoh_result::Error>;
 }
 
 impl SessionExt for Session {
+    type PublicationCacheBuilder<'a, 'b, 'c> = PublicationCacheBuilder<'a, 'b, 'c>;
     fn declare_publication_cache<'a, 'b, 'c, TryIntoKeyExpr>(
         &'a self,
         pub_key_expr: TryIntoKeyExpr,
@@ -44,6 +48,20 @@ impl SessionExt for Session {
     }
 }
 
+impl<T: ArcSessionExt + 'static> SessionExt for T {
+    type PublicationCacheBuilder<'a, 'b, 'c> = PublicationCacheBuilder<'static, 'b, 'c>;
+    fn declare_publication_cache<'a, 'b, 'c, TryIntoKeyExpr>(
+        &'a self,
+        pub_key_expr: TryIntoKeyExpr,
+    ) -> Self::PublicationCacheBuilder<'a, 'b, 'c>
+    where
+        TryIntoKeyExpr: TryInto<KeyExpr<'b>>,
+        <TryIntoKeyExpr as TryInto<KeyExpr<'b>>>::Error: Into<zenoh_result::Error>,
+    {
+        ArcSessionExt::declare_publication_cache(self, pub_key_expr)
+    }
+}
+
 pub trait ArcSessionExt {
     fn declare_publication_cache<'b, 'c, TryIntoKeyExpr>(
         &self,

zenoh/src/queryable.rs

@@ -28,6 +28,7 @@ use std::sync::Arc;
 use zenoh_core::{AsyncResolve, Resolvable, SyncResolve};
 use zenoh_protocol::core::WireExpr;
 use zenoh_protocol::network::{response, Mapping, RequestId, Response, ResponseFinal};
+use zenoh_protocol::zenoh::ext::ValueType;
 use zenoh_protocol::zenoh::reply::ext::ConsolidationType;
 use zenoh_protocol::zenoh::{self, ResponseBody};
 use zenoh_result::ZResult;
@@ -198,7 +199,36 @@ impl SyncResolve for ReplyBuilder<'_> {
                 });
                 Ok(())
             }
-            Err(_) => Err(zerror!("Replying errors is not yet supported!").into()),
+            Err(payload) => {
+                self.query.inner.primitives.send_response(Response {
+                    rid: self.query.inner.qid,
+                    wire_expr: WireExpr {
+                        scope: 0,
+                        suffix: std::borrow::Cow::Owned(self.query.key_expr().as_str().to_owned()),
+                        mapping: Mapping::Sender,
+                    },
+                    payload: ResponseBody::Err(zenoh::Err {
+                        timestamp: None,
+                        is_infrastructure: false,
+                        ext_sinfo: None,
+                        ext_unknown: vec![],
+                        ext_body: Some(ValueType {
+                            #[cfg(feature = "shared-memory")]
+                            ext_shm: None,
+                            payload: payload.payload,
+                            encoding: payload.encoding,
+                        }),
+                        code: 0, // TODO
+                    }),
+                    ext_qos: response::ext::QoSType::response_default(),
+                    ext_tstamp: None,
+                    ext_respid: Some(response::ext::ResponderIdType {
+                        zid: self.query.inner.zid,
+                        eid: 0, // TODO
+                    }),
+                });
+                Ok(())
+            }
         }
     }
 }

zenoh/src/scouting.rs

@@ -220,7 +220,8 @@ impl ScoutInner {
     /// # })
     /// ```
     pub fn stop(self) {
-        // drop
+        // This drops the inner `stop_sender` and hence stops the scouting receiver
+        std::mem::drop(self);
     }
 }