add examples

Author: wyfo

examples/z_advanced_pub.py

@@ -0,0 +1,79 @@
+#
+# Copyright (c) 2022 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>
+#
+import time
+from typing import Optional
+
+import zenoh
+from zenoh.ext import CacheConfig, MissDetectionConfig, declare_advanced_publisher
+
+
+def main(conf: zenoh.Config, key: str, payload: str, history: int):
+    # initiate logging
+    zenoh.init_log_from_env_or("error")
+
+    print("Opening session...")
+    with zenoh.open(conf) as session:
+        print(f"Declaring AdvancedPublisher on '{key}'...")
+        pub = declare_advanced_publisher(
+            session,
+            key,
+            cache=CacheConfig(max_samples=history),
+            sample_miss_detection=MissDetectionConfig(heartbeat=5),
+            publisher_detection=True,
+        )
+
+        print("Press CTRL-C to quit...")
+        for idx in itertools.count():
+            buf = f"[{idx:4d}] {payload}"
+            print(f"Putting Data ('{key}': '{buf}')...")
+            pub.put(buf)
+
+
+# --- Command line argument parsing --- --- --- --- --- ---
+if __name__ == "__main__":
+    import argparse
+    import itertools
+
+    import common
+
+    parser = argparse.ArgumentParser(prog="z_pub", description="zenoh pub example")
+    common.add_config_arguments(parser)
+    parser.add_argument(
+        "--key",
+        "-k",
+        dest="key",
+        default="demo/example/zenoh-python-pub",
+        type=str,
+        help="The key expression to publish onto.",
+    )
+    parser.add_argument(
+        "--payload",
+        "-p",
+        dest="payload",
+        default="Pub from Python!",
+        type=str,
+        help="The payload to publish.",
+    )
+    parser.add_argument(
+        "--history",
+        dest="history",
+        type=int,
+        default=1,
+        help="The number of publications to keep in cache",
+    )
+
+    args = parser.parse_args()
+    conf = common.get_config_from_args(args)
+
+    main(conf, args.key, args.payload, args.history, args.iter, args.interval)

examples/z_advanced_sub.py

@@ -0,0 +1,72 @@
+#
+# Copyright (c) 2022 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>
+#
+import time
+
+import zenoh
+from zenoh.ext import HistoryConfig, Miss, RecoveryConfig, declare_advanced_subscriber
+
+
+def main(conf: zenoh.Config, key: str):
+    # initiate logging
+    zenoh.init_log_from_env_or("error")
+
+    print("Opening session...")
+    with zenoh.open(conf) as session:
+        print(f"Declaring Subscriber on '{key}'...")
+
+        def listener(sample: zenoh.Sample):
+            print(
+                f">> [Subscriber] Received {sample.kind} ('{sample.key_expr}': '{sample.payload.to_string()}')"
+            )
+
+        advanced_sub = declare_advanced_subscriber(
+            session,
+            key,
+            listener,
+            history=HistoryConfig(detect_late_publishers=True),
+            recovery=RecoveryConfig(heartbeat=True),
+            subscriber_detection=True,
+        )
+
+        def miss_listener(miss: Miss):
+            print(f">> [Subscriber] Missed {miss.nb} samples from {miss.source} !!!")
+
+        advanced_sub.sample_miss_listener(miss_listener)
+
+        print("Press CTRL-C to quit...")
+        while True:
+            time.sleep(1)
+
+
+# --- Command line argument parsing --- --- --- --- --- ---
+if __name__ == "__main__":
+    import argparse
+
+    import common
+
+    parser = argparse.ArgumentParser(prog="z_sub", description="zenoh sub example")
+    common.add_config_arguments(parser)
+    parser.add_argument(
+        "--key",
+        "-k",
+        dest="key",
+        default="demo/example/**",
+        type=str,
+        help="The key expression to subscribe to.",
+    )
+
+    args = parser.parse_args()
+    conf = common.get_config_from_args(args)
+
+    main(conf, args.key)

src/ext.rs

@@ -18,7 +18,7 @@ use crate::{
     bytes::{Encoding, ZBytes},
     handlers::{into_handler, HandlerImpl},
     key_expr::KeyExpr,
-    macros::{build, downcast_or_new, import, option_wrapper, py_static, try_import, wrapper},
+    macros::{build, import, option_wrapper, py_static, try_import, wrapper},
     pubsub::Subscriber,
     qos::{CongestionControl, Priority, Reliability},
     sample::Sample,
@@ -594,7 +594,6 @@ impl AdvancedSubscriber {
 }
 
 wrapper!(zenoh_ext::CacheConfig: Clone);
-downcast_or_new!(CacheConfig => Option<usize>, None);
 
 #[pymethods]
 impl CacheConfig {
@@ -634,12 +633,12 @@ wrapper!(zenoh_ext::Miss);
 impl Miss {
     #[getter]
     fn source(&self) -> EntityGlobalId {
-        self.source().into()
+        self.0.source().into()
     }
 
     #[getter]
-    fn nb(&self) -> usize {
-        self.nb()
+    fn nb(&self) -> u32 {
+        self.0.nb()
     }
 }
 
@@ -745,7 +744,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))]
-fn declare_publisher(
+pub(crate) fn declare_advanced_publisher(
     py: Python,
     session: &Session,
     #[pyo3(from_py_with = "KeyExpr::from_py")] key_expr: KeyExpr,
@@ -777,7 +776,7 @@ fn declare_publisher(
 #[allow(clippy::too_many_arguments)]
 #[pyfunction]
 #[pyo3(signature = (session, key_expr, handler = None, *, history = None, recovery = None, subscriber_detection = None))]
-fn declare_subscriber(
+pub(crate) fn declare_advanced_subscriber(
     session: &Session,
     py: Python,
     #[pyo3(from_py_with = "KeyExpr::from_py")] key_expr: KeyExpr,

src/lib.rs

@@ -80,7 +80,11 @@ pub(crate) mod zenoh {
     #[pymodule]
     mod _ext {
         #[pymodule_export]
-        use crate::ext::{z_deserialize, z_serialize};
+        use crate::ext::{
+            declare_advanced_publisher, declare_advanced_subscriber, z_deserialize, z_serialize,
+            AdvancedPublisher, AdvancedSubscriber, CacheConfig, HistoryConfig, Miss,
+            MissDetectionConfig, RecoveryConfig, RepliesConfig, SampleMissListener,
+        };
         #[pymodule_export]
         use crate::ZDeserializeError;
     }

src/session.rs

@@ -313,5 +313,7 @@ impl EntityGlobalId {
         self.0.eid()
     }
 
-    // TODO __repr__
+    fn __repr__(&self) -> String {
+        format!("{:?}", self.0)
+    }
 }

zenoh/ext.pyi

@@ -304,7 +304,7 @@ def declare_advanced_publisher(
     priority: Priority | None = None,
     express: bool | None = None,
     reliability: Reliability | None = None,
-    cache: CacheConfig | int | None = None,
+    cache: CacheConfig | None = None,
     sample_miss_detection: MissDetectionConfig | None = None,
     publisher_detection: bool | None = None,
 ) -> AdvancedPublisher: