Merge branch 'main' into tagging

eclipse-zenoh · Feb 6, 2024 · 2819698 · 2819698
2 parents efb4a45 + e8dca1e
commit 2819698
Show file tree

Hide file tree

Showing 45 changed files with 12,577 additions and 1,752 deletions.
diff --git a/examples/src/lib.rs b/examples/src/lib.rs
@@ -48,7 +48,6 @@ impl From<&CommonArgs> for Config {
             Some(path) => Config::from_file(path).unwrap(),
             None => Config::default(),
         };
-        println!("ARGS mode: {:?} ", value.mode);
         match value.mode {
             Some(Wai::Peer) => config.set_mode(Some(zenoh::scouting::WhatAmI::Peer)),
             Some(Wai::Client) => config.set_mode(Some(zenoh::scouting::WhatAmI::Client)),

diff --git a/io/zenoh-transport/src/lib.rs b/io/zenoh-transport/src/lib.rs
@@ -20,7 +20,6 @@
 pub mod common;
 pub mod manager;
 pub mod multicast;
-pub mod primitives;
 pub mod unicast;
 
 #[cfg(feature = "stats")]

diff --git a/io/zenoh-transport/src/primitives/demux.rs b/io/zenoh-transport/src/primitives/demux.rs
diff --git a/io/zenoh-transport/src/primitives/mux.rs b/io/zenoh-transport/src/primitives/mux.rs
diff --git a/zenoh-ext/examples/z_pub_cache.rs b/zenoh-ext/examples/z_pub_cache.rs
@@ -23,15 +23,16 @@ async fn main() {
     // Initiate logging
     env_logger::init();
 
-    let (config, key_expr, value, history, prefix) = parse_args();
+    let (config, key_expr, value, history, prefix, complete) = parse_args();
 
     println!("Opening session...");
     let session = zenoh::open(config).res().await.unwrap();
 
     println!("Declaring PublicationCache on {}", &key_expr);
     let mut publication_cache_builder = session
         .declare_publication_cache(&key_expr)
-        .history(history);
+        .history(history)
+        .queryable_complete(complete);
     if let Some(prefix) = prefix {
         publication_cache_builder = publication_cache_builder.queryable_prefix(prefix);
     }
@@ -45,7 +46,7 @@ async fn main() {
     }
 }
 
-fn parse_args() -> (Config, String, String, usize, Option<String>) {
+fn parse_args() -> (Config, String, String, usize, Option<String>, bool) {
     let args = Command::new("zenoh-ext pub cache example")
         .arg(
             arg!(-m --mode [MODE] "The zenoh session mode (peer by default)")
@@ -59,11 +60,12 @@ fn parse_args() -> (Config, String, String, usize, Option<String>) {
         )
         .arg(arg!(-v --value [VALUE]      "The value to publish.").default_value("Pub from Rust!"))
         .arg(
-            arg!(-h --history [SIZE] "The number of publications to keep in cache")
+            arg!(-i --history [SIZE] "The number of publications to keep in cache")
                 .default_value("1"),
         )
         .arg(arg!(-x --prefix [STRING] "An optional queryable prefix"))
         .arg(arg!(-c --config [FILE]      "A configuration file."))
+        .arg(arg!(-o --complete  "Set `complete` option to true. This means that this queryable is ulitmate data source, no need to scan other queryables."))
         .arg(arg!(--"no-multicast-scouting" "Disable the multicast-based scouting mechanism."))
         .get_matches();
 
@@ -101,6 +103,7 @@ fn parse_args() -> (Config, String, String, usize, Option<String>) {
     let value = args.get_one::<String>("value").unwrap().to_string();
     let history: usize = args.get_one::<String>("history").unwrap().parse().unwrap();
     let prefix = args.get_one::<String>("prefix").map(|s| (*s).to_owned());
+    let complete = args.get_flag("complete");
 
-    (config, key_expr, value, history, prefix)
+    (config, key_expr, value, history, prefix, complete)
 }
diff --git a/zenoh-ext/src/lib.rs b/zenoh-ext/src/lib.rs
@@ -20,7 +20,7 @@ pub use publication_cache::{PublicationCache, PublicationCacheBuilder};
 pub use querying_subscriber::{
     FetchingSubscriber, FetchingSubscriberBuilder, QueryingSubscriberBuilder,
 };
-pub use session_ext::{ArcSessionExt, SessionExt};
+pub use session_ext::SessionExt;
 pub use subscriber_ext::SubscriberBuilderExt;
 pub use subscriber_ext::SubscriberForward;
 

diff --git a/zenoh-ext/src/publication_cache.rs b/zenoh-ext/src/publication_cache.rs
@@ -32,7 +32,8 @@ pub struct PublicationCacheBuilder<'a, 'b, 'c> {
     session: SessionRef<'a>,
     pub_key_expr: ZResult<KeyExpr<'b>>,
     queryable_prefix: Option<ZResult<KeyExpr<'c>>>,
-    queryable_origin: Locality,
+    queryable_origin: Option<Locality>,
+    complete: Option<bool>,
     history: usize,
     resources_limit: Option<usize>,
 }
@@ -46,7 +47,8 @@ impl<'a, 'b, 'c> PublicationCacheBuilder<'a, 'b, 'c> {
             session,
             pub_key_expr,
             queryable_prefix: None,
-            queryable_origin: Locality::default(),
+            queryable_origin: None,
+            complete: None,
             history: 1,
             resources_limit: None,
         }
@@ -67,7 +69,13 @@ impl<'a, 'b, 'c> PublicationCacheBuilder<'a, 'b, 'c> {
     #[zenoh_macros::unstable]
     #[inline]
     pub fn queryable_allowed_origin(mut self, origin: Locality) -> Self {
-        self.queryable_origin = origin;
+        self.queryable_origin = Some(origin);
+        self
+    }
+
+    /// Set completeness option for the queryable.
+    pub fn queryable_complete(mut self, complete: bool) -> Self {
+        self.complete = Some(complete);
         self
     }
 
@@ -137,28 +145,21 @@ impl<'a> PublicationCache<'a> {
         }
 
         // declare the local subscriber that will store the local publications
-        let (local_sub, queryable) = match conf.session.clone() {
-            SessionRef::Borrow(session) => (
-                session
-                    .declare_subscriber(&key_expr)
-                    .allowed_origin(Locality::SessionLocal)
-                    .res_sync()?,
-                session
-                    .declare_queryable(&queryable_key_expr)
-                    .allowed_origin(conf.queryable_origin)
-                    .res_sync()?,
-            ),
-            SessionRef::Shared(session) => (
-                session
-                    .declare_subscriber(&key_expr)
-                    .allowed_origin(Locality::SessionLocal)
-                    .res_sync()?,
-                session
-                    .declare_queryable(&queryable_key_expr)
-                    .allowed_origin(conf.queryable_origin)
-                    .res_sync()?,
-            ),
-        };
+        let local_sub = conf
+            .session
+            .declare_subscriber(&key_expr)
+            .allowed_origin(Locality::SessionLocal)
+            .res_sync()?;
+
+        // declare the queryable which returns the cached publications
+        let mut queryable = conf.session.declare_queryable(&queryable_key_expr);
+        if let Some(origin) = conf.queryable_origin {
+            queryable = queryable.allowed_origin(origin);
+        }
+        if let Some(complete) = conf.complete {
+            queryable = queryable.complete(complete);
+        }
+        let queryable = queryable.res_sync()?;
 
         // take local ownership of stuff to be moved into task
         let sub_recv = local_sub.receiver.clone();

diff --git a/zenoh-ext/src/querying_subscriber.rs b/zenoh-ext/src/querying_subscriber.rs
@@ -680,33 +680,20 @@ impl<'a, Receiver> FetchingSubscriber<'a, Receiver> {
         // register fetch handler
         let handler = register_handler(state.clone(), callback.clone());
         // declare subscriber
-        let subscriber = match conf.session.clone() {
-            SessionRef::Borrow(session) => match conf.key_space.into() {
-                crate::KeySpace::User => session
-                    .declare_subscriber(&key_expr)
-                    .callback(sub_callback)
-                    .reliability(conf.reliability)
-                    .allowed_origin(conf.origin)
-                    .res_sync()?,
-                crate::KeySpace::Liveliness => session
-                    .liveliness()
-                    .declare_subscriber(&key_expr)
-                    .callback(sub_callback)
-                    .res_sync()?,
-            },
-            SessionRef::Shared(session) => match conf.key_space.into() {
-                crate::KeySpace::User => session
-                    .declare_subscriber(&key_expr)
-                    .callback(sub_callback)
-                    .reliability(conf.reliability)
-                    .allowed_origin(conf.origin)
-                    .res_sync()?,
-                crate::KeySpace::Liveliness => session
-                    .liveliness()
-                    .declare_subscriber(&key_expr)
-                    .callback(sub_callback)
-                    .res_sync()?,
-            },
+        let subscriber = match conf.key_space.into() {
+            crate::KeySpace::User => conf
+                .session
+                .declare_subscriber(&key_expr)
+                .callback(sub_callback)
+                .reliability(conf.reliability)
+                .allowed_origin(conf.origin)
+                .res_sync()?,
+            crate::KeySpace::Liveliness => conf
+                .session
+                .liveliness()
+                .declare_subscriber(&key_expr)
+                .callback(sub_callback)
+                .res_sync()?,
         };
 
         let fetch_subscriber = FetchingSubscriber {