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

zenoh/src/publication.rs

@@ -22,6 +22,7 @@ use crate::prelude::*;
 #[zenoh_macros::unstable]
 use crate::sample::Attachment;
 use crate::sample::DataInfo;
+use crate::sample::QoS;
 use crate::Encoding;
 use crate::SessionRef;
 use crate::Undeclarable;
@@ -859,12 +860,11 @@ fn resolve_put(
             timestamp,
             source_id: None,
             source_sn: None,
-            qos: ext::QoSType::new(
+            qos: QoS::from_or_default(ext::QoSType::new(
                 publisher.priority.into(),
                 publisher.congestion_control,
                 false,
-            )
-            .into(),
+            )),
         };
 
         publisher.session.handle_data(
@@ -902,12 +902,6 @@ 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;
 
@@ -938,7 +932,8 @@ impl TryFrom<u8> for Priority {
     }
 }
 
-impl From<Priority> for zenoh_protocol::core::Priority {
+type ProtocolPriority = zenoh_protocol::core::Priority;
+impl From<Priority> for ProtocolPriority {
     fn from(prio: Priority) -> Self {
         // The Priority in the prelude differs from the Priority in the core protocol only from
         // the missing Control priority. The Control priority is reserved for zenoh internal use
@@ -952,6 +947,22 @@ impl From<Priority> for zenoh_protocol::core::Priority {
     }
 }
 
+impl TryFrom<ProtocolPriority> for Priority {
+    type Error = zenoh_result::Error;
+    fn try_from(priority: ProtocolPriority) -> Result<Self, Self::Error> {
+        match priority {
+            ProtocolPriority::Control => bail!("'Control' is not a valid priority value."),
+            ProtocolPriority::RealTime => Ok(Priority::RealTime),
+            ProtocolPriority::InteractiveHigh => Ok(Priority::InteractiveHigh),
+            ProtocolPriority::InteractiveLow => Ok(Priority::InteractiveLow),
+            ProtocolPriority::DataHigh => Ok(Priority::DataHigh),
+            ProtocolPriority::Data => Ok(Priority::Data),
+            ProtocolPriority::DataLow => Ok(Priority::DataLow),
+            ProtocolPriority::Background => Ok(Priority::Background),
+        }
+    }
+}
+
 /// A struct that indicates if there exist Subscribers matching the Publisher's key expression.
 ///
 /// # Examples

zenoh/src/sample.rs

@@ -527,12 +527,26 @@ pub struct QoS {
     pub express: bool,
 }
 
-impl From<QoSType> for QoS {
-    fn from(ext: QoSType) -> Self {
-        QoS {
-            priority: ext.get_priority().into(),
+impl QoS {
+    /// Helper function to fallback to default QoS value and log a error in case of conversion failure
+    pub(crate) fn from_or_default(ext: QoSType) -> QoS {
+        match QoS::try_from(ext) {
+            Ok(qos) => return qos,
+            Err(e) => {
+                log::error!("Failed to convert: {}", e.to_string());
+                QoS::default()
+            }
+        }
+    }
+}
+
+impl TryFrom<QoSType> for QoS {
+    type Error = zenoh_result::Error;
+    fn try_from(ext: QoSType) -> Result<Self, Self::Error> {
+        Ok(QoS {
+            priority: ext.get_priority().try_into()?,
             congestion_control: ext.get_congestion_control(),
             express: ext.is_express(),
-        }
+        })
     }
 }

zenoh/src/session.rs

@@ -31,6 +31,7 @@ use crate::queryable::*;
 #[cfg(feature = "unstable")]
 use crate::sample::Attachment;
 use crate::sample::DataInfo;
+use crate::sample::QoS;
 use crate::selector::TIME_RANGE_KEY;
 use crate::subscriber::*;
 use crate::Id;
@@ -2192,7 +2193,7 @@ impl Primitives for Session {
                     kind: SampleKind::Put,
                     encoding: Some(m.encoding),
                     timestamp: m.timestamp,
-                    qos: msg.ext_qos.into(),
+                    qos: QoS::from_or_default(msg.ext_qos),
                     source_id: m.ext_sinfo.as_ref().map(|i| i.zid),
                     source_sn: m.ext_sinfo.as_ref().map(|i| i.sn as u64),
                 };
@@ -2210,7 +2211,7 @@ impl Primitives for Session {
                     kind: SampleKind::Delete,
                     encoding: None,
                     timestamp: m.timestamp,
-                    qos: msg.ext_qos.into(),
+                    qos: QoS::from_or_default(msg.ext_qos),
                     source_id: m.ext_sinfo.as_ref().map(|i| i.zid),
                     source_sn: m.ext_sinfo.as_ref().map(|i| i.sn as u64),
                 };
@@ -2347,7 +2348,7 @@ impl Primitives for Session {
                             kind: SampleKind::Put,
                             encoding: Some(m.encoding),
                             timestamp: m.timestamp,
-                            qos: msg.ext_qos.into(),
+                            qos: QoS::from_or_default(msg.ext_qos),
                             source_id: m.ext_sinfo.as_ref().map(|i| i.zid),
                             source_sn: m.ext_sinfo.as_ref().map(|i| i.sn as u64),
                         };