Namespace prefix support

commons/zenoh-config/src/lib.rs

@@ -43,7 +43,10 @@ pub use zenoh_protocol::core::{
     whatami, EndPoint, Locator, WhatAmI, WhatAmIMatcher, WhatAmIMatcherVisitor,
 };
 use zenoh_protocol::{
-    core::{key_expr::OwnedKeyExpr, Bits},
+    core::{
+        key_expr::{OwnedKeyExpr, OwnedNonWildKeyExpr},
+        Bits,
+    },
     transport::{BatchSize, TransportSn},
 };
 use zenoh_result::{bail, zerror, ZResult};
@@ -632,6 +635,13 @@ validated_struct::validator! {
 
         },
 
+        /// Namespace prefix
+        /// If not None, all outgoing key expressions will be
+        /// automatically prefixed with specified string,
+        /// and all incoming key expressions will be stripped
+        /// of specified prefix
+        pub namespace: Option<OwnedNonWildKeyExpr>,
+
         /// Configuration of the downsampling.
         downsampling: Vec<DownsamplingItemConf>,
 

commons/zenoh-keyexpr/src/key_expr/borrowed.rs

@@ -27,7 +27,7 @@ use core::{
 
 use zenoh_result::{bail, Error as ZError, ZResult};
 
-use super::{canon::Canonize, OwnedKeyExpr, FORBIDDEN_CHARS};
+use super::{canon::Canonize, OwnedKeyExpr, OwnedNonWildKeyExpr, FORBIDDEN_CHARS};
 
 /// A [`str`] newtype that is statically known to be a valid key expression.
 ///
@@ -283,6 +283,87 @@ impl keyexpr {
         result
     }
 
+    /// Remove the specified namespace `prefix` from `self`.
+    ///
+    /// This method works essentially like [`keyexpr::strip_prefix()`], but returns only the longest possible suffix.
+    /// Prefix can not contain '*' character.
+    #[cfg(feature = "internal")]
+    #[doc(hidden)]
+    pub fn strip_nonwild_prefix(&self, prefix: &nonwild_keyexpr) -> Option<&keyexpr> {
+        fn is_chunk_matching(target: &[u8], prefix: &[u8]) -> bool {
+            let mut target_idx: usize = 0;
+            let mut prefix_idx: usize = 0;
+            let mut target_prev: u8 = b'/';
+
+            while target_idx < target.len() && prefix_idx < prefix.len() {
+                if target[target_idx] == b'*' {
+                    if target_prev == b'*' || target_idx + 1 == target.len() {
+                        // either a ** wild chunk or a single * chunk at the end of the string - this matches anything
+                        return true;
+                    } else if target_prev == b'$' {
+                        for i in prefix_idx..prefix.len() - 1 {
+                            if is_chunk_matching(&target[target_idx + 1..], &prefix[i..]) {
+                                return true;
+                            }
+                        }
+                    }
+                } else if target[target_idx] == prefix[prefix_idx] {
+                    prefix_idx += 1;
+                } else if target[target_idx] != b'$' {
+                    // non-special character, which do not match the one in prefix
+                    return false;
+                }
+                target_prev = target[target_idx];
+                target_idx += 1;
+            }
+            if prefix_idx != prefix.len() {
+                // prefix was not matched entirely
+                return false;
+            }
+            target_idx == target.len()
+                || (target_idx + 2 == target.len() && target[target_idx] == b'$')
+        }
+
+        let target_bytes = self.0.as_bytes();
+        let prefix_bytes = prefix.0.as_bytes();
+
+        let mut target_idx = 0;
+        let mut prefix_idx = 0;
+
+        while target_idx < target_bytes.len() && prefix_idx < prefix_bytes.len() {
+            let target_end = target_idx
+                + target_bytes[target_idx..]
+                    .iter()
+                    .position(|&i| i == b'/')
+                    .unwrap_or(target_bytes.len() - target_idx);
+            let prefix_end = prefix_idx
+                + prefix_bytes[prefix_idx..]
+                    .iter()
+                    .position(|&i| i == b'/')
+                    .unwrap_or(prefix_bytes.len() - prefix_idx);
+            let target_chunk = &target_bytes[target_idx..target_end];
+            let prefix_chunk = &prefix_bytes[prefix_idx..prefix_end];
+            if target_chunk.len() == 2 && target_chunk[0] == b'*' {
+                // Safety: every chunk of keyexpr is also a valid keyexpr
+                return unsafe { Some(keyexpr::from_str_unchecked(&self.0[target_idx..])) };
+            }
+            if target_end == target_bytes.len() {
+                // target contains no more chunks than prefix and the last one is non double-wild - so it can not match
+                return None;
+            }
+            if !is_chunk_matching(target_chunk, prefix_chunk) {
+                return None;
+            }
+            if prefix_end == prefix_bytes.len() {
+                // Safety: every chunk of keyexpr is also a valid keyexpr
+                return unsafe { Some(keyexpr::from_str_unchecked(&self.0[(target_end + 1)..])) };
+            }
+            target_idx = target_end + 1;
+            prefix_idx = prefix_end + 1;
+        }
+        None
+    }
+
     pub const fn as_str(&self) -> &str {
         &self.0
     }
@@ -759,6 +840,66 @@ impl ToOwned for keyexpr {
     }
 }
 
