feat: unify pub/sub and query/reply modules

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

@@ -14,7 +14,10 @@
 use std::{convert::TryFrom, time::Duration};
 
 use zenoh::{
-    config::Config, key_expr::KeyExpr, prelude::*, query::QueryTarget, selector::Selector,
+    config::Config,
+    key_expr::KeyExpr,
+    prelude::*,
+    query::{QueryTarget, Selector},
 };
 
 #[tokio::main]

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::Canonizable, OwnedKeyExpr, FORBIDDEN_CHARS};
+use super::{canon::Canonize, OwnedKeyExpr, FORBIDDEN_CHARS};
 
 /// A [`str`] newtype that is statically known to be a valid key expression.
 ///
@@ -72,7 +72,7 @@ impl keyexpr {
     pub fn autocanonize<'a, T, E>(t: &'a mut T) -> Result<&'a Self, E>
     where
         &'a Self: TryFrom<&'a T, Error = E>,
-        T: Canonizable + ?Sized,
+        T: Canonize + ?Sized,
     {
         t.canonize();
         Self::new(t)

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

@@ -19,13 +19,13 @@ use crate::key_expr::{
     DELIMITER, DOUBLE_WILD, SINGLE_WILD,
 };
 
-pub trait Canonizable {
+pub trait Canonize {
     fn canonize(&mut self);
 }
 
 const DOLLAR_STAR: &[u8; 2] = b"$*";
 
-impl Canonizable for &mut str {
+impl Canonize for &mut str {
     fn canonize(&mut self) {
         let mut writer = Writer {
             ptr: self.as_mut_ptr(),
@@ -114,7 +114,7 @@ impl Canonizable for &mut str {
     }
 }
 
-impl Canonizable for String {
+impl Canonize for String {
     fn canonize(&mut self) {
         let mut s = self.as_mut();
         s.canonize();

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

@@ -22,7 +22,7 @@ use core::{
     str::FromStr,
 };
 
-use super::{canon::Canonizable, keyexpr};
+use super::{canon::Canonize, keyexpr};
 
 /// A [`Arc<str>`] newtype that is statically known to be a valid key expression.
 ///
@@ -60,7 +60,7 @@ impl OwnedKeyExpr {
     pub fn autocanonize<T, E>(mut t: T) -> Result<Self, E>
     where
         Self: TryFrom<T, Error = E>,
-        T: Canonizable,
+        T: Canonize,
     {
         t.canonize();
         Self::new(t)

commons/zenoh-macros/src/lib.rs

@@ -287,7 +287,7 @@ fn keformat_support(source: &str) -> proc_macro2::TokenStream {
     let formatter_doc = format!("And instance of a formatter for `{source}`.");
 
     quote! {
-            use ::zenoh::core::Result as ZResult;
+            use ::zenoh::Result as ZResult;
             const FORMAT_INNER: ::zenoh::key_expr::format::KeFormat<'static, [::zenoh::key_expr::format::Segment<'static>; #len]> = unsafe {
                 ::zenoh::key_expr::format::macro_support::const_new(#source, [#(#segments)*])
             };

examples/examples/z_get.rs

@@ -14,7 +14,10 @@
 use std::time::Duration;
 
 use clap::Parser;
-use zenoh::{query::QueryTarget, selector::Selector, Config};
+use zenoh::{
+    query::{QueryTarget, Selector},
+    Config,
+};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_get_shm.rs

@@ -15,8 +15,7 @@ use std::time::Duration;
 
 use clap::Parser;
 use zenoh::{
-    query::QueryTarget,
-    selector::Selector,
+    query::{QueryTarget, Selector},
     shm::{
         zshm, BlockOn, GarbageCollect, PosixShmProviderBackend, ShmProviderBuilder,
         POSIX_PROTOCOL_ID,

examples/examples/z_info.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{info::ZenohId, prelude::*};
+use zenoh::{prelude::*, session::ZenohId};
 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::*, publisher::CongestionControl, Config};
+use zenoh::{bytes::ZBytes, key_expr::keyexpr, prelude::*, qos::CongestionControl, Config};
 use zenoh_examples::CommonArgs;
 
 fn main() {

examples/examples/z_ping_shm.rs

@@ -18,7 +18,7 @@ use zenoh::{
     bytes::ZBytes,
     key_expr::keyexpr,
     prelude::*,
-    publisher::CongestionControl,
+    qos::CongestionControl,
     shm::{PosixShmProviderBackend, ShmProviderBuilder, POSIX_PROTOCOL_ID},
     Config,
 };

examples/examples/z_pong.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::keyexpr, prelude::*, publisher::CongestionControl, Config};
+use zenoh::{key_expr::keyexpr, prelude::*, qos::CongestionControl, Config};
 use zenoh_examples::CommonArgs;
 
 fn main() {

examples/examples/z_pub.rs

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

examples/examples/z_pub_shm_thr.rs

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

examples/examples/z_pub_thr.rs

@@ -15,7 +15,11 @@
 use std::convert::TryInto;
 
 use clap::Parser;
-use zenoh::{bytes::ZBytes, core::Priority, prelude::*, publisher::CongestionControl};
+use zenoh::{
+    bytes::ZBytes,
+    prelude::*,
+    qos::{CongestionControl, Priority},
+};
 use zenoh_examples::CommonArgs;
 
 fn main() {

plugins/zenoh-backend-traits/src/config.rs

@@ -18,8 +18,8 @@ use derive_more::{AsMut, AsRef};
 use schemars::JsonSchema;
 use serde_json::{Map, Value};
 use zenoh::{
-    core::Result as ZResult,
     key_expr::{keyexpr, OwnedKeyExpr},
+    Result as ZResult,
 };
 use zenoh_plugin_trait::{PluginStartArgs, StructVersion};
 use zenoh_result::{bail, zerror, Error};

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

@@ -124,10 +124,10 @@
 use async_trait::async_trait;
 use const_format::concatcp;
 use zenoh::{
-    core::Result as ZResult,
     internal::Value,
     key_expr::{keyexpr, OwnedKeyExpr},
     time::Timestamp,
+    Result as ZResult,
 };
 use zenoh_plugin_trait::{PluginControl, PluginInstance, PluginStatusRec, StructVersion};
 use zenoh_util::concat_enabled_features;

plugins/zenoh-plugin-rest/examples/z_serve_sse.rs

@@ -15,7 +15,9 @@ use std::time::Duration;
 
 use clap::{arg, Command};
 use zenoh::{
-    config::Config, key_expr::keyexpr, publisher::CongestionControl, sample::QoSBuilderTrait,
+    config::Config,
+    key_expr::keyexpr,
+    qos::{CongestionControl, QoSBuilderTrait},
     session::SessionDeclarations,
 };
 

plugins/zenoh-plugin-rest/src/lib.rs

@@ -26,8 +26,7 @@ use http_types::Method;
 use serde::{Deserialize, Serialize};
 use tide::{http::Mime, sse::Sender, Request, Response, Server, StatusCode};
 use zenoh::{
-    bytes::ZBytes,
-    encoding::Encoding,
+    bytes::{Encoding, ZBytes},
     internal::{
         bail,
         plugins::{RunningPluginTrait, ZenohPlugin},
@@ -36,16 +35,15 @@ use zenoh::{
     },
     key_expr::{keyexpr, KeyExpr},
     prelude::*,
-    query::{QueryConsolidation, Reply},
-    sample::{EncodingBuilderTrait, Sample, SampleKind},
-    selector::{Parameters, Selector, ZenohParameters},
+    query::{Parameters, QueryConsolidation, Reply, Selector, ZenohParameters},
+    sample::{Sample, SampleKind},
     session::{Session, SessionDeclarations},
 };
 use zenoh_plugin_trait::{plugin_long_version, plugin_version, Plugin, PluginControl};
 
 mod config;
 pub use config::Config;
-use zenoh::query::ReplyError;
+use zenoh::{bytes::EncodingBuilderTrait, query::ReplyError};
 
 const GIT_VERSION: &str = git_version::git_version!(prefix = "v", cargo_prefix = "v");
 lazy_static::lazy_static! {

plugins/zenoh-plugin-storage-manager/src/backends_mgt.rs

@@ -14,7 +14,7 @@
 use std::sync::Arc;
 
 use flume::Sender;
-use zenoh::{core::Result as ZResult, session::Session};
+use zenoh::{session::Session, Result as ZResult};
 use zenoh_backend_traits::{config::StorageConfig, Capability, VolumeInstance};
 
 use super::storages_mgt::*;

plugins/zenoh-plugin-storage-manager/src/lib.rs

@@ -30,7 +30,6 @@ use flume::Sender;
 use memory_backend::MemoryBackend;
 use storages_mgt::StorageMessage;
 use zenoh::{
-    core::Result as ZResult,
     internal::{
         plugins::{Response, RunningPlugin, RunningPluginTrait, ZenohPlugin},
         runtime::Runtime,
@@ -39,6 +38,7 @@ use zenoh::{
     key_expr::{keyexpr, KeyExpr},
     prelude::Wait,
     session::Session,
+    Result as ZResult,
 };
 use zenoh_backend_traits::{
     config::{ConfigDiff, PluginConfig, StorageConfig, VolumeConfig},

plugins/zenoh-plugin-storage-manager/src/memory_backend/mod.rs

@@ -15,7 +15,7 @@ use std::{collections::HashMap, sync::Arc};
 
 use async_std::sync::RwLock;
 use async_trait::async_trait;
-use zenoh::{core::Result as ZResult, internal::Value, key_expr::OwnedKeyExpr, time::Timestamp};
+use zenoh::{internal::Value, key_expr::OwnedKeyExpr, time::Timestamp, Result as ZResult};
 use zenoh_backend_traits::{
     config::{StorageConfig, VolumeConfig},
     *,

plugins/zenoh-plugin-storage-manager/src/replica/align_queryable.rs

@@ -21,7 +21,7 @@ use std::{
 
 use async_std::sync::Arc;
 use zenoh::{
-    internal::Value, key_expr::OwnedKeyExpr, prelude::*, sample::Sample, selector::Parameters,
+    internal::Value, key_expr::OwnedKeyExpr, prelude::*, query::Parameters, sample::Sample,
     time::Timestamp, Session,
 };
 

plugins/zenoh-plugin-storage-manager/src/replica/aligner.rs

@@ -24,8 +24,8 @@ use zenoh::{
     internal::Value,
     key_expr::{KeyExpr, OwnedKeyExpr},
     prelude::*,
+    query::Selector,
     sample::{Sample, SampleBuilder},
-    selector::Selector,
     time::Timestamp,
     Session,
 };

plugins/zenoh-plugin-storage-manager/src/replica/snapshotter.rs

@@ -24,7 +24,7 @@ use async_std::{
 };
 use flume::Receiver;
 use futures::join;
-use zenoh::{info::ZenohId, key_expr::OwnedKeyExpr, time::Timestamp};
+use zenoh::{key_expr::OwnedKeyExpr, session::ZenohId, time::Timestamp};
 use zenoh_backend_traits::config::ReplicaConfig;
 
 use super::{Digest, DigestConfig, LogEntry};

plugins/zenoh-plugin-storage-manager/src/replica/storage.rs

@@ -22,7 +22,7 @@ use async_trait::async_trait;
 use flume::{Receiver, Sender};
 use futures::select;
 use zenoh::{
-    core::Result as ZResult,
+    bytes::EncodingBuilderTrait,
     internal::{
         bail,
         buffers::{SplitBuffer, ZBuf},
@@ -35,9 +35,10 @@ use zenoh::{
         KeyExpr, OwnedKeyExpr,
     },
     query::{ConsolidationMode, QueryTarget},
-    sample::{EncodingBuilderTrait, Sample, SampleBuilder, SampleKind, TimestampBuilderTrait},
+    sample::{Sample, SampleBuilder, SampleKind, TimestampBuilderTrait},
     session::{Session, SessionDeclarations},
     time::{new_timestamp, Timestamp, NTP64},
+    Result as ZResult,
 };
 use zenoh_backend_traits::{
     config::{GarbageCollectionConfig, StorageConfig},

plugins/zenoh-plugin-storage-manager/src/storages_mgt.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use async_std::sync::Arc;
-use zenoh::{core::Result as ZResult, session::Session};
+use zenoh::{session::Session, Result as ZResult};
 use zenoh_backend_traits::config::StorageConfig;
 
 pub use super::replica::{Replica, StorageService};

zenoh-ext/src/group.rs

@@ -28,11 +28,11 @@ use serde::{Deserialize, Serialize};
 use tokio::sync::Mutex;
 use zenoh::{
     bytes::ZBytesReader,
-    core::Priority,
     internal::{bail, Condition, TaskController},
     key_expr::{keyexpr, KeyExpr, OwnedKeyExpr},
     prelude::*,
-    publisher::Publisher,
+    pubsub::Publisher,
+    qos::Priority,
     Session,
 };
 

zenoh-ext/src/lib.rs

@@ -22,7 +22,7 @@ pub use querying_subscriber::{
 };
 pub use session_ext::SessionExt;
 pub use subscriber_ext::{SubscriberBuilderExt, SubscriberForward};
-use zenoh::{core::Result as ZResult, internal::zerror, query::Reply, sample::Sample};
+use zenoh::{internal::zerror, query::Reply, sample::Sample, Result as ZResult};
 
 /// The space of keys to use in a [`FetchingSubscriber`].
 pub enum KeySpace {

zenoh-ext/src/publication_cache.rs

@@ -19,16 +19,14 @@ use std::{
 };
 
 use zenoh::{
-    core::{Error, Resolvable, Resolve, Result as ZResult},
     internal::{bail, runtime::ZRuntime, ResolveFuture, TerminatableTask},
     key_expr::{keyexpr, KeyExpr, OwnedKeyExpr},
     prelude::Wait,
-    query::Query,
-    queryable::Queryable,
+    pubsub::FlumeSubscriber,
+    query::{Query, Queryable, ZenohParameters},
     sample::{Locality, Sample},
-    selector::ZenohParameters,
     session::{SessionDeclarations, SessionRef},
-    subscriber::FlumeSubscriber,
+    Error, Resolvable, Resolve, Result as ZResult,
 };
 
 /// The builder of PublicationCache, allowing to configure it.

zenoh-ext/src/querying_subscriber.rs

@@ -21,17 +21,16 @@ use std::{
 };
 
 use zenoh::{
-    core::{Error, Resolvable, Resolve, Result as ZResult},
     handlers::{locked, DefaultHandler, IntoHandler},
     internal::zlock,
     key_expr::KeyExpr,
     prelude::Wait,
-    query::{QueryConsolidation, QueryTarget, ReplyKeyExpr},
+    pubsub::{Reliability, Subscriber},
+    query::{QueryConsolidation, QueryTarget, ReplyKeyExpr, Selector},
     sample::{Locality, Sample, SampleBuilder, TimestampBuilderTrait},
-    selector::Selector,
     session::{SessionDeclarations, SessionRef},
-    subscriber::{Reliability, Subscriber},
     time::{new_timestamp, Timestamp},
+    Error, Resolvable, Resolve, Result as ZResult,
 };
 
 use crate::ExtractSample;