Fix and cleanup tokio example using async/await

Signed-off-by: Paul Osborne <[email protected]>

Author: posborne

Cargo.toml

@@ -22,3 +22,10 @@ futures = { version = "0.3", optional = true }
 nix = "0.22"
 mio = { version = "0.7", optional = true, features = ["os-ext"]}
 tokio = { version = "1", optional = true, features = ["net"] }
+
+[dev-dependencies]
+tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
+
+[[example]]
+name = "tokio"
+required-features = ["async-tokio"]

examples/tokio.rs

@@ -1,63 +1,44 @@
-#[cfg(feature = "async-tokio")]
-extern crate futures;
-#[cfg(feature = "async-tokio")]
-extern crate sysfs_gpio;
-#[cfg(feature = "async-tokio")]
-extern crate tokio;
+// Copyright (c) 2020.  The sysfs-gpio Authors.
 
-#[cfg(feature = "async-tokio")]
+use futures::future::join_all;
+use futures::StreamExt;
 use std::env;
-
-#[cfg(feature = "async-tokio")]
-use futures::{lazy, Future, Stream};
-
-#[cfg(feature = "async-tokio")]
 use sysfs_gpio::{Direction, Edge, Pin};
 
-#[cfg(feature = "async-tokio")]
-fn stream(pin_nums: Vec<u64>) -> sysfs_gpio::Result<()> {
+async fn monitor_pin(pin: Pin) -> Result<(), sysfs_gpio::Error> {
+    pin.export()?;
+    pin.set_direction(Direction::In)?;
+    pin.set_edge(Edge::BothEdges)?;
+    let mut gpio_events = pin.get_value_stream()?;
+    while let Some(evt) = gpio_events.next().await {
+        let val = evt.unwrap();
+        println!("Pin {} changed value to {}", pin.get_pin_num(), val);
+    }
+    Ok(())
+}
+
+async fn stream(pin_nums: Vec<u64>) {
     // NOTE: this currently runs forever and as such if
     // the app is stopped (Ctrl-C), no cleanup will happen
     // and the GPIO will be left exported.  Not much
     // can be done about this as Rust signal handling isn't
     // really present at the moment.  Revisit later.
-    let pins: Vec<_> = pin_nums.iter().map(|&p| (p, Pin::new(p))).collect();
-    let task = lazy(move || {
-        for &(i, ref pin) in pins.iter() {
-            pin.export().unwrap();
-            pin.set_direction(Direction::In).unwrap();
-            pin.set_edge(Edge::BothEdges).unwrap();
-            tokio::spawn(
-                pin.get_value_stream()
-                    .unwrap()
-                    .for_each(move |val| {
-                        println!("Pin {} changed value to {}", i, val);
-                        Ok(())
-                    })
-                    .map_err(|_| ()),
-            );
-        }
-        Ok(())
-    });
-    tokio::run(task);
-
-    Ok(())
+    join_all(pin_nums.into_iter().map(|p| {
+        let pin = Pin::new(p);
+        tokio::task::spawn(monitor_pin(pin))
+    }))
+    .await;
 }
 
-#[cfg(feature = "async-tokio")]
-fn main() {
+#[tokio::main]
+async fn main() {
     let pins: Vec<u64> = env::args()
         .skip(1)
         .map(|a| a.parse().expect("Pins must be specified as integers"))
         .collect();
     if pins.is_empty() {
         println!("Usage: ./tokio <pin> [pin ...]");
     } else {
-        stream(pins).unwrap();
+        stream(pins).await;
     }
 }
-
-#[cfg(not(feature = "async-tokio"))]
-fn main() {
-    println!("This example requires the `tokio` feature to be enabled.");
-}