Add QoS configuration overwrites for publication messages

DEFAULT_CONFIG.json5

@@ -184,6 +184,22 @@
     },
   },
 
+  //  /// Overwrite default builders for specific key expressions
+  //  builders: {
+  //      /// A list of publisher builder configurations for specific key expressions.
+  //      publishers: [
+  //        // key_expressions
+  //        key_exprs: ["demo/**", "example/key"],
+  //        // builder configuration
+  //        congestion_control: "Block",
+  //        encoding: "zenoh/example",
+  //        priority: "data_high",
+  //        express: true,
+  //        reliability: "best_effort",
+  //        allowed_destination: "remote",
+  //      ],
+  //  },
+
   //  /// The declarations aggregation strategy.
   //  aggregation: {
   //      /// A list of key-expressions for which all included subscribers will be aggregated into.

commons/zenoh-config/Cargo.toml

@@ -25,6 +25,7 @@ description = "Internal crate for zenoh."
 
 [features]
 internal = []
+unstable = []
 
 [dependencies]
 tracing = { workspace = true }
@@ -35,6 +36,7 @@ serde_json = { workspace = true }
 serde_yaml = { workspace = true }
 validated_struct = { workspace = true, features = ["json5", "json_get"] }
 zenoh-core = { workspace = true }
+zenoh-keyexpr = { workspace = true }
 zenoh-protocol = { workspace = true }
 zenoh-result = { workspace = true }
 zenoh-util = { workspace = true }

commons/zenoh-config/src/builders.rs

@@ -0,0 +1,127 @@
+use std::collections::HashSet;
+
+use serde::{Deserialize, Deserializer, Serialize};
+pub use validated_struct::{GetError, ValidatedMap};
+use zenoh_keyexpr::keyexpr_tree::{IKeyExprTreeMut, KeBoxTree};
+use zenoh_protocol::core::{key_expr::OwnedKeyExpr, CongestionControl, Reliability};
+pub use zenoh_protocol::core::{
+    whatami, EndPoint, Locator, WhatAmI, WhatAmIMatcher, WhatAmIMatcherVisitor,
+};
+
+#[derive(Debug, Deserialize, Default, Serialize, Clone)]
+#[serde(remote = "Self")]
+pub struct PublisherBuildersConf(pub(crate) Vec<PublisherBuildersInnerConf>);
+
+impl<'de> Deserialize<'de> for PublisherBuildersConf {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let builders = PublisherBuildersConf::deserialize(deserializer)?;
+        // check for invariant: each key_expr should be unique
+        let mut key_set = HashSet::new();
+        for builder in &builders.0 {
+            for key_expr in &builder.key_exprs {
+                if !key_set.insert(key_expr) {
+                    return Err(format!(
+                        "duplicated key_expr '{key_expr}' found in publisher builders config"
+                    ))
+                    .map_err(serde::de::Error::custom);
+                }
+            }
+        }
+        Ok(builders)
+    }
+}
+
+impl Serialize for PublisherBuildersConf {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        PublisherBuildersConf::serialize(self, serializer)
+    }
+}
+
+impl From<PublisherBuildersConf> for KeBoxTree<PublisherBuilderOptionsConf> {
+    fn from(value: PublisherBuildersConf) -> KeBoxTree<PublisherBuilderOptionsConf> {
+        let mut tree = KeBoxTree::new();
+        for conf in value.0 {
+            for key_expr in conf.key_exprs {
+                // key_expr unicity is checked at deserialization
+                tree.insert(&key_expr, conf.builder_conf.clone());
+            }
+        }
+        tree
+    }
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub(crate) struct PublisherBuildersInnerConf {
+    pub key_exprs: Vec<OwnedKeyExpr>,
+    #[serde(flatten)]
+    pub builder_conf: PublisherBuilderOptionsConf,
+}
+
+#[derive(Debug, Default, Deserialize, Serialize, Clone)]
+pub struct PublisherBuilderOptionsConf {
+    pub congestion_control: Option<PublisherCongestionControlConf>,
+    pub encoding: Option<String>, // Encoding has From<&str>
+    pub priority: Option<PublisherPriorityConf>,
+    pub express: Option<bool>,
+    #[cfg(feature = "unstable")]
+    pub reliability: Option<PublisherReliabilityConf>,
+    pub allowed_destination: Option<PublisherLocalityConf>,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone, Copy)]
+#[serde(rename_all = "lowercase")]
+pub enum PublisherCongestionControlConf {
+    Drop,
+    Block,
+}
+
+impl From<PublisherCongestionControlConf> for CongestionControl {
+    fn from(value: PublisherCongestionControlConf) -> Self {
+        match value {
+            PublisherCongestionControlConf::Drop => CongestionControl::Drop,
+            PublisherCongestionControlConf::Block => CongestionControl::Block,
+        }
+    }
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone, Copy)]
+#[serde(rename_all = "snake_case")]
+pub enum PublisherPriorityConf {
+    RealTime = 1,
+    InteractiveHigh = 2,
+    InteractiveLow = 3,
+    DataHigh = 4,
+    Data = 5,
+    DataLow = 6,
+    Background = 7,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone, Copy)]
+#[serde(rename_all = "snake_case")]
+pub enum PublisherReliabilityConf {
+    BestEffort,
+    Reliable,
+}
+
+impl From<PublisherReliabilityConf> for Reliability {
+    fn from(value: PublisherReliabilityConf) -> Reliability {
+        match value {
+            PublisherReliabilityConf::BestEffort => Reliability::BestEffort,
+            PublisherReliabilityConf::Reliable => Reliability::Reliable,
+        }
+    }
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone, Copy)]
+#[serde(rename_all = "snake_case")]
+pub enum PublisherLocalityConf {
+    SessionLocal,
+    Remote,
+    Any,
+}

