Add express support to publisher and put

zenoh/src/publication.rs

@@ -102,6 +102,13 @@ impl PutBuilder<'_, '_> {
         self
     }
 
+    /// Change the `congestion_control` to apply when routing the data.
+    #[inline]
+    pub fn express(mut self, is_express: bool) -> Self {
+        self.publisher = self.publisher.express(is_express);
+        self
+    }
+
     /// Restrict the matching subscribers that will receive the published data
     /// to the ones that have the given [`Locality`](crate::prelude::Locality).
     #[zenoh_macros::unstable]
@@ -141,6 +148,7 @@ impl SyncResolve for PutBuilder<'_, '_> {
             key_expr,
             congestion_control,
             priority,
+            is_express,
             destination,
         } = self.publisher;
 
@@ -151,6 +159,7 @@ impl SyncResolve for PutBuilder<'_, '_> {
             key_expr: key_expr?,
             congestion_control,
             priority,
+            is_express,
             destination,
         };
 
@@ -248,6 +257,7 @@ pub struct Publisher<'a> {
     pub(crate) key_expr: KeyExpr<'a>,
     pub(crate) congestion_control: CongestionControl,
     pub(crate) priority: Priority,
+    pub(crate) is_express: bool,
     pub(crate) destination: Locality,
 }
 
@@ -738,6 +748,7 @@ pub struct PublisherBuilder<'a, 'b: 'a> {
     pub(crate) key_expr: ZResult<KeyExpr<'b>>,
     pub(crate) congestion_control: CongestionControl,
     pub(crate) priority: Priority,
+    pub(crate) is_express: bool,
     pub(crate) destination: Locality,
 }
 
@@ -751,6 +762,7 @@ impl<'a, 'b> Clone for PublisherBuilder<'a, 'b> {
             },
             congestion_control: self.congestion_control,
             priority: self.priority,
+            is_express: self.is_express,
             destination: self.destination,
         }
     }
@@ -771,6 +783,13 @@ impl<'a, 'b> PublisherBuilder<'a, 'b> {
         self
     }
 
+    /// Change the `congestion_control` to apply when routing the data.
+    #[inline]
+    pub fn express(mut self, is_express: bool) -> Self {
+        self.is_express = is_express;
+        self
+    }
+
     /// Restrict the matching subscribers that will receive the published data
     /// to the ones that have the given [`Locality`](crate::prelude::Locality).
     #[zenoh_macros::unstable]
@@ -830,6 +849,7 @@ impl<'a, 'b> SyncResolve for PublisherBuilder<'a, 'b> {
             key_expr,
             congestion_control: self.congestion_control,
             priority: self.priority,
+            is_express: self.is_express,
             destination: self.destination,
         };
         log::trace!("publish({:?})", publisher.key_expr);
@@ -867,7 +887,7 @@ fn resolve_put(
             ext_qos: ext::QoSType::new(
                 publisher.priority.into(),
                 publisher.congestion_control,
-                false,
+                publisher.is_express,
             ),
             ext_tstamp: None,
             ext_nodeid: ext::NodeIdType::DEFAULT,
@@ -933,7 +953,7 @@ fn resolve_put(
             qos: QoS::from(ext::QoSType::new(
                 publisher.priority.into(),
                 publisher.congestion_control,
-                false,
+                publisher.is_express,
             )),
         };
 

zenoh/src/session.rs

@@ -337,6 +337,7 @@ impl<'s, 'a> SessionDeclarations<'s, 'a> for SessionRef<'a> {
             key_expr: key_expr.try_into().map_err(Into::into),
             congestion_control: CongestionControl::DEFAULT,
             priority: Priority::DEFAULT,
+            is_express: false,
             destination: Locality::default(),
         }
     }
@@ -1909,6 +1910,7 @@ impl<'s> SessionDeclarations<'s, 'static> for Arc<Session> {
             key_expr: key_expr.try_into().map_err(Into::into),
             congestion_control: CongestionControl::DEFAULT,
             priority: Priority::DEFAULT,
+            is_express: false,
             destination: Locality::default(),
         }
     }