Merge branch 'main' into bug/liveliness_duplicate

ci/valgrind-check/src/pub_sub/bin/z_pub_sub.rs

@@ -13,7 +13,7 @@
 //
 use std::time::Duration;
 
-use zenoh::{config::Config, key_expr::KeyExpr, prelude::*};
+use zenoh::{config::Config, key_expr::KeyExpr};
 
 #[tokio::main]
 async fn main() {

ci/valgrind-check/src/queryable_get/bin/z_queryable_get.rs

@@ -16,7 +16,6 @@ use std::{convert::TryFrom, time::Duration};
 use zenoh::{
     config::Config,
     key_expr::KeyExpr,
-    prelude::*,
     query::{QueryTarget, Selector},
 };
 

commons/zenoh-config/src/lib.rs

@@ -224,9 +224,8 @@ pub fn client<I: IntoIterator<Item = T>, T: Into<EndPoint>>(peers: I) -> Config
 
 #[test]
 fn config_keys() {
-    use validated_struct::ValidatedMap;
     let c = Config::default();
-    dbg!(c.keys());
+    dbg!(Vec::from_iter(c.keys()));
 }
 
 validated_struct::validator! {
@@ -659,6 +658,40 @@ fn config_deser() {
 }
 
 impl Config {
+    pub fn insert<'d, D: serde::Deserializer<'d>>(
+        &mut self,
+        key: &str,
+        value: D,
+    ) -> Result<(), validated_struct::InsertionError>
+    where
+        validated_struct::InsertionError: From<D::Error>,
+    {
+        <Self as ValidatedMap>::insert(self, key, value)
+    }
+
+    pub fn get(
+        &self,
+        key: &str,
+    ) -> Result<<Self as ValidatedMapAssociatedTypes>::Accessor, GetError> {
+        <Self as ValidatedMap>::get(self, key)
+    }
+
+    pub fn get_json(&self, key: &str) -> Result<String, GetError> {
+        <Self as ValidatedMap>::get_json(self, key)
+    }
+
+    pub fn insert_json5(
+        &mut self,
+        key: &str,
+        value: &str,
+    ) -> Result<(), validated_struct::InsertionError> {
+        <Self as ValidatedMap>::insert_json5(self, key, value)
+    }
+
+    pub fn keys(&self) -> impl Iterator<Item = String> {
+        <Self as ValidatedMap>::keys(self).into_iter()
+    }
+
     pub fn set_plugin_validator<T: ConfigValidator + 'static>(&mut self, validator: Weak<T>) {
         self.plugins.validator = validator;
     }
@@ -778,7 +811,6 @@ impl std::fmt::Display for Config {
 
 #[test]
 fn config_from_json() {
-    use validated_struct::ValidatedMap;
     let from_str = serde_json::Deserializer::from_str;
     let mut config = Config::from_deserializer(&mut from_str(r#"{}"#)).unwrap();
     config
@@ -944,6 +976,52 @@ where
         self.lock().keys()
     }
 }
+impl<T: ValidatedMap + 'static> Notifier<T>
+where
+    T: for<'a> ValidatedMapAssociatedTypes<'a, Accessor = &'a dyn Any>,
+{
+    pub fn insert<'d, D: serde::Deserializer<'d>>(
+        &self,
+        key: &str,
+        value: D,
+    ) -> Result<(), validated_struct::InsertionError>
+    where
+        validated_struct::InsertionError: From<D::Error>,
+    {
+        self.lock().insert(key, value)?;
+        self.notify(key);
+        Ok(())
+    }
+
+    pub fn get(
+        &self,
+        key: &str,
+    ) -> Result<<Self as ValidatedMapAssociatedTypes>::Accessor, GetError> {
+        let guard = zlock!(self.inner.inner);
+        // SAFETY: MutexGuard pins the mutex behind which the value is held.
+        let subref = guard.get(key.as_ref())? as *const _;
+        Ok(GetGuard {
+            _guard: guard,
+            subref,
+        })
+    }
+
+    pub fn get_json(&self, key: &str) -> Result<String, GetError> {
+        self.lock().get_json(key)
+    }
+
+    pub fn insert_json5(
+        &self,
+        key: &str,
+        value: &str,
+    ) -> Result<(), validated_struct::InsertionError> {
+        self.insert(key, &mut json5::Deserializer::from_str(value)?)
+    }
+
+    pub fn keys(&self) -> impl Iterator<Item = String> {
+        self.lock().keys().into_iter()
+    }
+}
 
 pub struct GetGuard<'a, T> {
     _guard: MutexGuard<'a, T>,

examples/examples/z_alloc_shm.rs

@@ -12,7 +12,6 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use zenoh::{
-    prelude::*,
     shm::{
         AllocAlignment, BlockOn, Deallocate, Defragment, GarbageCollect, PosixShmProviderBackend,
         ShmProviderBuilder, POSIX_PROTOCOL_ID,
@@ -27,7 +26,7 @@ async fn main() {
     run().await.unwrap()
 }
 
-async fn run() -> ZResult<()> {
+async fn run() -> zenoh::Result<()> {
     // create an SHM backend...
     // NOTE: For extended PosixShmProviderBackend API please check z_posix_shm_provider.rs
     let backend = PosixShmProviderBackend::builder()

examples/examples/z_bytes_shm.rs

@@ -13,11 +13,11 @@
 //
 use zenoh::{
     bytes::ZBytes,
-    prelude::*,
     shm::{
         zshm, zshmmut, PosixShmProviderBackend, ShmProviderBuilder, ZShm, ZShmMut,
         POSIX_PROTOCOL_ID,
     },
+    Wait,
 };
 
 fn main() {

examples/examples/z_forward.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 use zenoh_ext::SubscriberForward;
 

examples/examples/z_get_liveliness.rs

@@ -14,7 +14,7 @@
 use std::time::Duration;
 
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_info.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{prelude::*, session::ZenohId};
+use zenoh::session::ZenohId;
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_liveliness.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_ping.rs

@@ -14,7 +14,7 @@
 use std::time::{Duration, Instant};
 
 use clap::Parser;
-use zenoh::{bytes::ZBytes, key_expr::keyexpr, prelude::*, qos::CongestionControl, Config};
+use zenoh::{bytes::ZBytes, key_expr::keyexpr, qos::CongestionControl, Config, Wait};
 use zenoh_examples::CommonArgs;
 
 fn main() {

examples/examples/z_ping_shm.rs

@@ -17,7 +17,6 @@ use clap::Parser;
 use zenoh::{
     bytes::ZBytes,
     key_expr::keyexpr,
-    prelude::*,
     qos::CongestionControl,
     shm::{PosixShmProviderBackend, ShmProviderBuilder, POSIX_PROTOCOL_ID},
     Config, Wait,

examples/examples/z_pong.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::keyexpr, prelude::*, qos::CongestionControl, Config};
+use zenoh::{key_expr::keyexpr, qos::CongestionControl, Config, Wait};
 use zenoh_examples::CommonArgs;
 
 fn main() {
@@ -21,7 +21,7 @@ fn main() {
 
     let (config, express) = parse_args();
 
-    let session = zenoh::open(config).wait().unwrap().into_arc();
+    let session = zenoh::open(config).wait().unwrap();
 
     // The key expression to read the data from
     let key_expr_ping = keyexpr::new("test/ping").unwrap();

examples/examples/z_pub.rs

@@ -14,7 +14,7 @@
 use std::time::Duration;
 
 use clap::Parser;
-use zenoh::{bytes::Encoding, key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{bytes::Encoding, key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_pub_shm.rs

@@ -14,7 +14,6 @@
 use clap::Parser;
 use zenoh::{
     key_expr::KeyExpr,
-    prelude::*,
     shm::{
         BlockOn, GarbageCollect, PosixShmProviderBackend, ShmProviderBuilder, POSIX_PROTOCOL_ID,
     },
@@ -25,7 +24,7 @@ use zenoh_examples::CommonArgs;
 const N: usize = 10;
 
 #[tokio::main]
-async fn main() -> Result<(), ZError> {
+async fn main() -> zenoh::Result<()> {
     // Initiate logging
     zenoh::init_log_from_env_or("error");
 

examples/examples/z_pub_shm_thr.rs

@@ -14,7 +14,6 @@
 use clap::Parser;
 use zenoh::{
     bytes::ZBytes,
-    prelude::*,
     qos::CongestionControl,
     shm::{PosixShmProviderBackend, ShmProviderBuilder, POSIX_PROTOCOL_ID},
     Config, Wait,

examples/examples/z_pub_thr.rs

@@ -17,8 +17,8 @@ use std::convert::TryInto;
 use clap::Parser;
 use zenoh::{
     bytes::ZBytes,
-    prelude::*,
     qos::{CongestionControl, Priority},
+    Wait,
 };
 use zenoh_examples::CommonArgs;
 

examples/examples/z_pull.rs

@@ -14,7 +14,7 @@
 use std::time::Duration;
 
 use clap::Parser;
-use zenoh::{handlers::RingChannel, key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{handlers::RingChannel, key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_queryable.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_queryable_shm.rs

@@ -15,7 +15,6 @@ use clap::Parser;
 use zenoh::{
     bytes::ZBytes,
     key_expr::KeyExpr,
-    prelude::*,
     shm::{
         zshm, BlockOn, GarbageCollect, PosixShmProviderBackend, ShmProviderBuilder,
         POSIX_PROTOCOL_ID,

examples/examples/z_storage.rs

@@ -19,7 +19,6 @@ use clap::Parser;
 use futures::select;
 use zenoh::{
     key_expr::{keyexpr, KeyExpr},
-    prelude::*,
     sample::{Sample, SampleKind},
     Config,
 };

examples/examples/z_sub.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_sub_liveliness.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, sample::SampleKind, Config};
+use zenoh::{key_expr::KeyExpr, sample::SampleKind, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_sub_shm.rs

@@ -14,7 +14,7 @@
 use clap::Parser;
 #[cfg(all(feature = "shared-memory", feature = "unstable"))]
 use zenoh::shm::zshm;
-use zenoh::{bytes::ZBytes, config::Config, key_expr::KeyExpr, prelude::*};
+use zenoh::{bytes::ZBytes, config::Config, key_expr::KeyExpr};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_sub_thr.rs

@@ -14,7 +14,7 @@
 use std::time::Instant;
 
 use clap::Parser;
-use zenoh::{prelude::*, Config};
+use zenoh::{Config, Wait};
 use zenoh_examples::CommonArgs;
 
 struct Stats {
@@ -87,9 +87,7 @@ fn main() {
             }
         })
         .wait()
-        .unwrap()
-        // Make the subscriber run in background, until the session is closed.
-        .background();
+        .unwrap();
 
     println!("Press CTRL-C to quit...");
     std::thread::park();

examples/src/lib.rs

@@ -3,7 +3,7 @@
 //! Check ../README.md for usage.
 //!
 
-use zenoh::config::{Config, ValidatedMap};
+use zenoh::config::Config;
 
 #[derive(clap::ValueEnum, Clone, Copy, PartialEq, Eq, Hash, Debug)]
 pub enum Wai {

plugins/zenoh-backend-example/src/lib.rs

@@ -15,7 +15,7 @@ use std::collections::{hash_map::Entry, HashMap};
 
 use async_trait::async_trait;
 use tokio::sync::RwLock;
-use zenoh::{internal::Value, key_expr::OwnedKeyExpr, prelude::*, time::Timestamp};
+use zenoh::{internal::Value, key_expr::OwnedKeyExpr, time::Timestamp, Result as ZResult};
 use zenoh_backend_traits::{
     config::{StorageConfig, VolumeConfig},
     Capability, History, Persistence, Storage, StorageInsertionResult, StoredData, Volume,