add Querier

commons/zenoh-config/src/lib.rs

@@ -359,6 +359,8 @@ validated_struct::validator! {
             subscribers: Vec<OwnedKeyExpr>,
             /// A list of key-expressions for which all included publishers will be aggregated into.
             publishers: Vec<OwnedKeyExpr>,
+            /// A list of key-expressions for which all included queriers will be aggregated into.
+            queriers: Vec<OwnedKeyExpr>,
         },
         pub transport: #[derive(Default)]
         TransportConf {

examples/examples/z_querier.rs

@@ -0,0 +1,142 @@
+//
+// Copyright (c) 2024 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <[email protected]>
+//
+use std::time::Duration;
+
+use clap::Parser;
+use zenoh::{key_expr::KeyExpr, query::QueryTarget, Config};
+use zenoh_examples::CommonArgs;
+
+#[tokio::main]
+async fn main() {
+    // initiate logging
+    zenoh::init_log_from_env_or("error");
+
+    let (config, keyexpr, payload, target, timeout) = parse_args();
+
+    println!("Opening session...");
+    let session = zenoh::open(config).await.unwrap();
+
+    println!("Declaring Querier on '{keyexpr}'...");
+    let querier = session
+        .declare_querier(keyexpr)
+        .target(target)
+        .timeout(timeout)
+        .await
+        .unwrap();
+
+    #[cfg(feature = "unstable")]
+    querier
+        .matching_listener()
+        .callback(|matching_status| {
+            if matching_status.matching() {
+                println!("Querier has matching queryables.");
+            } else {
+                println!("Querier has NO MORE matching queryables.");
+            }
+        })
+        .background()
+        .await
+        .unwrap();
+
+    println!("Press CTRL-C to quit...");
+    for idx in 0..u32::MAX {
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        let buf = format!("[{idx:4}] {}", payload.clone().unwrap_or_default());
+        println!(
+            "Querying '{}' with payload: '{}')...",
+            &querier.key_expr(),
+            buf
+        );
+        let replies = querier
+            .get()
+            // // By default get receives replies from a FIFO.
+            // // Uncomment this line to use a ring channel instead.
+            // // More information on the ring channel are available in the z_pull example.
+            // .with(zenoh::handlers::RingChannel::default())
+            // Refer to z_bytes.rs to see how to serialize different types of message
+            .payload(buf)
+            .await
+            .unwrap();
+        while let Ok(reply) = replies.recv_async().await {
+            match reply.result() {
+                Ok(sample) => {
+                    // Refer to z_bytes.rs to see how to deserialize different types of message
+                    let payload = sample
+                        .payload()
+                        .try_to_string()
+                        .unwrap_or_else(|e| e.to_string().into());
+                    println!(
+                        ">> Received ('{}': '{}')",
+                        sample.key_expr().as_str(),
+                        payload,
+                    );
+                }
+                Err(err) => {
+                    let payload = err
+                        .payload()
+                        .try_to_string()
+                        .unwrap_or_else(|e| e.to_string().into());
+                    println!(">> Received (ERROR: '{}')", payload);
+                }
+            }
+        }
+    }
+}
+
+#[derive(clap::ValueEnum, Clone, Copy, Debug)]
+#[value(rename_all = "SCREAMING_SNAKE_CASE")]
+enum Qt {
+    BestMatching,
+    All,
+    AllComplete,
+}
+
+#[derive(Parser, Clone, Debug)]
+struct Args {
+    #[arg(short, long, default_value = "demo/example/**")]
+    /// The selection of resources to query
+    key_expr: KeyExpr<'static>,
+    #[arg(short, long)]
+    /// An optional payload to put in the query.
+    payload: Option<String>,
+    #[arg(short, long, default_value = "BEST_MATCHING")]
+    /// The target queryables of the query.
+    target: Qt,
+    #[arg(short = 'o', long, default_value = "10000")]
+    /// The query timeout in milliseconds.
+    timeout: u64,
+    #[command(flatten)]
+    common: CommonArgs,
+}
+
+fn parse_args() -> (
+    Config,
+    KeyExpr<'static>,
+    Option<String>,
+    QueryTarget,
+    Duration,
+) {
+    let args = Args::parse();
+    (
+        args.common.into(),
+        args.key_expr,
+        args.payload,
+        match args.target {
+            Qt::BestMatching => QueryTarget::BestMatching,
+            Qt::All => QueryTarget::All,
+            Qt::AllComplete => QueryTarget::AllComplete,
+        },
+        Duration::from_millis(args.timeout),
+    )
+}

zenoh/src/api/admin.rs

@@ -48,9 +48,7 @@ lazy_static::lazy_static!(
 
 pub(crate) fn init(session: WeakSession) {
     if let Ok(own_zid) = keyexpr::new(&session.zid().to_string()) {
-        let admin_key = KeyExpr::from(*KE_PREFIX / own_zid / *KE_SESSION / *KE_STARSTAR)
-            .to_wire(&session)
-            .to_owned();
+        let admin_key = KeyExpr::from(*KE_PREFIX / own_zid / *KE_SESSION / *KE_STARSTAR);
 
         let _admin_qabl = session.declare_queryable_inner(
             &admin_key,

zenoh/src/api/builders/matching_listener.rs

@@ -22,21 +22,31 @@ use zenoh_result::ZResult;
 use {
     crate::api::{
         handlers::{Callback, DefaultHandler, IntoHandler},
-        publisher::{MatchingListener, MatchingListenerInner, MatchingStatus, Publisher},
+        publisher::{MatchingListener, MatchingListenerInner, MatchingStatus, MatchingStatusType},
+        Id,
     },
+    crate::sample::Locality,
     std::sync::Arc,
+    std::{collections::HashSet, sync::Mutex},
 };
 
+#[cfg(feature = "unstable")]
+use crate::{api::session::WeakSession, key_expr::KeyExpr};
+
 /// A builder for initializing a [`MatchingListener`].
 #[zenoh_macros::unstable]
 #[derive(Debug)]
-pub struct MatchingListenerBuilder<'a, 'b, Handler, const BACKGROUND: bool = false> {
-    pub(crate) publisher: &'a Publisher<'b>,
+pub struct MatchingListenerBuilder<'a, Handler, const BACKGROUND: bool = false> {
+    pub(crate) session: &'a WeakSession,
+    pub(crate) key_expr: &'a KeyExpr<'a>,
+    pub(crate) destination: Locality,
+    pub(crate) matching_listeners: &'a Arc<Mutex<HashSet<Id>>>,
+    pub(crate) matching_status_type: MatchingStatusType,
     pub handler: Handler,
 }
 
 #[zenoh_macros::unstable]
-impl<'a, 'b> MatchingListenerBuilder<'a, 'b, DefaultHandler> {
+impl<'a> MatchingListenerBuilder<'a, DefaultHandler> {
     /// Receive the MatchingStatuses for this listener with a callback.
     ///
     /// # Examples
@@ -49,7 +59,7 @@ impl<'a, 'b> MatchingListenerBuilder<'a, 'b, DefaultHandler> {
     /// let matching_listener = publisher
     ///     .matching_listener()
     ///     .callback(|matching_status| {
-    ///         if matching_status.matching_subscribers() {
+    ///         if matching_status.matching() {
     ///             println!("Publisher has matching subscribers.");
     ///         } else {
     ///             println!("Publisher has NO MORE matching subscribers.");
@@ -61,10 +71,7 @@ impl<'a, 'b> MatchingListenerBuilder<'a, 'b, DefaultHandler> {
     /// ```
     #[inline]
     #[zenoh_macros::unstable]
-    pub fn callback<F>(
-        self,
-        callback: F,
-    ) -> MatchingListenerBuilder<'a, 'b, Callback<MatchingStatus>>
+    pub fn callback<F>(self, callback: F) -> MatchingListenerBuilder<'a, Callback<MatchingStatus>>
     where
         F: Fn(MatchingStatus) + Send + Sync + 'static,
     {
@@ -93,7 +100,7 @@ impl<'a, 'b> MatchingListenerBuilder<'a, 'b, DefaultHandler> {
     pub fn callback_mut<F>(
         self,
         callback: F,
-    ) -> MatchingListenerBuilder<'a, 'b, Callback<MatchingStatus>>
+    ) -> MatchingListenerBuilder<'a, Callback<MatchingStatus>>
     where
         F: FnMut(MatchingStatus) + Send + Sync + 'static,
     {
@@ -115,7 +122,7 @@ impl<'a, 'b> MatchingListenerBuilder<'a, 'b, DefaultHandler> {
     ///     .await
     ///     .unwrap();
     /// while let Ok(matching_status) = matching_listener.recv_async().await {
-    ///     if matching_status.matching_subscribers() {
+    ///     if matching_status.matching() {
     ///         println!("Publisher has matching subscribers.");
     ///     } else {
     ///         println!("Publisher has NO MORE matching subscribers.");
@@ -125,20 +132,31 @@ impl<'a, 'b> MatchingListenerBuilder<'a, 'b, DefaultHandler> {
     /// ```
     #[inline]
     #[zenoh_macros::unstable]
-    pub fn with<Handler>(self, handler: Handler) -> MatchingListenerBuilder<'a, 'b, Handler>
+    pub fn with<Handler>(self, handler: Handler) -> MatchingListenerBuilder<'a, Handler>
     where
         Handler: IntoHandler<MatchingStatus>,
     {
         let MatchingListenerBuilder {
-            publisher,
+            session,
+            key_expr,
+            destination,
+            matching_listeners,
+            matching_status_type,
             handler: _,
         } = self;
-        MatchingListenerBuilder { publisher, handler }
+        MatchingListenerBuilder {
+            session,
+            key_expr,
+            destination,
+            matching_listeners,
+            matching_status_type,
+            handler,
+        }
     }
 }
 
 #[zenoh_macros::unstable]
-impl<'a, 'b> MatchingListenerBuilder<'a, 'b, Callback<MatchingStatus>> {
+impl<'a> MatchingListenerBuilder<'a, Callback<MatchingStatus>> {
     /// Register the listener callback to be run in background until the publisher is undeclared.
     ///
     /// Background builder doesn't return a `MatchingListener` object anymore.
@@ -154,7 +172,7 @@ impl<'a, 'b> MatchingListenerBuilder<'a, 'b, Callback<MatchingStatus>> {
     /// publisher
     ///     .matching_listener()
     ///     .callback(|matching_status| {
-    ///         if matching_status.matching_subscribers() {
+    ///         if matching_status.matching() {
     ///             println!("Publisher has matching subscribers.");
     ///         } else {
     ///             println!("Publisher has NO MORE matching subscribers.");
@@ -165,16 +183,20 @@ impl<'a, 'b> MatchingListenerBuilder<'a, 'b, Callback<MatchingStatus>> {
     ///     .unwrap();
     /// # }
     /// ```
-    pub fn background(self) -> MatchingListenerBuilder<'a, 'b, Callback<MatchingStatus>, true> {
+    pub fn background(self) -> MatchingListenerBuilder<'a, Callback<MatchingStatus>, true> {
         MatchingListenerBuilder {
-            publisher: self.publisher,
+            session: self.session,
+            destination: self.destination,
+            matching_listeners: self.matching_listeners,
+            key_expr: self.key_expr,
+            matching_status_type: self.matching_status_type,
             handler: self.handler,
         }
     }
 }
 
 #[zenoh_macros::unstable]
-impl<Handler> Resolvable for MatchingListenerBuilder<'_, '_, Handler>
+impl<Handler> Resolvable for MatchingListenerBuilder<'_, Handler>
 where
     Handler: IntoHandler<MatchingStatus> + Send,
     Handler::Handler: Send,
@@ -183,23 +205,25 @@ where
 }
 
 #[zenoh_macros::unstable]
-impl<Handler> Wait for MatchingListenerBuilder<'_, '_, Handler>
+impl<Handler> Wait for MatchingListenerBuilder<'_, Handler>
 where
     Handler: IntoHandler<MatchingStatus> + Send,
     Handler::Handler: Send,
 {
     #[zenoh_macros::unstable]
     fn wait(self) -> <Self as Resolvable>::To {
         let (callback, handler) = self.handler.into_handler();
-        let state = self
-            .publisher
-            .session
-            .declare_matches_listener_inner(self.publisher, callback)?;
-        zlock!(self.publisher.matching_listeners).insert(state.id);
+        let state = self.session.declare_matches_listener_inner(
+            self.key_expr,
+            self.destination,
+            self.matching_status_type,
+            callback,
+        )?;
+        zlock!(self.matching_listeners).insert(state.id);
         Ok(MatchingListener {
             inner: MatchingListenerInner {
-                session: self.publisher.session.clone(),
-                matching_listeners: self.publisher.matching_listeners.clone(),
+                session: self.session.clone(),
+                matching_listeners: self.matching_listeners.clone(),
                 id: state.id,
                 undeclare_on_drop: true,
             },
@@ -209,7 +233,7 @@ where
 }
 
 #[zenoh_macros::unstable]
-impl<Handler> IntoFuture for MatchingListenerBuilder<'_, '_, Handler>
+impl<Handler> IntoFuture for MatchingListenerBuilder<'_, Handler>
 where
     Handler: IntoHandler<MatchingStatus> + Send,
     Handler::Handler: Send,
@@ -224,25 +248,27 @@ where
 }
 
 #[zenoh_macros::unstable]
-impl Resolvable for MatchingListenerBuilder<'_, '_, Callback<MatchingStatus>, true> {
+impl Resolvable for MatchingListenerBuilder<'_, Callback<MatchingStatus>, true> {
     type To = ZResult<()>;
 }
 
 #[zenoh_macros::unstable]
-impl Wait for MatchingListenerBuilder<'_, '_, Callback<MatchingStatus>, true> {
+impl Wait for MatchingListenerBuilder<'_, Callback<MatchingStatus>, true> {
     #[zenoh_macros::unstable]
     fn wait(self) -> <Self as Resolvable>::To {
-        let state = self
-            .publisher
-            .session
-            .declare_matches_listener_inner(self.publisher, self.handler)?;
-        zlock!(self.publisher.matching_listeners).insert(state.id);
+        let state = self.session.declare_matches_listener_inner(
+            self.key_expr,
+            self.destination,
+            self.matching_status_type,
+            self.handler,
+        )?;
+        zlock!(self.matching_listeners).insert(state.id);
         Ok(())
     }
 }
 
 #[zenoh_macros::unstable]
-impl IntoFuture for MatchingListenerBuilder<'_, '_, Callback<MatchingStatus>, true> {
+impl IntoFuture for MatchingListenerBuilder<'_, Callback<MatchingStatus>, true> {
     type Output = <Self as Resolvable>::To;
     type IntoFuture = Ready<<Self as Resolvable>::To>;
 

zenoh/src/api/builders/mod.rs

@@ -15,6 +15,7 @@
 pub(crate) mod info;
 pub(crate) mod matching_listener;
 pub(crate) mod publisher;
+pub(crate) mod querier;
 pub(crate) mod query;
 pub(crate) mod queryable;
 pub(crate) mod reply;