commons/zenoh-config/src/lib.rs

@@ -19,6 +19,7 @@
 //! [Click here for Zenoh's documentation](https://docs.rs/zenoh/latest/zenoh)
 //!
 //! Configuration to pass to `zenoh::open()` and `zenoh::scout()` functions and associated constants.
+pub mod builders;
 pub mod defaults;
 mod include;
 pub mod wrappers;
@@ -29,6 +30,7 @@ use std::{
     any::Any, collections::HashSet, fmt, io::Read, net::SocketAddr, ops, path::Path, sync::Weak,
 };
 
+use builders::PublisherBuildersConf;
 use include::recursive_include;
 use secrecy::{CloneableSecret, DebugSecret, Secret, SerializableSecret, Zeroize};
 use serde::{Deserialize, Serialize};
@@ -360,6 +362,14 @@ validated_struct::validator! {
             /// A list of key-expressions for which all included publishers will be aggregated into.
             publishers: Vec<OwnedKeyExpr>,
         },
+
+        /// Overwrite default builders for specific key expressions
+        pub builders: #[derive(Default)]
+        BuildersConf {
+            /// A list of publisher builder configurations for specific key expressions.
+            publishers: PublisherBuildersConf,
+        },
+
         pub transport: #[derive(Default)]
         TransportConf {
             pub unicast: TransportUnicastConf {

zenoh/Cargo.toml

@@ -64,7 +64,7 @@ transport_udp = ["zenoh-transport/transport_udp"]
 transport_unixsock-stream = ["zenoh-transport/transport_unixsock-stream"]
 transport_ws = ["zenoh-transport/transport_ws"]
 transport_vsock = ["zenoh-transport/transport_vsock"]
-unstable = ["internal_config", "zenoh-keyexpr/unstable"]
+unstable = ["internal_config", "zenoh-keyexpr/unstable", "zenoh-config/unstable"]
 internal_config = []
 
 [dependencies]

zenoh/src/api/builders/session.rs

@@ -16,6 +16,8 @@ use std::future::{IntoFuture, Ready};
 #[cfg(feature = "shared-memory")]
 use std::sync::Arc;
 
+#[cfg(feature = "internal")]
+use zenoh_config::builders::PublisherBuildersConf;
 use zenoh_core::{Resolvable, Wait};
 #[cfg(feature = "internal")]
 use zenoh_keyexpr::OwnedKeyExpr;
@@ -122,6 +124,7 @@ pub fn init(runtime: Runtime) -> InitBuilder {
         runtime,
         aggregated_subscribers: vec![],
         aggregated_publishers: vec![],
+        publisher_builders: PublisherBuildersConf::default(),
     }
 }
 
@@ -133,6 +136,7 @@ pub struct InitBuilder {
     runtime: Runtime,
     aggregated_subscribers: Vec<OwnedKeyExpr>,
     aggregated_publishers: Vec<OwnedKeyExpr>,
+    publisher_builders: PublisherBuildersConf,
 }
 
 #[zenoh_macros::internal]
@@ -148,6 +152,11 @@ impl InitBuilder {
         self.aggregated_publishers = exprs;
         self
     }
+
+    pub fn publisher_builders(mut self, builder_conf: PublisherBuildersConf) -> Self {
+        self.publisher_builders = builder_conf;
+        self
+    }
 }
 
 #[zenoh_macros::internal]
