Liveliness & FetchingSubscriber

examples/Cargo.toml

@@ -27,6 +27,7 @@ readme = "README.md"
 
 [features]
 shared-memory = ["zenoh/shared-memory"]
+unstable = ["zenoh/unstable"]
 
 [dependencies]
 async-std = { workspace = true, features = ["attributes"] }
@@ -118,6 +119,21 @@ path = "examples/z_get.rs"
 name = "z_forward"
 path = "examples/z_forward.rs"
 
+[[example]]
+name = "z_liveliness"
+path = "examples/z_liveliness.rs"
+required-features = ["unstable"]
+
+[[example]]
+name = "z_sub_liveliness"
+path = "examples/z_sub_liveliness.rs"
+required-features = ["unstable"]
+
+[[example]]
+name = "z_get_liveliness"
+path = "examples/z_get_liveliness.rs"
+required-features = ["unstable"]
+
 [[example]]
 name = "z_pub_thr"
 path = "examples/z_pub_thr.rs"

examples/README.md

@@ -204,3 +204,51 @@
    ```bash
       z_pub_shm_thr
    ```
+
+### z_liveliness
+
+   Declares a liveliness token on a given key expression (`group1/zenoh-rs` by default).
+   This token will be seen alive byt the `z_get_liveliness` and `z_sub_liveliness` until
+   user explicitely drops the token by pressing `'d'` or implicitely dropped by terminating
+   or killing the `z_liveliness` example.
+
+   Typical usage:
+   ```bash
+      z_liveliness
+   ```
+   or
+   ```bash
+      z_liveliness -k group1/member1
+   ```
+
+### z_get_liveliness
+
+   Queries all the currently alive liveliness tokens that match a given key expression
+   (`group1/**` by default). Those tokens could be declared by the `z_liveliness` example.
+
+   Typical usage:
+   ```bash
+      z_get_liveliness
+   ```
+   or
+   ```bash
+      z_get_liveliness -k group1/**
+   ```
+
+### z_sub_liveliness
+
+   Subscribe to all liveliness changes (liveliness tokens getting alive or
+   liveliness tokens being dropped) that match a given key expression
+   (`group1/**` by default). Those tokens could be declared by the `z_liveliness`
+   example.
+   Note: the `z_sub_liveliness` example will not receive informations about
+   matching liveliness tokens that were alive before it's start.
+
+   Typical usage:
+   ```bash
+      z_sub_liveliness
+   ```
+   or
+   ```bash
+      z_sub_liveliness -k group1/**
+   ```

examples/examples/z_get_liveliness.rs

@@ -0,0 +1,103 @@
+//
+// 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, <[email protected]>
+//
+use clap::{App, Arg};
+use std::convert::TryFrom;
+use std::time::Duration;
+use zenoh::config::Config;
+use zenoh::prelude::r#async::*;
+
+#[async_std::main]
+async fn main() {
+    // initiate logging
+    env_logger::init();
+
+    let (config, key_expr, timeout) = parse_args();
+
+    println!("Opening session...");
+    let session = zenoh::open(config).res().await.unwrap();
+
+    println!("Sending Liveliness Query '{key_expr}'...");
+    let replies = session
+        .liveliness()
+        .get(&key_expr)
+        .timeout(timeout)
+        .res()
+        .await
+        .unwrap();
+    while let Ok(reply) = replies.recv_async().await {
+        match reply.sample {
+            Ok(sample) => println!(">> Alive token ('{}')", sample.key_expr.as_str(),),
+            Err(err) => println!(">> Received (ERROR: '{}')", String::try_from(&err).unwrap()),
+        }
+    }
+}
+
+fn parse_args() -> (Config, String, Duration) {
+    let args = App::new("zenoh liveliness query example")
+        .arg(
+            Arg::from_usage("-m, --mode=[MODE]  'The zenoh session mode (peer by default).")
+                .possible_values(["peer", "client"]),
+        )
+        .arg(Arg::from_usage(
+            "-e, --connect=[ENDPOINT]...   'Endpoints to connect to.'",
+        ))
+        .arg(Arg::from_usage(
+            "-l, --listen=[ENDPOINT]...   'Endpoints to listen on.'",
+        ))
+        .arg(
+            Arg::from_usage("-k, --key_expr=[KEYEXPR] 'The key expression matching liveliness tokens to query.'")
+                .default_value("group1/**"),
+        )
+        .arg(
+            Arg::from_usage("-o, --timeout=[TIME] 'The query timeout in milliseconds'")
+                .default_value("10000"),
+        )
+        .arg(Arg::from_usage(
+            "-c, --config=[FILE]      'A configuration file.'",
+        ))
+        .arg(Arg::from_usage(
+            "--no-multicast-scouting 'Disable the multicast-based scouting mechanism.'",
+        ))
+        .get_matches();
+
+    let mut config = if let Some(conf_file) = args.value_of("config") {
+        Config::from_file(conf_file).unwrap()
+    } else {
+        Config::default()
+    };
+    if let Some(Ok(mode)) = args.value_of("mode").map(|mode| mode.parse()) {
+        config.set_mode(Some(mode)).unwrap();
+    }
+    if let Some(values) = args.values_of("connect") {
+        config
+            .connect
+            .endpoints
+            .extend(values.map(|v| v.parse().unwrap()))
+    }
+    if let Some(values) = args.values_of("listen") {
+        config
+            .listen
+            .endpoints
+            .extend(values.map(|v| v.parse().unwrap()))
+    }
+    if args.is_present("no-multicast-scouting") {
+        config.scouting.multicast.set_enabled(Some(false)).unwrap();
+    }
+
+    let key_expr = args.value_of("key_expr").unwrap().to_string();
+
+    let timeout = Duration::from_millis(args.value_of("timeout").unwrap().parse::<u64>().unwrap());
+
+    (config, key_expr, timeout)
+}

