Align examples and remove reading from stdin (eclipse-zenoh#768)

* Remove reading from stdin, align example implementations * Change counter start to 0 * Replace sleep with thread-parking * Remove select, format files * Uncomment liveliness token undeclaration
fuzzypixelz · Mar 13, 2024 · 61838d5 · 61838d5
1 parent c75401b
commit 61838d5
Show file tree

Hide file tree

Showing 14 changed files with 74 additions and 192 deletions.
diff --git a/examples/examples/z_liveliness.rs b/examples/examples/z_liveliness.rs
@@ -11,10 +11,7 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
-use async_std::task::sleep;
 use clap::Parser;
-use futures::prelude::*;
-use std::time::Duration;
 use zenoh::config::Config;
 use zenoh::prelude::r#async::*;
 use zenoh_examples::CommonArgs;
@@ -39,23 +36,14 @@ async fn main() {
             .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,
-            _ => (),
-        }
-    }
+    println!("Press CTRL-C to undeclare LivelinessToken and quit...");
+    std::thread::park();
+    // LivelinessTokens are automatically closed when dropped
+    // Use the code below to manually undeclare it if needed
+    if let Some(token) = token.take() {
+        println!("Undeclaring LivelinessToken...");
+        token.undeclare().res().await.unwrap();
+    };
 }
 
 #[derive(clap::Parser, Clone, PartialEq, Eq, Hash, Debug)]

diff --git a/examples/examples/z_pong.rs b/examples/examples/z_pong.rs
@@ -1,5 +1,3 @@
-use std::io::{stdin, Read};
-
 //
 // Copyright (c) 2023 ZettaScale Technology
 //
@@ -44,7 +42,7 @@ fn main() {
         .callback(move |sample| publisher.put(sample.value).res().unwrap())
         .res()
         .unwrap();
-    for _ in stdin().bytes().take_while(|b| !matches!(b, Ok(b'q'))) {}
+    std::thread::park();
 }
 
 #[derive(clap::Parser, Clone, PartialEq, Eq, Hash, Debug)]

