add querier (#399)

* add querier

* add docs

* address review comments

* switch zenoh branch to main

Author: DenisBiryukov91

examples/README.md

@@ -106,6 +106,24 @@ or
       python3 z_get.py -s 'demo/**'
    ```
 
+### z_querier
+
+Continuously sends query messages for a selector.
+The queryables with a matching path or selector (for instance [z_queryable](#z_queryable) and [z_storage](#z_storage))
+will receive these queries and reply with paths/payloads that will be received by the querier's query callback.
+
+Typical usage:
+
+   ```bash
+      python3 z_querier.py
+   ```
+
+or
+
+   ```bash
+      python3 z_get.py -s 'demo/**'
+   ```
+
 ### z_queryable
 
 Creates a queryable function with a key expression.

examples/z_pub.py

@@ -12,11 +12,14 @@
 #   ZettaScale Zenoh Team, <[email protected]>
 #
 import time
+from typing import Optional
 
 import zenoh
 
 
-def main(conf: zenoh.Config, key: str, payload: str, iter: int, interval: int):
+def main(
+    conf: zenoh.Config, key: str, payload: str, iter: Optional[int], interval: int
+):
     # initiate logging
     zenoh.init_log_from_env_or("error")
 

examples/z_querier.py

@@ -0,0 +1,152 @@
+#
+# 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]>
+#
+import itertools
+import time
+from typing import Optional, Tuple
+
+import zenoh
+
+
+def main(
+    conf: zenoh.Config,
+    selector: str,
+    target: zenoh.QueryTarget,
+    payload: str,
+    timeout: float,
+    iter: Optional[int],
+):
+    # initiate logging
+    zenoh.init_log_from_env_or("error")
+    print("Opening session...")
+    with zenoh.open(conf) as session:
+        query_selector = zenoh.Selector(selector)
+
+        print(f"Declaring Querier on '{query_selector.key_expr}'...")
+        querier = session.declare_querier(
+            query_selector.key_expr, target=target, timeout=timeout
+        )
+
+        print("Press CTRL-C to quit...")
+        for idx in itertools.count() if iter is None else range(iter):
+            time.sleep(1.0)
+            buf = f"[{idx:4d}] {payload if payload else ''}"
+            print(f"Querying '{selector}' with payload '{buf}')...")
+
+            replies = querier.get(parameters=query_selector.parameters, payload=buf)
+            for reply in replies:
+                try:
+                    print(
+                        f">> Received ('{reply.ok.key_expr}': '{reply.ok.payload.to_string()}')"
+                    )
+                except:
+                    print(f">> Received (ERROR: '{reply.err.payload.to_string()}')")
+
+
+if __name__ == "__main__":
+    # --- Command line argument parsing --- --- --- --- --- ---
+    import argparse
+    import json
+
+    parser = argparse.ArgumentParser(
+        prog="z_querier", description="zenoh querier example"
+    )
+    parser.add_argument(
+        "--mode",
+        "-m",
+        dest="mode",
+        choices=["peer", "client"],
+        type=str,
+        help="The zenoh session mode.",
+    )
+    parser.add_argument(
+        "--connect",
+        "-e",
+        dest="connect",
+        metavar="ENDPOINT",
+        action="append",
+        type=str,
+        help="Endpoints to connect to.",
+    )
+    parser.add_argument(
+        "--listen",
+        "-l",
+        dest="listen",
+        metavar="ENDPOINT",
+        action="append",
+        type=str,
+        help="Endpoints to listen on.",
+    )
+    parser.add_argument(
+        "--selector",
+        "-s",
+        dest="selector",
+        default="demo/example/**",
+        type=str,
+        help="The selection of resources to query.",
+    )
+    parser.add_argument(
+        "--target",
+        "-t",
+        dest="target",
+        choices=["ALL", "BEST_MATCHING", "ALL_COMPLETE", "NONE"],
+        default="BEST_MATCHING",
+        type=str,
+        help="The target queryables of the query.",
+    )
+    parser.add_argument(
+        "--payload",
+        "-p",
+        dest="payload",
+        type=str,
+        help="An optional payload to send in the query.",
+    )
+    parser.add_argument(
+        "--timeout",
+        "-o",
+        dest="timeout",
+        default=10.0,
+        type=float,
+        help="The query timeout",
+    )
+    parser.add_argument(
+        "--config",
+        "-c",
+        dest="config",
+        metavar="FILE",
+        type=str,
+        help="A configuration file.",
+    )
+    parser.add_argument(
+        "--iter", dest="iter", type=int, help="How many gets to perform"
+    )
+
+    args = parser.parse_args()
+    conf = (
+        zenoh.Config.from_file(args.config)
+        if args.config is not None
+        else zenoh.Config()
+    )
+    if args.mode is not None:
+        conf.insert_json5("mode", json.dumps(args.mode))
+    if args.connect is not None:
+        conf.insert_json5("connect/endpoints", json.dumps(args.connect))
+    if args.listen is not None:
+        conf.insert_json5("listen/endpoints", json.dumps(args.listen))
+    target = {
+        "ALL": zenoh.QueryTarget.ALL,
+        "BEST_MATCHING": zenoh.QueryTarget.BEST_MATCHING,
+        "ALL_COMPLETE": zenoh.QueryTarget.ALL_COMPLETE,
+    }.get(args.target)
+
+    main(conf, args.selector, target, args.payload, args.timeout, args.iter)

src/lib.rs

@@ -60,8 +60,8 @@ pub(crate) mod zenoh {
         pubsub::{Publisher, Subscriber},
         qos::{CongestionControl, Priority, Reliability},
         query::{
-            ConsolidationMode, Parameters, Query, QueryConsolidation, QueryTarget, Queryable,
-            Reply, ReplyError, Selector,
+            ConsolidationMode, Parameters, Querier, Query, QueryConsolidation, QueryTarget,
+            Queryable, Reply, ReplyError, Selector,
         },
         sample::{Sample, SampleKind},
         scouting::{scout, Hello, Scout},

src/query.rs

@@ -21,7 +21,7 @@ use pyo3::{
 use crate::{
     bytes::{Encoding, ZBytes},
     config::ZenohId,
-    handlers::HandlerImpl,
+    handlers::{into_handler, HandlerImpl},
     key_expr::KeyExpr,
     macros::{build, downcast_or_new, enum_mapper, option_wrapper, wrapper},
     qos::{CongestionControl, Priority},
@@ -284,6 +284,61 @@ impl Queryable {
     }
 }
 
+option_wrapper!(zenoh::query::Querier<'static>, "Undeclared querier");
+
+#[pymethods]
+impl Querier {
+    #[classmethod]
+    fn __class_getitem__(cls: &Bound<PyType>, args: &Bound<PyAny>) -> PyObject {
+        generic(cls, args)
+    }
+
+    fn __enter__<'a, 'py>(this: &'a Bound<'py, Self>) -> PyResult<&'a Bound<'py, Self>> {
+        Self::check(this)
+    }
+
+    #[pyo3(signature = (*_args, **_kwargs))]
+    fn __exit__(
+        &mut self,
+        py: Python,
+        _args: &Bound<PyTuple>,
+        _kwargs: Option<&Bound<PyDict>>,
+    ) -> PyResult<PyObject> {
+        self.undeclare(py)?;
+        Ok(py.None())
+    }
+
+    #[getter]
+    fn key_expr(&self) -> PyResult<KeyExpr> {
+        Ok(self.get_ref()?.key_expr().clone().into())
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    #[pyo3(signature = (handler = None, *, parameters = None, payload = None, encoding = None, attachment = None))]
+    fn get(
+        &self,
+        py: Python,
+        handler: Option<&Bound<PyAny>>,
+        #[pyo3(from_py_with = "Parameters::from_py_opt")] parameters: Option<Parameters>,
+        #[pyo3(from_py_with = "ZBytes::from_py_opt")] payload: Option<ZBytes>,
+        #[pyo3(from_py_with = "Encoding::from_py_opt")] encoding: Option<Encoding>,
+        #[pyo3(from_py_with = "ZBytes::from_py_opt")] attachment: Option<ZBytes>,
+    ) -> PyResult<HandlerImpl<Reply>> {
+        let this = self.get_ref()?;
+        let (handler, _) = into_handler(py, handler)?;
+        let builder = build!(this.get(), parameters, payload, encoding, attachment);
+        wait(py, builder.with(handler)).map_into()
+    }
+
+    fn undeclare(&mut self, py: Python) -> PyResult<()> {
+        wait(py, self.take()?.undeclare())
+    }
+
+    fn __repr__(&self) -> PyResult<String> {
+        Ok(format!("{:?}", self.get_ref()?))
+    }
+}
+
 wrapper!(zenoh::query::Selector<'static>: Clone);
 downcast_or_new!(Selector, None);
 

src/session.rs

@@ -28,7 +28,7 @@ use crate::{
     macros::{build, wrapper},
     pubsub::{Publisher, Subscriber},
     qos::{CongestionControl, Priority, Reliability},
-    query::{QueryConsolidation, QueryTarget, Queryable, Reply, Selector},
+    query::{Querier, QueryConsolidation, QueryTarget, Queryable, Reply, Selector},
     time::Timestamp,
     utils::{timeout, wait, IntoPython, MapInto},
 };
@@ -225,6 +225,33 @@ impl Session {
         wait(py, builder).map_into()
     }
 
+    #[allow(clippy::too_many_arguments)]
+    #[pyo3(signature = (key_expr, *, target = None, consolidation = None, timeout = None, congestion_control = None, priority = None, express = None))]
+    fn declare_querier(
+        &self,
+        py: Python,
+        #[pyo3(from_py_with = "KeyExpr::from_py")] key_expr: KeyExpr,
+        target: Option<QueryTarget>,
+        #[pyo3(from_py_with = "QueryConsolidation::from_py_opt")] consolidation: Option<
+            QueryConsolidation,
+        >,
+        #[pyo3(from_py_with = "timeout")] timeout: Option<Duration>,
+        congestion_control: Option<CongestionControl>,
+        priority: Option<Priority>,
+        express: Option<bool>,
+    ) -> PyResult<Querier> {
+        let builder = build!(
+            self.0.declare_querier(key_expr),
+            target,
+            consolidation,
+            timeout,
+            congestion_control,
+            priority,
+            express,
+        );
+        wait(py, builder).map_into()
+    }
+
     fn liveliness(&self) -> Liveliness {
         Liveliness(self.0.clone())
     }

tests/examples_check.py

@@ -149,6 +149,43 @@ def test_z_get_z_queryable():
     assert not z_queryable.errors
 
 
+def test_z_querier_z_queryable():
+    """Test z_querier & z_queryable"""
+    z_queryable = Pyrun("z_queryable.py", ["-k=demo/example/zenoh-python-queryable"])
+    time.sleep(3)
+    ## z_querier: Able to get reply from queryable
+    z_querier = Pyrun(
+        "z_querier.py", ["-s=demo/example/zenoh-python-queryable", "-p=value"]
+    )
+    time.sleep(5)
+    z_queryable.interrupt()
+    z_querier.interrupt()
+
+    if not (
+        "Received ('demo/example/zenoh-python-queryable': 'Queryable from Python!')"
+        in "".join(z_querier.stdout)
+    ):
+        z_querier.dbg()
+        z_queryable.dbg()
+        z_querier.errors.append("z_querier didn't get a response from z_queryable")
+    queryableout = "".join(z_queryable.stdout)
+    if not (
+        "Received Query 'demo/example/zenoh-python-queryable' with payload: [   0] value"
+        in queryableout
+    ):
+        z_queryable.errors.append("z_queryable didn't catch query [0]")
+    elif not (
+        "Received Query 'demo/example/zenoh-python-queryable' with payload: [   2] value"
+        in queryableout
+    ):
+        z_queryable.errors.append("z_queryable didn't catch query [2]")
+    if any(("z_queryable" in error) for error in z_queryable.errors):
+        z_queryable.dbg()
+
+    assert not z_querier.errors
+    assert not z_queryable.errors
+
+
 def test_z_storage_z_sub():
     """Test z_storage & z_sub."""
     z_storage = Pyrun("z_storage.py")