add matching listener (#557)

* add matching listener;
make querier stable;

* update zenoh

* fix matching status update message for querier example

* move on_matching_status_update to matching listener declaration

Author: DenisBiryukov91

Cargo.lock

@@ -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_pub.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,
+    )

examples/z_querier.py

@@ -21,6 +21,7 @@ mod handlers;
 mod key_expr;
 mod liveliness;
 mod macros;
+mod matching;
 mod pubsub;
 mod qos;
 mod query;
@@ -57,6 +58,7 @@ pub(crate) mod zenoh {
         handlers::Handler,
         key_expr::{KeyExpr, SetIntersectionLevel},
         liveliness::{Liveliness, LivelinessToken},
+        matching::{MatchingListener, MatchingStatus},
         pubsub::{Publisher, Subscriber},
         qos::{CongestionControl, Priority, Reliability},
         query::{