Merge branch 'main' into api/make-replier-id-global

Author: oteffahi

examples/z_pub.py

@@ -18,7 +18,12 @@
 
 
 def main(
-    conf: zenoh.Config, key: str, payload: str, iter: Optional[int], interval: int
+    conf: zenoh.Config,
+    key: str,
+    payload: str,
+    iter: Optional[int],
+    interval: int,
+    add_matching_listener: bool,
 ):
     # initiate logging
     zenoh.init_log_from_env_or("error")
@@ -28,6 +33,16 @@ def main(
         print(f"Declaring Publisher on '{key}'...")
         pub = session.declare_publisher(key)
 
+        if add_matching_listener:
+
+            def on_matching_status_update(status: zenoh.MatchingStatus):
+                if status.matching:
+                    print("Publisher has matching subscribers.")
+                else:
+                    print("Publisher has NO MORE matching subscribers")
+
+            pub.declare_matching_listener(on_matching_status_update)
+
         print("Press CTRL-C to quit...")
         for idx in itertools.count() if iter is None else range(iter):
             time.sleep(interval)
@@ -71,8 +86,21 @@ def main(
         default=1.0,
         help="Interval between each put",
     )
+    parser.add_argument(
+        "--add-matching-listener",
+        default=False,
+        action="store_true",
+        help="Add matching listener",
+    )
 
     args = parser.parse_args()
     conf = common.get_config_from_args(args)
 
-    main(conf, args.key, args.payload, args.iter, args.interval)
+    main(
+        conf,
+        args.key,
+        args.payload,
+        args.iter,
+        args.interval,
+        args.add_matching_listener,
+    )

examples/z_querier.py

@@ -13,7 +13,7 @@
 #
 import itertools
 import time
-from typing import Optional, Tuple
+from typing import Optional
 
 import zenoh
 
@@ -25,6 +25,7 @@ def main(
     payload: str,
     timeout: float,
     iter: Optional[int],
+    add_matching_listener: bool,
 ):
     # initiate logging
     zenoh.init_log_from_env_or("error")
@@ -37,6 +38,16 @@ def main(
             query_selector.key_expr, target=target, timeout=timeout
         )
 
+        if add_matching_listener:
+
+            def on_matching_status_update(status: zenoh.MatchingStatus):
+                if status.matching:
+                    print("Querier has matching queryables.")
+                else:
+                    print("Querier has NO MORE matching queryables")
+
+            querier.declare_matching_listener(on_matching_status_update)
+
         print("Press CTRL-C to quit...")
         for idx in itertools.count() if iter is None else range(iter):
             time.sleep(1.0)
@@ -99,6 +110,12 @@ def main(
     parser.add_argument(
         "--iter", dest="iter", type=int, help="How many gets to perform"
     )
+    parser.add_argument(
+        "--add-matching-listener",
+        default=False,
+        action="store_true",
+        help="Add matching listener",
+    )
 
     args = parser.parse_args()
     conf = common.get_config_from_args(args)
@@ -109,4 +126,12 @@ def main(
         "ALL_COMPLETE": zenoh.QueryTarget.ALL_COMPLETE,
     }.get(args.target)
 
-    main(conf, args.selector, target, args.payload, args.timeout, args.iter)
+    main(
+        conf,
+        args.selector,
+        target,
+        args.payload,
+        args.timeout,
+        args.iter,
+        args.add_matching_listener,
+    )

src/ext.rs

@@ -21,8 +21,9 @@ use crate::{
     macros::{build, import, option_wrapper, py_static, try_import, wrapper},
     pubsub::Subscriber,
     qos::{CongestionControl, Priority, Reliability},
-    sample::Sample,
+    sample::{Locality, Sample},
     session::{EntityGlobalId, Session},
+    time::Timestamp,
     utils::{duration, generic, wait, MapInto},
     ZDeserializeError,
 };
@@ -481,26 +482,30 @@ impl AdvancedPublisher {
         Ok(self.get_ref()?.priority().into())
     }
 
-    // TODO add timestamp
-    #[pyo3(signature = (payload, *, encoding = None, attachment = None))]
+    #[pyo3(signature = (payload, *, encoding = None, attachment = None, timestamp = None))]
     fn put(
         &self,
         py: Python,
         #[pyo3(from_py_with = "ZBytes::from_py")] payload: ZBytes,
         #[pyo3(from_py_with = "Encoding::from_py_opt")] encoding: Option<Encoding>,
         #[pyo3(from_py_with = "ZBytes::from_py_opt")] attachment: Option<ZBytes>,
+        timestamp: Option<Timestamp>,
     ) -> PyResult<()> {
         let this = self.get_ref()?;
-        wait(py, build!(this.put(payload), encoding, attachment))
+        wait(
+            py,
+            build!(this.put(payload), encoding, attachment, timestamp),
+        )
     }
 
-    #[pyo3(signature = (*, attachment = None))]
+    #[pyo3(signature = (*, attachment = None, timestamp = None))]
     fn delete(
         &self,
         py: Python,
         #[pyo3(from_py_with = "ZBytes::from_py_opt")] attachment: Option<ZBytes>,
+        timestamp: Option<Timestamp>,
     ) -> PyResult<()> {
-        wait(py, build!(self.get_ref()?.delete(), attachment))
+        wait(py, build!(self.get_ref()?.delete(), attachment, timestamp))
     }
 
     fn undeclare(&mut self, py: Python) -> PyResult<()> {
@@ -743,7 +748,7 @@ impl SampleMissListener {
 
 #[allow(clippy::too_many_arguments)]
 #[pyfunction]
-#[pyo3(signature = (session, key_expr, *, encoding = None, congestion_control = None, priority = None, express = None, reliability = None, cache = None, sample_miss_detection = None, publisher_detection = None))]
+#[pyo3(signature = (session, key_expr, *, encoding = None, congestion_control = None, priority = None, express = None, reliability = None, allowed_destination = None, cache = None, sample_miss_detection = None, publisher_detection = None))]
 pub(crate) fn declare_advanced_publisher(
     py: Python,
     session: &Session,
@@ -753,6 +758,7 @@ pub(crate) fn declare_advanced_publisher(
     priority: Option<Priority>,
     express: Option<bool>,
     reliability: Option<Reliability>,
+    allowed_destination: Option<Locality>,
     cache: Option<CacheConfig>,
     sample_miss_detection: Option<MissDetectionConfig>,
     publisher_detection: Option<bool>,
@@ -764,6 +770,7 @@ pub(crate) fn declare_advanced_publisher(
         priority,
         express,
         reliability,
+        allowed_destination,
         cache,
         sample_miss_detection,
     );
@@ -775,19 +782,21 @@ pub(crate) fn declare_advanced_publisher(
 
 #[allow(clippy::too_many_arguments)]
 #[pyfunction]
-#[pyo3(signature = (session, key_expr, handler = None, *, history = None, recovery = None, subscriber_detection = None))]
+#[pyo3(signature = (session, key_expr, handler = None, *, allowed_origin = None, history = None, recovery = None, subscriber_detection = None))]
 pub(crate) fn declare_advanced_subscriber(
     session: &Session,
     py: Python,
     #[pyo3(from_py_with = "KeyExpr::from_py")] key_expr: KeyExpr,
     handler: Option<&Bound<PyAny>>,
+    allowed_origin: Option<Locality>,
     history: Option<HistoryConfig>,
     recovery: Option<RecoveryConfig>,
     subscriber_detection: Option<bool>,
 ) -> PyResult<AdvancedSubscriber> {
     let (handler, background) = into_handler(py, handler)?;
     let mut builder = build!(
         session.0.declare_subscriber(key_expr).advanced(),
+        allowed_origin,
         history,
         recovery
     );

src/lib.rs

@@ -21,6 +21,7 @@ mod handlers;
 mod key_expr;
 mod liveliness;
 mod macros;
+mod matching;
 mod pubsub;
 mod qos;
 mod query;
@@ -57,13 +58,14 @@ pub(crate) mod zenoh {
         handlers::Handler,
         key_expr::{KeyExpr, SetIntersectionLevel},
         liveliness::{Liveliness, LivelinessToken},
+        matching::{MatchingListener, MatchingStatus},
         pubsub::{Publisher, Subscriber},
         qos::{CongestionControl, Priority, Reliability},
         query::{
             ConsolidationMode, Parameters, Querier, Query, QueryConsolidation, QueryTarget,
             Queryable, Reply, ReplyError, Selector,
         },
-        sample::{Sample, SampleKind},
+        sample::{Locality, Sample, SampleKind},
         scouting::{scout, Hello, Scout},
         session::{open, EntityGlobalId, Session, SessionInfo},
         time::Timestamp,

src/matching.rs

@@ -0,0 +1,90 @@
+//
+// Copyright (c) 2025 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, <zenoh@zettascale.tech>
+//
+use pyo3::{
+    prelude::*,
+    types::{PyDict, PyIterator, PyTuple, PyType},
+};
+
+use crate::{
+    handlers::HandlerImpl,
+    macros::{option_wrapper, wrapper},
+    utils::{generic, wait},
+};
+
+wrapper!(zenoh::matching::MatchingStatus);
+
+#[pymethods]
+impl MatchingStatus {
+    #[getter]
+    fn matching(&self) -> bool {
+        self.0.matching()
+    }
+
+    fn __repr__(&self) -> String {
+        format!("{:?}", self.0)
+    }
+}
+
+option_wrapper!(
+    zenoh::matching::MatchingListener<HandlerImpl<MatchingStatus>>,
+    "Undeclared matching listener"
+);
+
+#[pymethods]
+impl MatchingListener {
+    #[classmethod]
+    fn __class_getitem__(cls: &Bound<PyType>, args: &Bound<PyAny>) -> PyObject {
+        generic(cls, args)
+    }
+
+    fn __enter__<'a, 'py>(this: &'a Bound<'py, Self>) -> &'a Bound<'py, Self> {
+        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 handler(&self, py: Python) -> PyResult<PyObject> {
+        Ok(self.get_ref()?.handler().to_object(py))
+    }
+
+    fn try_recv(&self, py: Python) -> PyResult<PyObject> {
+        self.get_ref()?.handler().try_recv(py)
+    }
+
+    fn recv(&self, py: Python) -> PyResult<PyObject> {
+        self.get_ref()?.handler().recv(py)
+    }
+
+    fn undeclare(&mut self, py: Python) -> PyResult<()> {
+        wait(py, self.take()?.undeclare())
+    }
+
+    fn __iter__<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyIterator>> {
+        self.handler(py)?.bind(py).iter()
+    }
+
+    fn __repr__(&self) -> PyResult<String> {
+        Ok(format!("{:?}", self.get_ref()?))
+    }
+}

src/pubsub.rs

@@ -18,11 +18,13 @@ use pyo3::{
 
 use crate::{
     bytes::{Encoding, ZBytes},
-    handlers::HandlerImpl,
+    handlers::{into_handler, HandlerImpl},
     key_expr::KeyExpr,
     macros::{build, option_wrapper},
+    matching::{MatchingListener, MatchingStatus},
     qos::{CongestionControl, Priority, Reliability},
     sample::Sample,
+    time::Timestamp,
     utils::{generic, wait},
 };
 
@@ -70,26 +72,49 @@ impl Publisher {
         Ok(self.get_ref()?.reliability().into())
     }
 
-    // TODO add timestamp
-    #[pyo3(signature = (payload, *, encoding = None, attachment = None))]
+    #[getter]
+    fn matching_status(&self, py: Python) -> PyResult<MatchingStatus> {
+        Ok(wait(py, self.get_ref()?.matching_status())?.into())
+    }
+
+    #[pyo3(signature = (payload, *, encoding = None, attachment = None, timestamp = None))]
     fn put(
         &self,
         py: Python,
         #[pyo3(from_py_with = "ZBytes::from_py")] payload: ZBytes,
         #[pyo3(from_py_with = "Encoding::from_py_opt")] encoding: Option<Encoding>,
         #[pyo3(from_py_with = "ZBytes::from_py_opt")] attachment: Option<ZBytes>,
+        timestamp: Option<Timestamp>,
     ) -> PyResult<()> {
         let this = self.get_ref()?;
-        wait(py, build!(this.put(payload), encoding, attachment))
+        wait(
+            py,
+            build!(this.put(payload), encoding, attachment, timestamp),
+        )
     }
 
-    #[pyo3(signature = (*, attachment = None))]
+    #[pyo3(signature = (*, attachment = None, timestamp = None))]
     fn delete(
         &self,
         py: Python,
         #[pyo3(from_py_with = "ZBytes::from_py_opt")] attachment: Option<ZBytes>,
+        timestamp: Option<Timestamp>,
     ) -> PyResult<()> {
-        wait(py, build!(self.get_ref()?.delete(), attachment))
+        wait(py, build!(self.get_ref()?.delete(), attachment, timestamp))
+    }
+
+    #[pyo3(signature = (handler = None))]
+    fn declare_matching_listener(
+        &self,
+        py: Python,
+        handler: Option<&Bound<PyAny>>,
+    ) -> PyResult<MatchingListener> {
+        let (handler, background) = into_handler(py, handler)?;
+        let mut listener = wait(py, self.get_ref()?.matching_listener().with(handler))?;
+        if background {
+            listener.set_background(true);
+        }
+        Ok(listener.into())
     }
 
     fn undeclare(&mut self, py: Python) -> PyResult<()> {