+/// A keyexpr that is statically known not to contain any wild chunks.
+#[allow(non_camel_case_types)]
+#[repr(transparent)]
+#[derive(PartialEq, Eq, Hash, PartialOrd, Ord, Debug)]
+pub struct nonwild_keyexpr(keyexpr);
+
+impl nonwild_keyexpr {
+    /// Attempts to construct a non-wild key expression from anything convertible to keyexpression.
+    ///
+    /// Will return an Err if `t` isn't a valid key expression.
+    pub fn new<'a, T, E>(t: &'a T) -> Result<&'a Self, ZError>
+    where
+        &'a keyexpr: TryFrom<&'a T, Error = E>,
+        E: Into<ZError>,
+        T: ?Sized,
+    {
+        let ke: &'a keyexpr = t.try_into().map_err(|e: E| e.into())?;
+        ke.try_into()
+    }
+
+    /// # Safety
+    /// This constructs a [`nonwild_keyexpr`] without ensuring that it is a valid key-expression without wild chunks.
+    ///
+    /// Much like [`core::str::from_utf8_unchecked`], this is memory-safe, but calling this without maintaining
+    /// [`nonwild_keyexpr`]'s invariants yourself may lead to unexpected behaviors, the Zenoh network dropping your messages.
+    pub const unsafe fn from_str_unchecked(s: &str) -> &Self {
+        core::mem::transmute(s)
+    }
+}
+
+impl Deref for nonwild_keyexpr {
+    type Target = keyexpr;
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl<'a> TryFrom<&'a keyexpr> for &'a nonwild_keyexpr {
+    type Error = ZError;
+    fn try_from(value: &'a keyexpr) -> Result<Self, Self::Error> {
+        if value.is_wild_impl() {
+            bail!("nonwild_keyexpr can not contain any wild chunks")
+        }
+        Ok(unsafe { core::mem::transmute::<&keyexpr, &nonwild_keyexpr>(value) })
+    }
+}
+
+impl Borrow<nonwild_keyexpr> for OwnedNonWildKeyExpr {
+    fn borrow(&self) -> &nonwild_keyexpr {
+        self
+    }
+}
+
+impl ToOwned for nonwild_keyexpr {
+    type Owned = OwnedNonWildKeyExpr;
+    fn to_owned(&self) -> Self::Owned {
+        OwnedNonWildKeyExpr::from(self)
+    }
+}
+
 #[test]
 fn test_keyexpr_strip_prefix() {
     let expectations = [
@@ -799,3 +940,46 @@ fn test_keyexpr_strip_prefix() {
         assert_eq!(ke.strip_prefix(prefix), expected)
     }
 }
+
+#[test]
+fn test_keyexpr_strip_nonwild_prefix() {
+    let expectations = [
+        (("demo/example/test/**", "demo/example/test"), Some("**")),
+        (("demo/example/**", "demo/example/test"), Some("**")),
+        (("**", "demo/example/test"), Some("**")),
+        (
+            ("demo/example/test/**/x$*/**", "demo/example/test"),
+            Some("**/x$*/**"),
+        ),
+        (("demo/**/xyz", "demo/example/test"), Some("**/xyz")),
+        (("demo/**/test/**", "demo/example/test"), Some("**/test/**")),
+        (
+            ("demo/**/ex$*/*/xyz", "demo/example/test"),
+            Some("**/ex$*/*/xyz"),
+        ),
+        (
+            ("demo/**/ex$*/t$*/xyz", "demo/example/test"),
+            Some("**/ex$*/t$*/xyz"),
+        ),
+        (
+            ("demo/**/te$*/*/xyz", "demo/example/test"),
+            Some("**/te$*/*/xyz"),
+        ),
+        (("demo/example/test", "demo/example/test"), None),
+        (("demo/example/test1/something", "demo/example/test"), None),
+        (
+            ("demo/example/test$*/something", "demo/example/test"),
+            Some("something"),
+        ),
+    ]
+    .map(|((a, b), expected)| {
+        (
+            (keyexpr::new(a).unwrap(), nonwild_keyexpr::new(b).unwrap()),
+            expected.map(|t| keyexpr::new(t).unwrap()),
+        )
+    });
+    for ((ke, prefix), expected) in expectations {
+        dbg!(ke, prefix);
+        assert_eq!(ke.strip_nonwild_prefix(prefix), expected)
+    }
+}

commons/zenoh-keyexpr/src/key_expr/mod.rs

@@ -21,7 +21,7 @@ pub(crate) const STAR_DSL: &[u8] = b"$*";
 pub(crate) const FORBIDDEN_CHARS: [u8; 3] = [b'#', b'?', b'$'];
 
 pub(crate) mod owned;
-pub use owned::OwnedKeyExpr;
+pub use owned::{OwnedKeyExpr, OwnedNonWildKeyExpr};
 
 pub(crate) mod borrowed;
 pub use borrowed::*;

commons/zenoh-keyexpr/src/key_expr/owned.rs

@@ -22,7 +22,7 @@ use core::{
     str::FromStr,
 };
 
-use super::{canon::Canonize, keyexpr};
+use super::{canon::Canonize, keyexpr, nonwild_keyexpr};
 
 /// A [`Arc<str>`] newtype that is statically known to be a valid key expression.
 ///
@@ -165,3 +165,58 @@ impl From<OwnedKeyExpr> for String {
         ke.as_str().to_owned()
     }
 }
