Merge pull request #926 from eclipse-zenoh/protocol_valgrind

Fix ci/valgrind

ci/valgrind-check/src/pub_sub/bin/z_pub_sub.rs

@@ -38,9 +38,12 @@ async fn main() {
         .callback(|sample| {
             println!(
                 ">> [Subscriber] Received {} ('{}': '{}')",
-                sample.kind,
-                sample.key_expr.as_str(),
-                sample.value
+                sample.kind(),
+                sample.key_expr().as_str(),
+                sample
+                    .payload()
+                    .deserialize::<String>()
+                    .unwrap_or_else(|e| format!("{}", e))
             );
         })
         .res()

ci/valgrind-check/src/queryable_get/bin/z_queryable_get.rs

@@ -30,13 +30,16 @@ async fn main() {
         .declare_queryable(&queryable_key_expr.clone())
         .callback(move |query| {
             println!(">> Handling query '{}'", query.selector());
-            let reply = Ok(Sample::new(
-                queryable_key_expr.clone(),
-                query.value().unwrap().clone(),
-            ));
-            zenoh_runtime::ZRuntime::Application.block_in_place(
-                async move { query.reply(reply).res().await.unwrap(); }
-            );
+            zenoh_runtime::ZRuntime::Application.block_in_place(async move {
+                query
+                    .reply(
+                        query.selector().key_expr,
+                        query.value().unwrap().payload.clone(),
+                    )
+                    .res()
+                    .await
+                    .unwrap();
+            });
         })
         .complete(true)
         .res()
@@ -51,7 +54,7 @@ async fn main() {
         println!("Sending Query '{get_selector}'...");
         let replies = get_session
             .get(&get_selector)
-            .with_value(idx)
+            .value(idx)
             .target(QueryTarget::All)
             .res()
             .await
@@ -60,10 +63,18 @@ async fn main() {
             match reply.sample {
                 Ok(sample) => println!(
                     ">> Received ('{}': '{}')",
-                    sample.key_expr.as_str(),
-                    sample.value,
+                    sample.key_expr().as_str(),
+                    sample
+                        .payload()
+                        .deserialize::<String>()
+                        .unwrap_or_else(|e| format!("{}", e))
+                ),
+                Err(err) => println!(
+                    ">> Received (ERROR: '{}')",
+                    err.payload
+                        .deserialize::<String>()
+                        .unwrap_or_else(|e| format!("{}", e))
                 ),
-                Err(err) => println!(">> Received (ERROR: '{}')", String::try_from(&err).unwrap()),
             }
         }
     }

zenoh/src/publication.rs

@@ -14,16 +14,15 @@
 
 //! Publishing primitives.
 use crate::net::primitives::Primitives;
-use crate::payload::OptionPayload;
 use crate::prelude::*;
-#[zenoh_macros::unstable]
-use crate::sample::Attachment;
 use crate::sample::{DataInfo, QoS, Sample, SampleFields, SampleKind};
 use crate::SessionRef;
 use crate::Undeclarable;
 #[cfg(feature = "unstable")]
 use crate::{
     handlers::{Callback, DefaultHandler, IntoHandler},
+    payload::OptionPayload,
+    sample::Attachment,
     Id,
 };
 use std::future::Ready;

zenoh/src/query.rs

@@ -14,12 +14,11 @@
 
 //! Query primitives.
 use crate::handlers::{locked, Callback, DefaultHandler};
-use crate::payload::OptionPayload;
 use crate::prelude::*;
-#[zenoh_macros::unstable]
-use crate::sample::Attachment;
 use crate::sample::QoSBuilder;
 use crate::Session;
+#[cfg(feature = "unstable")]
+use crate::{payload::OptionPayload, sample::Attachment};
 use std::collections::HashMap;
 use std::future::Ready;
 use std::time::Duration;

zenoh/src/queryable.rs

@@ -17,17 +17,18 @@
 use crate::encoding::Encoding;
 use crate::handlers::{locked, DefaultHandler};
 use crate::net::primitives::Primitives;
-use crate::payload::OptionPayload;
 use crate::prelude::*;
 use crate::sample::builder::SampleBuilder;
 use crate::sample::QoSBuilder;
-#[cfg(feature = "unstable")]
-use crate::sample::SourceInfo;
 use crate::Id;
 use crate::SessionRef;
 use crate::Undeclarable;
 #[cfg(feature = "unstable")]
-use crate::{query::ReplyKeyExpr, sample::Attachment};
+use crate::{
+    payload::OptionPayload,
+    query::ReplyKeyExpr,
+    sample::{Attachment, SourceInfo},
+};
 use std::fmt;
 use std::future::Ready;
 use std::ops::Deref;

zenoh/src/sample/builder.rs

@@ -11,12 +11,6 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
-
-use std::marker::PhantomData;
-
-use crate::payload::OptionPayload;
-#[cfg(feature = "unstable")]
-use crate::sample::SourceInfo;
 use crate::sample::{QoS, QoSBuilder};
 use crate::Encoding;
 use crate::KeyExpr;
@@ -25,6 +19,9 @@ use crate::Priority;
 use crate::Sample;
 use crate::SampleKind;
 use crate::Value;
+#[cfg(feature = "unstable")]
+use crate::{payload::OptionPayload, sample::SourceInfo};
+use std::marker::PhantomData;
 use uhlc::Timestamp;
 use zenoh_core::zresult;
 use zenoh_protocol::core::CongestionControl;

zenoh/src/sample/mod.rs

@@ -19,7 +19,7 @@ use crate::prelude::{KeyExpr, Value};
 use crate::sample::builder::{QoSBuilderTrait, ValueBuilderTrait};
 use crate::time::Timestamp;
 use crate::Priority;
-#[zenoh_macros::unstable]
+#[cfg(feature = "unstable")]
 use serde::Serialize;
 use std::{convert::TryFrom, fmt};
 use zenoh_protocol::core::CongestionControl;
@@ -212,8 +212,9 @@ impl From<Option<DataInfo>> for SourceInfo {
 }
 
 mod attachment {
-    use crate::Payload;
-    #[zenoh_macros::unstable]
+    #[cfg(feature = "unstable")]
+    use crate::payload::Payload;
+    #[cfg(feature = "unstable")]
     use zenoh_protocol::zenoh::ext::AttachmentType;
 
     #[zenoh_macros::unstable]