NBFTReliability

zenoh-ext/Cargo.toml

@@ -35,13 +35,15 @@ async-std = { version = "=1.12.0", default-features = false, features = [
     "attributes",
     "unstable",
 ] }
+async-trait = "0.1.57"
 bincode = "1.3.3"
 env_logger = "0.10.0"
 flume = "0.10.14"
 futures = "0.3.25"
 log = "0.4.17"
 serde = "1.0.149"
 zenoh = { version = "0.7.0-rc", path = "../zenoh", default-features = false, features = ["unstable"] }
+zenoh-collections = { version = "0.7.0-rc", path = "../commons/zenoh-collections/" }
 zenoh-core = { version = "0.7.0-rc", path = "../commons/zenoh-core/" }
 zenoh-sync = { version = "0.7.0-rc", path = "../commons/zenoh-sync" }
 zenoh-util = { version = "0.7.0-rc", path = "../commons/zenoh-util" }
@@ -64,3 +66,18 @@ path = "examples/z_member.rs"
 [[example]]
 name = "z_view_size"
 path = "examples/z_view_size.rs"
+
+[[example]]
+name = "z_nbftr_sub"
+path = "examples/z_nbftr_sub.rs"
+required-features = ["unstable"]
+
+[[example]]
+name = "z_nbftr_pub"
+path = "examples/z_nbftr_pub.rs"
+required-features = ["unstable"]
+
+[[example]]
+name = "z_nbftr_cache"
+path = "examples/z_nbftr_cache.rs"
+required-features = ["unstable"]

zenoh-ext/examples/z_nbftr_cache.rs

@@ -0,0 +1,115 @@
+//
+// 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::time::Duration;
+use zenoh::config::Config;
+use zenoh::prelude::r#async::*;
+use zenoh_ext::*;
+
+#[async_std::main]
+async fn main() {
+    // Initiate logging
+    env_logger::init();
+
+    let (config, key_expr, history, prefix) = parse_args();
+
+    println!("Opening session...");
+    let session = zenoh::open(config).res().await.unwrap().into_arc();
+
+    println!("Declaring NBFTReliabilityCache on {}", key_expr);
+    let _cache = session
+        .declare_nbftreliability_cache(key_expr)
+        .history(history)
+        .queryable_prefix(prefix)
+        .res()
+        .await
+        .unwrap();
+
+    println!("Enter '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,
+            0 => sleep(Duration::from_secs(1)).await,
+            _ => (),
+        }
+    }
+}
+
+fn parse_args() -> (Config, String, usize, String) {
+    let args = App::new("zenoh-ext non blocking fault tolerant reliability cache 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 to subscribe onto'")
+                .default_value("demo/example/**"),
+        )
+        .arg(Arg::from_usage(
+            "-c, --config=[FILE]      'A configuration file.'",
+        ))
+        .arg(
+            Arg::from_usage("-h, --history=[SIZE] 'The number of publications to keep in cache'")
+                .default_value("1024"),
+        )
+        .arg(
+            Arg::from_usage("-x, --prefix=[STRING] 'The id of publishers to cache'")
+                .default_value("*"),
+        )
+        .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").unwrap().to_string();
+    let history: usize = args.value_of("history").unwrap().parse().unwrap();
+    let prefix = args.value_of("prefix").unwrap().to_string();
+
+    (config, key_expr, history, prefix)
+}

zenoh-ext/examples/z_nbftr_pub.rs

@@ -0,0 +1,113 @@
+//
+// 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 std::time::Duration;
+use zenoh::config::Config;
+use zenoh::prelude::r#async::*;
+use zenoh_ext::*;
+
+#[async_std::main]
+async fn main() {
+    // Initiate logging
+    env_logger::init();
+
+    let (config, key_expr, value, cache, history) = parse_args();
+
+    println!("Opening session...");
+    let session = zenoh::open(config).res().await.unwrap();
+
+    println!("Declaring NBFTReliablePublisher on {}", &key_expr);
+    let publ = session
+        .declare_nbftreliable_publisher(&key_expr)
+        .with_cache(cache)
+        .history(history)
+        .res()
+        .await
+        .unwrap();
+
+    for idx in 0..u32::MAX {
+        sleep(Duration::from_secs(1)).await;
+        let buf = format!("[{:4}] {}", idx, value);
+        println!("Pub Data ('{}': '{}')", &key_expr, buf);
+        publ.put(buf).res().await.unwrap();
+    }
+}
+
+fn parse_args() -> (Config, String, String, bool, usize) {
+    let args = App::new("zenoh-ext non blocking fault tolerant reliable publisher 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 to publish.'")
+                .default_value("demo/example/zenoh-rs-pub"),
+        )
+        .arg(
+            Arg::from_usage("-v, --value=[VALUE]      'The value to publish.'")
+                .default_value("Pub from Rust!"),
+        )
+        .arg(Arg::from_usage(
+            "-n, --no-cache 'Disable local reliability cache'",
+        ))
+        .arg(
+            Arg::from_usage("-h, --history=[SIZE] 'The number of publications to keep in cache'")
+                .default_value("1024"),
+        )
+        .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").unwrap().to_string();
+    let value = args.value_of("value").unwrap().to_string();
+    let cache = !args.is_present("no-cache");
+    let history: usize = args.value_of("history").unwrap().parse().unwrap();
+
+    (config, key_expr, value, cache, history)
+}