implement advanced pub/sub (#537)

* implement advanced pub/sub

* add examples

* Merge branch 'main' into advanced-pubsub

* fix examples and add test

* format fix

* format fix

* fix docs for __new__

---------

Co-authored-by: Denis Biryukov <denis.biryukov@zettascale.tech>
Co-authored-by: DenisBiryukov91 <155981813+DenisBiryukov91@users.noreply.github.com>

Author: wyfo

Cargo.toml

@@ -35,6 +35,7 @@ crate-type = ["cdylib"]
 
 [features]
 default = ["zenoh/default", "zenoh-ext"]
+zenoh-ext = ["dep:zenoh-ext", "zenoh-ext/unstable", "zenoh-ext/internal"]
 
 [badges]
 maintenance = { status = "actively-developed" }

docs/stubs_to_sources.py

@@ -28,6 +28,7 @@
 
 PACKAGE = (Path(__file__) / "../../zenoh").resolve()
 __INIT__ = PACKAGE / "__init__.py"
+EXT = PACKAGE / "ext.py"
 
 
 def _unstable(item):
@@ -98,23 +99,26 @@ def visit_FunctionDef(self, node: ast.FunctionDef):
 
 
 def main():
-    # remove __init__.pyi
-    __INIT__.unlink()
+    fnames = [__INIT__, EXT]
+    for fname in fnames:
+        # remove *.py
+        fname.unlink()
     # rename stubs
     for entry in PACKAGE.glob("*.pyi"):
         entry.rename(PACKAGE / f"{entry.stem}.py")
-    # read stub code
-    with open(__INIT__) as f:
-        stub: ast.Module = ast.parse(f.read())
-        # replace _unstable
-        for i, stmt in enumerate(stub.body):
-            if isinstance(stmt, ast.FunctionDef) and stmt.name == "_unstable":
-                stub.body[i] = ast.parse(inspect.getsource(_unstable))
-        # remove overload
-        stub = RemoveOverload().visit(stub)
-    # write modified code
-    with open(__INIT__, "w") as f:
-        f.write(ast.unparse(stub))
+    for fname in fnames:
+        # read stub code
+        with open(fname) as f:
+            stub: ast.Module = ast.parse(f.read())
+            # replace _unstable
+            for i, stmt in enumerate(stub.body):
+                if isinstance(stmt, ast.FunctionDef) and stmt.name == "_unstable":
+                    stub.body[i] = ast.parse(inspect.getsource(_unstable))
+            # remove overload
+            stub = RemoveOverload().visit(stub)
+        # write modified code
+        with open(fname, "w") as f:
+            f.write(ast.unparse(stub))
 
 
 if __name__ == "__main__":

examples/z_advanced_pub.py

@@ -0,0 +1,82 @@
+#
+# 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():
+            time.sleep(1)
+            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_advanced_pub", description="zenoh advanced 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)

examples/z_advanced_sub.py

@@ -0,0 +1,74 @@
+#
+# 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_advanced_sub", description="zenoh advanced 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/bytes.rs

@@ -97,8 +97,8 @@ downcast_or_new!(Encoding => Option<String>);
 #[pymethods]
 impl Encoding {
     #[new]
-    fn new(s: Option<String>) -> PyResult<Self> {
-        Ok(s.map_into().map(Self).unwrap_or_default())
+    fn new(s: Option<String>) -> Self {
+        s.map_into().map(Self).unwrap_or_default()
     }
 
     fn with_schema(&self, schema: String) -> Self {