test: make example tests deterministic and isolated

docs/examples/custom_channel.py

@@ -21,15 +21,12 @@
 
 
 # Test support: send data in background
-def send_data():
+def send_data(session: zenoh.Session):
     time.sleep(3)
     for i in range(2):
         session.put("key/expression", f"sample_{i}")
 
 
-threading.Thread(target=send_data, daemon=True).start()
-
-
 # [custom_channel]
 class PriorityChannel:
     def __init__(self, maxsize=100):
@@ -53,9 +50,17 @@ def count(self) -> int:
 with zenoh.open(zenoh.Config()) as session:
     channel = PriorityChannel(maxsize=50)
     subscriber = session.declare_subscriber("key/expression", (channel.send, channel))
+    # Start publishing only after the session and subscriber are ready.
+    sender = threading.Thread(target=send_data, args=(session,))
+    sender.start()
     sample = subscriber.handler.recv()
     print(f">> Received: {sample.payload.to_string()}")
     # [custom_channel_usage]
-    # one sample should remain in the channel
-    time.sleep(1)  # wait a bit for the background sender
+    # One sample should remain in the channel. Wait for asynchronous delivery
+    # instead of relying on a fixed sleep.
+    deadline = time.monotonic() + 5
+    while channel.count() < 1 and time.monotonic() < deadline:
+        time.sleep(0.01)
     assert channel.count() == 1  # verify that one sample is still in the channel
+    sender.join(timeout=5)
+    assert not sender.is_alive()

examples/z_pub_thr.py

@@ -14,7 +14,7 @@
 import zenoh
 
 
-def main(conf: zenoh.Config, payload_size: int):
+def main(conf: zenoh.Config, key: str, payload_size: int):
     # initiate logging
     zenoh.init_log_from_env_or("error")
 
@@ -25,9 +25,7 @@ def main(conf: zenoh.Config, payload_size: int):
     congestion_control = zenoh.CongestionControl.BLOCK
 
     with zenoh.open(conf) as session:
-        pub = session.declare_publisher(
-            "test/thr", congestion_control=congestion_control
-        )
+        pub = session.declare_publisher(key, congestion_control=congestion_control)
 
         print("Press CTRL-C to quit...")
         while True:
@@ -45,11 +43,18 @@ def main(conf: zenoh.Config, payload_size: int):
         prog="z_pub_thr", description="zenoh throughput pub example"
     )
     common.add_config_arguments(parser)
+    parser.add_argument(
+        "--key",
+        "-k",
+        default="test/thr",
+        type=str,
+        help="The key expression to publish onto.",
+    )
     parser.add_argument(
         "payload_size", type=int, help="Sets the size of the payload to publish."
     )
 
     args = parser.parse_args()
     conf = common.get_config_from_args(args)
 
-    main(conf, args.payload_size)
+    main(conf, args.key, args.payload_size)

examples/z_sub_thr.py

@@ -21,7 +21,7 @@
 global_start = None
 
 
-def main(conf: zenoh.Config, number: int):
+def main(conf: zenoh.Config, key: str, number: int):
     def listener(_sample: zenoh.Sample):
         global count, batch_count, start, global_start
         if count == 0:
@@ -49,9 +49,7 @@ def report():
     zenoh.init_log_from_env_or("error")
 
     with zenoh.open(conf) as session:
-        session.declare_subscriber(
-            "test/thr", zenoh.handlers.Callback(listener, report)
-        )
+        session.declare_subscriber(key, zenoh.handlers.Callback(listener, report))
 
         print("Press CTRL-C to quit...")
         while True:
@@ -69,6 +67,7 @@ def report():
         prog="z_sub_thr", description="zenoh throughput sub example"
     )
     common.add_config_arguments(parser)
+    parser.add_argument("--key", "-k", dest="key", default="test/thr", type=str)
     parser.add_argument(
         "--number",
         "-n",
@@ -82,4 +81,4 @@ def report():
     args = parser.parse_args()
     conf = common.get_config_from_args(args)
 
-    main(conf, args.number)
+    main(conf, args.key, args.number)