@@ -162,6 +171,7 @@ impl Wait for InitBuilder {
             self.runtime,
             self.aggregated_subscribers,
             self.aggregated_publishers,
+            self.publisher_builders,
             false,
         )
         .wait())

zenoh/src/api/publisher.rs

@@ -22,6 +22,7 @@ use std::{
 
 use futures::Sink;
 use tracing::error;
+use zenoh_config::builders::PublisherPriorityConf;
 use zenoh_core::{Resolvable, Resolve, Wait};
 use zenoh_protocol::core::CongestionControl;
 use zenoh_result::{Error, ZResult};
@@ -469,6 +470,20 @@ impl TryFrom<u8> for Priority {
     }
 }
 
+impl From<PublisherPriorityConf> for Priority {
+    fn from(value: PublisherPriorityConf) -> Self {
+        match value {
+            PublisherPriorityConf::RealTime => Priority::RealTime,
+            PublisherPriorityConf::InteractiveHigh => Priority::InteractiveHigh,
+            PublisherPriorityConf::InteractiveLow => Priority::InteractiveLow,
+            PublisherPriorityConf::DataHigh => Priority::DataHigh,
+            PublisherPriorityConf::Data => Priority::Data,
+            PublisherPriorityConf::DataLow => Priority::DataLow,
+            PublisherPriorityConf::Background => Priority::Background,
+        }
+    }
+}
+
 type ProtocolPriority = zenoh_protocol::core::Priority;
 impl From<Priority> for ProtocolPriority {
     fn from(prio: Priority) -> Self {

zenoh/src/api/sample.rs

@@ -16,7 +16,7 @@
 use std::{convert::TryFrom, fmt};
 
 use serde::{Deserialize, Serialize};
-use zenoh_config::wrappers::EntityGlobalId;
+use zenoh_config::{builders::PublisherLocalityConf, wrappers::EntityGlobalId};
 #[cfg(feature = "unstable")]
 use zenoh_protocol::core::Reliability;
 use zenoh_protocol::{
@@ -50,6 +50,16 @@ pub(crate) enum Locality {
     Any,
 }
 
+impl From<PublisherLocalityConf> for Locality {
+    fn from(value: PublisherLocalityConf) -> Self {
+        match value {
+            PublisherLocalityConf::SessionLocal => Locality::SessionLocal,
+            PublisherLocalityConf::Remote => Locality::Remote,
+            PublisherLocalityConf::Any => Locality::Any,
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Default)]
 pub(crate) struct DataInfo {
     pub kind: SampleKind,