add support for errors in replies

commons/zenoh-config/src/lib.rs

@@ -623,7 +623,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),
                     _ => (),
                 }
             }

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(())
+            }
         }
     }
 }