+
+/// A [`Arc<str>`] newtype that is statically known to be a valid nonwild key expression.
+///
+/// See [`nonwild_keyexpr`](super::borrowed::nonwild_keyexpr).
+#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Deserialize)]
+#[cfg_attr(feature = "std", derive(schemars::JsonSchema))]
+#[serde(try_from = "String")]
+pub struct OwnedNonWildKeyExpr(pub(crate) Arc<str>);
+impl serde::Serialize for OwnedNonWildKeyExpr {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        self.0.serialize(serializer)
+    }
+}
+
+impl TryFrom<String> for OwnedNonWildKeyExpr {
+    type Error = zenoh_result::Error;
+    fn try_from(value: String) -> Result<Self, Self::Error> {
+        let ke = <&keyexpr as TryFrom<&str>>::try_from(value.as_str())?;
+        <&nonwild_keyexpr as TryFrom<&keyexpr>>::try_from(ke)?;
+        Ok(Self(value.into()))
+    }
+}
+impl<'a> From<&'a nonwild_keyexpr> for OwnedNonWildKeyExpr {
+    fn from(val: &'a nonwild_keyexpr) -> Self {
+        OwnedNonWildKeyExpr(Arc::from(val.as_str()))
+    }
+}
+
+impl Deref for OwnedNonWildKeyExpr {
+    type Target = nonwild_keyexpr;
+    fn deref(&self) -> &Self::Target {
+        unsafe { nonwild_keyexpr::from_str_unchecked(&self.0) }
+    }
+}
+
+#[allow(clippy::suspicious_arithmetic_impl)]
+impl Div<&keyexpr> for &OwnedNonWildKeyExpr {
+    type Output = OwnedKeyExpr;
+    fn div(self, rhs: &keyexpr) -> Self::Output {
+        let s: String = [self.as_str(), "/", rhs.as_str()].concat();
+        OwnedKeyExpr::autocanonize(s).unwrap() // Joining 2 key expressions should always result in a canonizable string.
+    }
+}
+
+#[allow(clippy::suspicious_arithmetic_impl)]
+impl Div<&nonwild_keyexpr> for &OwnedNonWildKeyExpr {
+    type Output = OwnedKeyExpr;
+    fn div(self, rhs: &nonwild_keyexpr) -> Self::Output {
+        let s: String = [self.as_str(), "/", rhs.as_str()].concat();
+        s.try_into().unwrap() // Joining 2 non wild key expressions should always result in a non wild string.
+    }
+}

commons/zenoh-macros/src/lib.rs

@@ -520,6 +520,17 @@ pub fn ke(tokens: TokenStream) -> TokenStream {
     }
 }
 
