diff --git a/commons/zenoh-protocol/src/core/endpoint.rs b/commons/zenoh-protocol/src/core/endpoint.rs index 96b9b40665..237842ea6e 100644 --- a/commons/zenoh-protocol/src/core/endpoint.rs +++ b/commons/zenoh-protocol/src/core/endpoint.rs @@ -187,6 +187,9 @@ impl fmt::Debug for AddressMut<'_> { pub struct Metadata<'a>(pub(super) &'a str); impl<'a> Metadata<'a> { + pub const RELIABILITY: &'static str = "reliability"; + pub const PRIORITIES: &'static str = "priorities"; + pub fn as_str(&self) -> &'a str { self.0 } diff --git a/commons/zenoh-protocol/src/core/mod.rs b/commons/zenoh-protocol/src/core/mod.rs index 57844d5a4b..73de2e5a8c 100644 --- a/commons/zenoh-protocol/src/core/mod.rs +++ b/commons/zenoh-protocol/src/core/mod.rs @@ -351,6 +351,12 @@ impl PriorityRange { } } +impl ToString for PriorityRange { + fn to_string(&self) -> String { + format!("{}..={}", *self.start() as u8, *self.end() as u8) + } +} + #[derive(Debug)] pub enum InvalidPriorityRange { InvalidSyntax { found: String }, @@ -458,6 +464,16 @@ pub enum Reliability { impl Reliability { pub const DEFAULT: Self = Self::BestEffort; + const BEST_EFFORT_STR: &str = "best_effort"; + const RELIABLE_STR: &str = "reliable"; + + pub fn as_str(&self) -> &str { + match self { + Reliability::BestEffort => Reliability::BEST_EFFORT_STR, + Reliability::Reliable => Reliability::RELIABLE_STR, + } + } + #[cfg(feature = "test")] pub fn rand() -> Self { use rand::Rng; @@ -500,7 +516,9 @@ impl Display for InvalidReliability { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!( f, - "invalid Reliability string, expected `best_effort` or `reliable` but found {}", + "invalid Reliability string, expected `{}` or `{}` but found {}", + Reliability::BEST_EFFORT_STR, + Reliability::RELIABLE_STR, self.found ) } @@ -514,8 +532,8 @@ impl FromStr for Reliability { fn from_str(s: &str) -> Result { match s { - "reliable" => Ok(Reliability::Reliable), - "best_effort" => Ok(Reliability::BestEffort), + Reliability::RELIABLE_STR => Ok(Reliability::Reliable), + Reliability::BEST_EFFORT_STR => Ok(Reliability::BestEffort), other => Err(InvalidReliability { found: other.to_string(), }), diff --git a/io/zenoh-link-commons/src/lib.rs b/io/zenoh-link-commons/src/lib.rs index 12b76385b8..224360e327 100644 --- a/io/zenoh-link-commons/src/lib.rs +++ b/io/zenoh-link-commons/src/lib.rs @@ -34,7 +34,7 @@ pub use multicast::*; use serde::Serialize; pub use unicast::*; use zenoh_protocol::{ - core::{Locator, PriorityRange, Reliability}, + core::{Locator, Metadata, PriorityRange, Reliability}, transport::BatchSize, }; use zenoh_result::ZResult; @@ -45,7 +45,6 @@ use zenoh_result::ZResult; pub const BIND_INTERFACE: &str = "iface"; -// TODO(fuzzypixelz): Patch the Locators to contain negotiated priority and reliability #[derive(Clone, Debug, Serialize, Hash, PartialEq, Eq)] pub struct Link { pub src: Locator, @@ -82,8 +81,8 @@ impl Link { reliability: Reliability, ) -> Self { Link { - src: link.get_src().to_owned(), - dst: link.get_dst().to_owned(), + src: Self::to_patched_locator(link.get_src(), priorities.as_ref(), reliability), + dst: Self::to_patched_locator(link.get_dst(), priorities.as_ref(), reliability), group: None, mtu: link.get_mtu(), is_streamed: link.is_streamed(), @@ -107,6 +106,24 @@ impl Link { reliability: Reliability::from(link.is_reliable()), } } + + /// Updates the metdata of the `locator` with `priorities` and `reliability`. + fn to_patched_locator( + locator: &Locator, + priorities: Option<&PriorityRange>, + reliability: Reliability, + ) -> Locator { + let mut locator = locator.clone(); + let mut metadata = locator.metadata_mut(); + metadata + .insert(Metadata::RELIABILITY, reliability.as_str()) + .expect("adding `reliability` to Locator metadata should not fail"); + priorities + .map(|ps| metadata.insert(Metadata::PRIORITIES, ps.to_string())) + .transpose() + .expect("adding `priorities` to Locator metadata should not fail"); + locator + } } impl PartialEq for Link { diff --git a/io/zenoh-transport/src/unicast/establishment/ext/qos.rs b/io/zenoh-transport/src/unicast/establishment/ext/qos.rs index 6f3c440886..d094b7607b 100644 --- a/io/zenoh-transport/src/unicast/establishment/ext/qos.rs +++ b/io/zenoh-transport/src/unicast/establishment/ext/qos.rs @@ -23,7 +23,7 @@ use zenoh_codec::{RCodec, WCodec, Zenoh080}; use zenoh_core::zerror; use zenoh_link::EndPoint; use zenoh_protocol::{ - core::{Priority, PriorityRange, Reliability}, + core::{Metadata, Priority, PriorityRange, Reliability}, transport::{init, open}, }; use zenoh_result::{Error as ZError, ZResult}; @@ -56,18 +56,15 @@ impl State { if !is_qos { Ok(State::NoQoS) } else { - const RELIABILITY_METADATA_KEY: &str = "reliability"; - const PRIORITY_METADATA_KEY: &str = "priorities"; - let metadata = endpoint.metadata(); let reliability = metadata - .get(RELIABILITY_METADATA_KEY) + .get(Metadata::RELIABILITY) .map(Reliability::from_str) .transpose()?; let priorities = metadata - .get(PRIORITY_METADATA_KEY) + .get(Metadata::PRIORITIES) .map(PriorityRange::from_str) .transpose()?;