Expose Quality of Service settings (Priority, Congestion Control, Express) the Sample was published with

commons/zenoh-protocol/src/core/mod.rs

@@ -431,3 +431,11 @@ pub enum QueryTarget {
     #[cfg(feature = "complete_n")]
     Complete(u64),
 }
+
+/// Structure containing quality of service data
+#[derive(Debug, Default, Copy, Clone, Eq, PartialEq)]
+pub struct QoS {
+    pub priority: Priority,
+    pub congestion_control: CongestionControl,
+    pub express: bool,
+}

commons/zenoh-protocol/src/network/mod.rs

@@ -331,6 +331,17 @@ pub mod ext {
         }
     }
 
+    type QoSPublic = crate::core::QoS;
+    impl<const ID: u8> From<QoSType<{ ID }>> for QoSPublic {
+        fn from(ext: QoSType<{ ID }>) -> Self {
+            QoSPublic {
+                priority: ext.get_priority(),
+                congestion_control: ext.get_congestion_control(),
+                express: ext.is_express(),
+            }
+        }
+    }
+
     /// ```text
     ///  7 6 5 4 3 2 1 0
     /// +-+-+-+-+-+-+-+-+

zenoh/src/publication.rs

@@ -857,7 +857,14 @@ fn resolve_put(
             kind,
             encoding: Some(value.encoding),
             timestamp,
-            ..Default::default()
+            source_id: None,
+            source_sn: None,
+            qos: ext::QoSType::new(
+                publisher.priority.into(),
+                publisher.congestion_control,
+                false,
+            )
+            .into(),
         };
 
         publisher.session.handle_data(

zenoh/src/queryable.rs

@@ -193,6 +193,7 @@ impl SyncResolve for ReplyBuilder<'_> {
                     value: Value { payload, encoding },
                     kind,
                     timestamp,
+                    qos,
                     #[cfg(feature = "unstable")]
                     source_info,
                     #[cfg(feature = "unstable")]
@@ -203,6 +204,7 @@ impl SyncResolve for ReplyBuilder<'_> {
                     kind,
                     encoding: Some(encoding),
                     timestamp,
+                    qos,
                     source_id: None,
                     source_sn: None,
                 };

zenoh/src/sample.rs

@@ -21,7 +21,7 @@ use crate::time::{new_reception_timestamp, Timestamp};
 #[zenoh_macros::unstable]
 use serde::Serialize;
 use std::convert::{TryFrom, TryInto};
-use zenoh_protocol::core::Encoding;
+use zenoh_protocol::core::{Encoding, QoS};
 
 pub type SourceSn = u64;
 
@@ -50,6 +50,7 @@ pub(crate) struct DataInfo {
     pub timestamp: Option<Timestamp>,
     pub source_id: Option<ZenohId>,
     pub source_sn: Option<SourceSn>,
+    pub qos: QoS,
 }
 
 /// Informations on the source of a zenoh [`Sample`].
@@ -326,6 +327,8 @@ pub struct Sample {
     pub kind: SampleKind,
     /// The [`Timestamp`] of this Sample.
     pub timestamp: Option<Timestamp>,
+    /// Quality of service settings this sample was sent with.
+    pub qos: QoS,
 
     #[cfg(feature = "unstable")]
     /// <div class="stab unstable">
@@ -361,6 +364,7 @@ impl Sample {
             value: value.into(),
             kind: SampleKind::default(),
             timestamp: None,
+            qos: QoS::default(),
             #[cfg(feature = "unstable")]
             source_info: SourceInfo::empty(),
             #[cfg(feature = "unstable")]
@@ -383,6 +387,7 @@ impl Sample {
             value: value.into(),
             kind: SampleKind::default(),
             timestamp: None,
+            qos: QoS::default(),
             #[cfg(feature = "unstable")]
             source_info: SourceInfo::empty(),
             #[cfg(feature = "unstable")]
@@ -407,6 +412,7 @@ impl Sample {
                 value,
                 kind: data_info.kind,
                 timestamp: data_info.timestamp,
+                qos: data_info.qos,
                 #[cfg(feature = "unstable")]
                 source_info: data_info.into(),
                 #[cfg(feature = "unstable")]
@@ -418,6 +424,7 @@ impl Sample {
                 value,
                 kind: SampleKind::default(),
                 timestamp: None,
+                qos: QoS::default(),
                 #[cfg(feature = "unstable")]
                 source_info: SourceInfo::empty(),
                 #[cfg(feature = "unstable")]

zenoh/src/session.rs

@@ -2192,6 +2192,7 @@ impl Primitives for Session {
                     kind: SampleKind::Put,
                     encoding: Some(m.encoding),
                     timestamp: m.timestamp,
+                    qos: msg.ext_qos.into(),
                     source_id: m.ext_sinfo.as_ref().map(|i| i.zid),
                     source_sn: m.ext_sinfo.as_ref().map(|i| i.sn as u64),
                 };
@@ -2209,6 +2210,7 @@ impl Primitives for Session {
                     kind: SampleKind::Delete,
                     encoding: None,
                     timestamp: m.timestamp,
+                    qos: msg.ext_qos.into(),
                     source_id: m.ext_sinfo.as_ref().map(|i| i.zid),
                     source_sn: m.ext_sinfo.as_ref().map(|i| i.sn as u64),
                 };
@@ -2345,6 +2347,7 @@ impl Primitives for Session {
                             kind: SampleKind::Put,
                             encoding: Some(m.encoding),
                             timestamp: m.timestamp,
+                            qos: msg.ext_qos.into(),
                             source_id: m.ext_sinfo.as_ref().map(|i| i.zid),
                             source_sn: m.ext_sinfo.as_ref().map(|i| i.sn as u64),
                         };

zenoh/tests/qos.rs

@@ -0,0 +1,45 @@
+use std::time::Duration;
+
+use async_std::task;
+use zenoh::{publication::Priority, SessionDeclarations};
+use zenoh_core::{zasync_executor_init, AsyncResolve};
+use zenoh_protocol::core::CongestionControl;
+
+const SLEEP: Duration = Duration::from_secs(1);
+
+#[test]
+fn pubsub() {
+    task::block_on(async {
+        zasync_executor_init!();
+        let session1 = zenoh::open(zenoh_config::peer()).res_async().await.unwrap();
+        let session2 = zenoh::open(zenoh_config::peer()).res_async().await.unwrap();
+
+        let publisher = session1
+            .declare_publisher("test/qos")
+            .res()
+            .await
+            .unwrap()
+            .priority(Priority::DataHigh)
+            .congestion_control(CongestionControl::Drop);
+
+        task::sleep(SLEEP).await;
+
+        let sub = session2.declare_subscriber("test/qos").res().await.unwrap();
+        task::sleep(SLEEP).await;
+
+        publisher.put("qos").res_async().await.unwrap();
+        let qos = sub.recv_async().await.unwrap().qos;
+
+        assert_eq!(qos.priority, Priority::DataHigh.into());
+        assert_eq!(qos.congestion_control, CongestionControl::Drop);
+
+        let publisher = publisher
+            .priority(Priority::DataLow)
+            .congestion_control(CongestionControl::Block);
+        publisher.put("qos").res_async().await.unwrap();
+        let qos = sub.recv_async().await.unwrap().qos;
+
+        assert_eq!(qos.priority, Priority::DataLow.into());
+        assert_eq!(qos.congestion_control, CongestionControl::Block);
+    });
+}