diff --git a/examples/examples/z_pub.rs b/examples/examples/z_pub.rs
@@ -31,6 +31,7 @@ async fn main() {
     println!("Declaring Publisher on '{key_expr}'...");
     let publisher = session.declare_publisher(&key_expr).res().await.unwrap();
 
+    println!("Press CTRL-C to quit...");
     for idx in 0..u32::MAX {
         sleep(Duration::from_secs(1)).await;
         let buf = format!("[{idx:4}] {value}");

diff --git a/examples/examples/z_pub_shm.rs b/examples/examples/z_pub_shm.rs
@@ -44,6 +44,7 @@ async fn main() -> Result<(), zenoh::Error> {
     println!("Allocating Shared Memory Buffer...");
     let publisher = session.declare_publisher(&path).res().await.unwrap();
 
+    println!("Press CTRL-C to quit...");
     for idx in 0..(K * N as u32) {
         sleep(Duration::from_secs(1)).await;
         let mut sbuf = match shm.alloc(1024) {

diff --git a/examples/examples/z_pub_shm_thr.rs b/examples/examples/z_pub_shm_thr.rs
@@ -42,6 +42,7 @@ async fn main() {
     // Make sure to not drop messages because of congestion control
     .congestion_control(CongestionControl::Block).res().await.unwrap();
 
+    println!("Press CTRL-C to quit...");
     loop {
         publisher.put(buf.clone()).res().await.unwrap();
     }

diff --git a/examples/examples/z_pub_thr.rs b/examples/examples/z_pub_thr.rs
@@ -44,6 +44,7 @@ fn main() {
         .res()
         .unwrap();
 
+    println!("Press CTRL-C to quit...");
     let mut count: usize = 0;
     let mut start = std::time::Instant::now();
     loop {

diff --git a/examples/examples/z_pull.rs b/examples/examples/z_pull.rs
@@ -11,10 +11,8 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
-use async_std::prelude::FutureExt;
 use async_std::task::sleep;
 use clap::Parser;
-use futures::prelude::*;
 use std::time::Duration;
 use zenoh::config::Config;
 use zenoh::prelude::r#async::*;
@@ -35,40 +33,24 @@ async fn main() {
     let subscriber = session
         .declare_subscriber(&key_expr)
         .pull_mode()
-        .res()
-        .await
-        .unwrap();
-
-    println!("Press <enter> to pull data...");
-
-    // Define the future to handle incoming samples of the subscription.
-    let subs = async {
-        while let Ok(sample) = subscriber.recv_async().await {
+        .callback(|sample| {
             println!(
                 ">> [Subscriber] Received {} ('{}': '{}')",
                 sample.kind,
                 sample.key_expr.as_str(),
                 sample.value,
             );
-        }
-    };
-
-    // Define the future to handle keyboard's input.
-    let keyb = async {
-        let mut stdin = async_std::io::stdin();
-        let mut input = [0_u8];
-        loop {
-            stdin.read_exact(&mut input).await.unwrap();
-            match input[0] {
-                b'q' => break,
-                0 => sleep(Duration::from_secs(1)).await,
-                _ => subscriber.pull().res().await.unwrap(),
-            }
-        }
-    };
+        })
+        .res()
+        .await
+        .unwrap();
 
-    // Execute both futures concurrently until one of them returns.
-    subs.race(keyb).await;
+    println!("Press CTRL-C to quit...");
+    for idx in 0..u32::MAX {
+        sleep(Duration::from_secs(1)).await;
+        println!("[{idx:4}] Pulling...");
+        subscriber.pull().res().await.unwrap();
+    }
 }
 
 #[derive(clap::Parser, Clone, PartialEq, Eq, Hash, Debug)]

diff --git a/examples/examples/z_queryable.rs b/examples/examples/z_queryable.rs
@@ -11,12 +11,7 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
-use async_std::task::sleep;
 use clap::Parser;
-use futures::prelude::*;
-use futures::select;
-use std::sync::atomic::Ordering::Relaxed;
-use std::time::Duration;
 use zenoh::config::Config;
 use zenoh::prelude::r#async::*;
 use zenoh_examples::CommonArgs;
@@ -27,7 +22,6 @@ async fn main() {
     env_logger::init();
 
     let (config, key_expr, value, complete) = parse_args();
-    let send_errors = std::sync::atomic::AtomicBool::new(false);
 
     println!("Opening session...");
     let session = zenoh::open(config).res().await.unwrap();
@@ -40,47 +34,27 @@ async fn main() {
         .await
         .unwrap();
 
-    println!("Enter 'q' to quit, 'e' to reply an error to next query...");
-    let mut stdin = async_std::io::stdin();
-    let mut input = [0_u8];
-    loop {
-        select!(
-            query = queryable.recv_async() => {
-                let query = query.unwrap();
-                match query.value() {
-                    None => println!(">> [Queryable ] Received Query '{}'", query.selector()),
-                    Some(value) => println!(">> [Queryable ] Received Query '{}' with value '{}'", query.selector(), value),
-                }
-                let reply = if send_errors.swap(false, Relaxed) {
-                    println!(
-                        ">> [Queryable ] Replying (ERROR: '{}')",
-                        value,
-                    );
-                    Err(value.clone().into())
-                } else {
-                    println!(
-                        ">> [Queryable ] Responding ('{}': '{}')",
-                        key_expr.as_str(),
-                        value,
-                    );
-                    Ok(Sample::new(key_expr.clone(), value.clone()))
-                };
-                query
-                    .reply(reply)
-                    .res()
-                    .await
-                    .unwrap_or_else(|e| println!(">> [Queryable ] Error sending reply: {e}"));
-            },
-
-            _ = stdin.read_exact(&mut input).fuse() => {
-                match input[0] {
-                    b'q' => break,
-                    0 => sleep(Duration::from_secs(1)).await,
-                    b'e' => send_errors.store(true, Relaxed),
-                    _ => (),
-                }
-            }
+    println!("Press CTRL-C to quit...");
+    while let Ok(query) = queryable.recv_async().await {
+        match query.value() {
+            None => println!(">> [Queryable ] Received Query '{}'", query.selector()),
+            Some(value) => println!(
+                ">> [Queryable ] Received Query '{}' with value '{}'",
+                query.selector(),
+                value
+            ),
+        }
+        println!(
+            ">> [Queryable ] Responding ('{}': '{}')",
+            key_expr.as_str(),
+            value,
         );
+        let reply = Ok(Sample::new(key_expr.clone(), value.clone()));
+        query
+            .reply(reply)
+            .res()
+            .await
+            .unwrap_or_else(|e| println!(">> [Queryable ] Error sending reply: {e}"));
     }
 }
 

diff --git a/examples/examples/z_storage.rs b/examples/examples/z_storage.rs
@@ -13,12 +13,9 @@
 //
 #![recursion_limit = "256"]
 
-use async_std::task::sleep;
 use clap::Parser;
-use futures::prelude::*;
 use futures::select;
 use std::collections::HashMap;
-use std::time::Duration;
 use zenoh::config::Config;
 use zenoh::prelude::r#async::*;
 use zenoh_examples::CommonArgs;
@@ -46,9 +43,7 @@ async fn main() {
         .await
         .unwrap();
 
-    println!("Enter 'q' to quit...");
-    let mut stdin = async_std::io::stdin();
-    let mut input = [0u8];
+    println!("Press CTRL-C to quit...");
     loop {
         select!(
             sample = subscriber.recv_async() => {
@@ -70,14 +65,6 @@ async fn main() {
                         query.reply(Ok(sample.clone())).res().await.unwrap();
                     }
                 }
-            },
-
-            _ = stdin.read_exact(&mut input).fuse() => {
-                match input[0] {
-                    b'q' => break,
-                    0 => sleep(Duration::from_secs(1)).await,
-                    _ => (),
-                }
             }
         );
     }

diff --git a/examples/examples/z_sub.rs b/examples/examples/z_sub.rs
@@ -11,11 +11,7 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
-use async_std::task::sleep;
 use clap::Parser;
-use futures::prelude::*;
-use futures::select;
-use std::time::Duration;
 use zenoh::config::Config;
 use zenoh::prelude::r#async::*;
 use zenoh_examples::CommonArgs;
@@ -39,24 +35,13 @@ async fn main() {
 
     let subscriber = session.declare_subscriber(&key_expr).res().await.unwrap();
 
-    println!("Enter 'q' to quit...");
-    let mut stdin = async_std::io::stdin();
-    let mut input = [0_u8];
-    loop {
-        select!(
-            sample = subscriber.recv_async() => {
-                let sample = sample.unwrap();
-                println!(">> [Subscriber] Received {} ('{}': '{}')",
-                    sample.kind, sample.key_expr.as_str(), sample.value);
-            },
-
-            _ = stdin.read_exact(&mut input).fuse() => {
-                match input[0] {
-                    b'q' => break,
-                    0 => sleep(Duration::from_secs(1)).await,
-                    _ => (),
-                }
-            }
+    println!("Press CTRL-C to quit...");
+    while let Ok(sample) = subscriber.recv_async().await {
+        println!(
+            ">> [Subscriber] Received {} ('{}': '{}')",
+            sample.kind,
+            sample.key_expr.as_str(),
+            sample.value
         );
     }
 }

diff --git a/examples/examples/z_sub_liveliness.rs b/examples/examples/z_sub_liveliness.rs
@@ -11,11 +11,7 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
-use async_std::task::sleep;
 use clap::Parser;
-use futures::prelude::*;
-use futures::select;
-use std::time::Duration;
 use zenoh::config::Config;
 use zenoh::prelude::r#async::*;
 use zenoh_examples::CommonArgs;
@@ -39,31 +35,18 @@ async fn main() {
         .await
         .unwrap();
 
-    println!("Enter 'q' to quit...");
-    let mut stdin = async_std::io::stdin();
-    let mut input = [0_u8];
-    loop {
-        select!(
-            sample = subscriber.recv_async() => {
-                let sample = sample.unwrap();
-                match sample.kind {
-                    SampleKind::Put => println!(
-                        ">> [LivelinessSubscriber] New alive token ('{}')",
-                        sample.key_expr.as_str()),
-                    SampleKind::Delete => println!(
-                        ">> [LivelinessSubscriber] Dropped token ('{}')",
-                        sample.key_expr.as_str()),
-                }
-            },
-
-            _ = stdin.read_exact(&mut input).fuse() => {
-                match input[0] {
-                    b'q' => break,
-                    0 => sleep(Duration::from_secs(1)).await,
-                    _ => (),
-                }
-            }
-        );
+    println!("Press CTRL-C to quit...");
+    while let Ok(sample) = subscriber.recv_async().await {
+        match sample.kind {
+            SampleKind::Put => println!(
+                ">> [LivelinessSubscriber] New alive token ('{}')",
+                sample.key_expr.as_str()
+            ),
+            SampleKind::Delete => println!(
+                ">> [LivelinessSubscriber] Dropped token ('{}')",
+                sample.key_expr.as_str()
+            ),
+        }
     }
 }