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

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

@@ -431,11 +431,3 @@ 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,17 +331,6 @@ 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

@@ -902,6 +902,12 @@ impl Priority {
     pub const NUM: usize = 1 + Self::MIN as usize - Self::MAX as usize;
 }
 
+impl From<zenoh_protocol::core::Priority> for Priority {
+    fn from(priority: zenoh_protocol::core::Priority) -> Self {
+        Self::try_from(priority as u8).unwrap()
+    }
+}
+
 impl TryFrom<u8> for Priority {
     type Error = zenoh_result::Error;
 

zenoh/src/sample.rs

@@ -18,10 +18,12 @@ use crate::prelude::ZenohId;
 use crate::prelude::{KeyExpr, SampleKind, Value};
 use crate::query::Reply;
 use crate::time::{new_reception_timestamp, Timestamp};
+use crate::Priority;
 #[zenoh_macros::unstable]
 use serde::Serialize;
 use std::convert::{TryFrom, TryInto};
-use zenoh_protocol::core::{Encoding, QoS};
+use zenoh_protocol::core::{CongestionControl, Encoding};
+use zenoh_protocol::network::push::ext::QoSType;
 
 pub type SourceSn = u64;
 
@@ -516,3 +518,21 @@ impl TryFrom<Reply> for Sample {
         value.sample
     }
 }
+
+/// 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,
+}
+
+impl From<QoSType> for QoS {
+    fn from(ext: QoSType) -> Self {
+        QoS {
+            priority: ext.get_priority().into(),
+            congestion_control: ext.get_congestion_control(),
+            express: ext.is_express(),
+        }
+    }
+}

zenoh/tests/qos.rs

@@ -1,45 +1,66 @@
-use std::time::Duration;
-
+//
+// Copyright (c) 2024 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <[email protected]>
+//
+use async_std::prelude::FutureExt;
 use async_std::task;
+use std::time::Duration;
+use zenoh::prelude::r#async::*;
 use zenoh::{publication::Priority, SessionDeclarations};
-use zenoh_core::{zasync_executor_init, AsyncResolve};
-use zenoh_protocol::core::CongestionControl;
+use zenoh_core::zasync_executor_init;
 
+const TIMEOUT: Duration = Duration::from_secs(60);
 const SLEEP: Duration = Duration::from_secs(1);
 
+macro_rules! ztimeout {
+    ($f:expr) => {
+        $f.timeout(TIMEOUT).await.unwrap()
+    };
+}
+
 #[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 session1 = ztimeout!(zenoh::open(zenoh_config::peer()).res_async()).unwrap();
+        let session2 = ztimeout!(zenoh::open(zenoh_config::peer()).res_async()).unwrap();
 
-        let publisher = session1
+        let publisher1 = ztimeout!(session1
             .declare_publisher("test/qos")
-            .res()
-            .await
-            .unwrap()
             .priority(Priority::DataHigh)
-            .congestion_control(CongestionControl::Drop);
+            .congestion_control(CongestionControl::Drop)
+            .res())
+        .unwrap();
 
-        task::sleep(SLEEP).await;
+        let publisher2 = ztimeout!(session1
+            .declare_publisher("test/qos")
+            .priority(Priority::DataLow)
+            .congestion_control(CongestionControl::Block)
+            .res())
+        .unwrap();
 
-        let sub = session2.declare_subscriber("test/qos").res().await.unwrap();
+        let subscriber = ztimeout!(session2.declare_subscriber("test/qos").res()).unwrap();
         task::sleep(SLEEP).await;
 
-        publisher.put("qos").res_async().await.unwrap();
-        let qos = sub.recv_async().await.unwrap().qos;
+        ztimeout!(publisher1.put("qos").res_async()).unwrap();
+        let qos = ztimeout!(subscriber.recv_async()).unwrap().qos;
 
-        assert_eq!(qos.priority, Priority::DataHigh.into());
+        assert_eq!(qos.priority, Priority::DataHigh);
         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;
+        ztimeout!(publisher2.put("qos").res_async()).unwrap();
+        let qos = ztimeout!(subscriber.recv_async()).unwrap().qos;
 
-        assert_eq!(qos.priority, Priority::DataLow.into());
+        assert_eq!(qos.priority, Priority::DataLow);
         assert_eq!(qos.congestion_control, CongestionControl::Block);
     });
 }