+/// Equivalent to [`nonwild_keyexpr::new`](zenoh_keyexpr::nonwild_keyexpr::new), but the check is run at compile-time and will throw a compile error in case of failure.
+#[proc_macro]
+pub fn nonwild_ke(tokens: TokenStream) -> TokenStream {
+    let value: LitStr = syn::parse(tokens).unwrap();
+    let ke = value.value();
+    match zenoh_keyexpr::nonwild_keyexpr::new(&ke) {
+        Ok(_) => quote!(unsafe { zenoh::key_expr::nonwild_keyexpr::from_str_unchecked(#ke)}).into(),
+        Err(e) => panic!("{}", e),
+    }
+}
+
 mod zenoh_runtime_derive;
 use syn::DeriveInput;
 use zenoh_runtime_derive::{derive_generic_runtime_param, derive_register_param};

zenoh-ext/src/advanced_cache.rs

@@ -28,13 +28,13 @@ use zenoh::{
     qos::{CongestionControl, Priority},
     query::{Queryable, ZenohParameters},
     sample::{Locality, Sample, SampleBuilder},
-    Resolvable, Result as ZResult, Session, Wait, KE_ADV_PREFIX, KE_AT, KE_STARSTAR,
+    Resolvable, Result as ZResult, Session, Wait, KE_ADV_PREFIX, KE_STARSTAR,
 };
 
 pub(crate) static KE_UHLC: &keyexpr = ke!("uhlc");
 #[zenoh_macros::unstable]
 kedefine!(
-    pub(crate) ke_liveliness: "@adv/${entity:*}/${zid:*}/${eid:*}/${meta:**}/@/${remaining:**}",
+    pub(crate) ke_liveliness: "${remaining:**}/@adv/${entity:*}/${zid:*}/${eid:*}/${meta:**}",
 );
 
 #[zenoh_macros::unstable]
@@ -122,7 +122,7 @@ impl CacheConfig {
 pub struct AdvancedCacheBuilder<'a, 'b, 'c> {
     session: &'a Session,
     pub_key_expr: ZResult<KeyExpr<'b>>,
-    queryable_prefix: Option<ZResult<KeyExpr<'c>>>,
+    queryable_suffix: Option<ZResult<KeyExpr<'c>>>,
     queryable_origin: Locality,
     history: CacheConfig,
     liveliness: bool,
@@ -138,21 +138,21 @@ impl<'a, 'b, 'c> AdvancedCacheBuilder<'a, 'b, 'c> {
         AdvancedCacheBuilder {
             session,
             pub_key_expr,
-            queryable_prefix: Some(Ok((KE_ADV_PREFIX / KE_STARSTAR / KE_AT).into())),
+            queryable_suffix: Some(Ok((KE_ADV_PREFIX / KE_STARSTAR).into())),
             queryable_origin: Locality::default(),
             history: CacheConfig::default(),
             liveliness: false,
         }
     }
 
-    /// Change the prefix used for queryable.
+    /// Change the suffix used for queryable.
     #[zenoh_macros::unstable]
-    pub fn queryable_prefix<TryIntoKeyExpr>(mut self, queryable_prefix: TryIntoKeyExpr) -> Self
+    pub fn queryable_suffix<TryIntoKeyExpr>(mut self, queryable_suffix: TryIntoKeyExpr) -> Self
     where
         TryIntoKeyExpr: TryInto<KeyExpr<'c>>,
         <TryIntoKeyExpr as TryInto<KeyExpr<'c>>>::Error: Into<zenoh::Error>,
     {
-        self.queryable_prefix = Some(queryable_prefix.try_into().map_err(Into::into));
+        self.queryable_suffix = Some(queryable_suffix.try_into().map_err(Into::into));
         self
     }
 
@@ -220,11 +220,11 @@ impl AdvancedCache {
     #[zenoh_macros::unstable]
     fn new(conf: AdvancedCacheBuilder<'_, '_, '_>) -> ZResult<AdvancedCache> {
         let key_expr = conf.pub_key_expr?.into_owned();
-        // the queryable_prefix (optional), and the key_expr for AdvancedCache's queryable ("[<queryable_prefix>]/<pub_key_expr>")
-        let queryable_key_expr = match conf.queryable_prefix {
+        // the queryable_suffix (optional), and the key_expr for AdvancedCache's queryable ("<pub_key_expr>/[<queryable_suffix>]")
+        let queryable_key_expr = match conf.queryable_suffix {
             None => key_expr.clone(),
-            Some(Ok(ke)) => (&ke) / &key_expr,
-            Some(Err(e)) => bail!("Invalid key expression for queryable_prefix: {}", e),
+            Some(Ok(ke)) => &key_expr / &ke,
+            Some(Err(e)) => bail!("Invalid key expression for queryable_suffix: {}", e),
         };
         tracing::debug!(
             "Create AdvancedCache on {} with max_samples={:?}",