examples/examples/z_liveliness.rs

@@ -0,0 +1,114 @@
+//
+// 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, <[email protected]>
+//
+use async_std::task::sleep;
+use clap::{App, Arg};
+use futures::prelude::*;
+use std::convert::TryFrom;
+use std::time::Duration;
+use zenoh::config::Config;
+use zenoh::prelude::r#async::*;
+
+#[async_std::main]
+async fn main() {
+    // Initiate logging
+    env_logger::init();
+
+    let (config, key_expr) = parse_args();
+
+    println!("Opening session...");
+    let session = zenoh::open(config).res().await.unwrap();
+
+    println!("Declaring LivelinessToken on '{}'...", &key_expr);
+    let mut token = Some(
+        session
+            .liveliness()
+            .declare_token(&key_expr)
+            .res()
+            .await
+            .unwrap(),
+    );
+
+    println!("Enter 'd' to undeclare LivelinessToken, 'q' to quit...");
+    let mut stdin = async_std::io::stdin();
+    let mut input = [0_u8];
+    loop {
+        let _ = stdin.read_exact(&mut input).await;
+        match input[0] {
+            b'q' => break,
+            b'd' => {
+                if let Some(token) = token.take() {
+                    println!("Undeclaring LivelinessToken...");
+                    token.undeclare().res().await.unwrap();
+                }
+            }
+            0 => sleep(Duration::from_secs(1)).await,
+            _ => (),
+        }
+    }
+}
+
+fn parse_args() -> (Config, KeyExpr<'static>) {
+    let args = App::new("zenoh liveliness example")
+        .arg(
+            Arg::from_usage("-m, --mode=[MODE]  'The zenoh session mode (peer by default).")
+                .possible_values(["peer", "client"]),
+        )
+        .arg(Arg::from_usage(
+            "-e, --connect=[ENDPOINT]...   'Endpoints to connect to.'",
+        ))
+        .arg(Arg::from_usage(
+            "-l, --listen=[ENDPOINT]...   'Endpoints to listen on.'",
+        ))
+        .arg(
+            Arg::from_usage("-k, --key=[KEYEXPR] 'The key expression of the liveliness token.'")
+                .default_value("group1/zenoh-rs"),
+        )
+        .arg(Arg::from_usage(
+            "-c, --config=[FILE]      'A configuration file.'",
+        ))
+        .arg(Arg::from_usage(
+            "--no-multicast-scouting 'Disable the multicast-based scouting mechanism.'",
+        ))
+        .get_matches();
+
+    let mut config = if let Some(conf_file) = args.value_of("config") {
+        Config::from_file(conf_file).unwrap()
+    } else {
+        Config::default()
+    };
+    if let Some(Ok(mode)) = args.value_of("mode").map(|mode| mode.parse()) {
+        config.set_mode(Some(mode)).unwrap();
+    }
+    if let Some(values) = args.values_of("connect") {
+        config
+            .connect
+            .endpoints
+            .extend(values.map(|v| v.parse().unwrap()))
+    }
+    if let Some(values) = args.values_of("listen") {
+        config
+            .listen
+            .endpoints
+            .extend(values.map(|v| v.parse().unwrap()))
+    }
+    if args.is_present("no-multicast-scouting") {
+        config.scouting.multicast.set_enabled(Some(false)).unwrap();
+    }
+
+    let key_expr = KeyExpr::try_from(args.value_of("key").unwrap())
+        .unwrap()
+        .into_owned();
+
+    (config, key_expr)
+}