RFC: Line event stream support for Tokio

.travis.yml

@@ -72,7 +72,7 @@ matrix:
 
     # MSRV
     - env: TARGET=x86_64-unknown-linux-gnu
-      rust: 1.34.0
+      rust: 1.36.0
       if: (branch = staging OR branch = trying) OR (type = pull_request AND branch = master)
 
 before_install:

Cargo.toml

@@ -9,11 +9,23 @@ readme = "README.md"
 categories = ["embedded", "hardware-support", "os", "os::unix-apis"]
 keywords = ["linux", "gpio", "gpiochip", "embedded"]
 license = "MIT OR Apache-2.0"
+edition = "2018"
+
+[features]
+default = []
+async-tokio = ["tokio", "futures", "mio"]
+
+[[example]]
+name = "async_tokio"
+required-features = ["async-tokio"]
 
 [dependencies]
 bitflags = "1.0"
 libc = "0.2"
 nix = "0.14"
+tokio = { version = "0.2", features = ["io-driver", "rt-threaded", "macros"], optional = true }
+futures = { version = "0.3", optional = true }
+mio = { version = "0.6", optional = true }
 
 [dev-dependencies]
 quicli = "0.2"

README.md

@@ -175,7 +175,7 @@ to be considered reliable.
 
 ## Minimum Supported Rust Version (MSRV)
 
-This crate is guaranteed to compile on stable Rust 1.34.0 and up. It *might*
+This crate is guaranteed to compile on stable Rust 1.36.0 and up. It *might*
 compile with older versions but that may change in any new patch release.
 
 ## License

examples/async_tokio.rs

@@ -0,0 +1,44 @@
+// Copyright (c) 2018 The rust-gpio-cdev Project Developers.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use futures::stream::StreamExt;
+use gpio_cdev::*;
+use quicli::prelude::*;
+
+#[derive(Debug, StructOpt)]
+struct Cli {
+    /// The gpiochip device (e.g. /dev/gpiochip0)
+    chip: String,
+    /// The offset of the GPIO line for the provided chip
+    line: u32,
+}
+
+async fn do_main(args: Cli) -> std::result::Result<(), errors::Error> {
+    let mut chip = Chip::new(args.chip)?;
+    let line = chip.get_line(args.line)?;
+    let mut events = AsyncLineEventHandle::new(line.events(
+        LineRequestFlags::INPUT,
+        EventRequestFlags::BOTH_EDGES,
+        "gpioevents",
+    )?)?;
+
+    loop {
+        match events.next().await {
+            Some(event) => println!("{:?}", event?),
+            None => break,
+        };
+    }
+
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() {
+    let args = Cli::from_args();
+    do_main(args).await.unwrap();
+}

src/async_tokio.rs

@@ -0,0 +1,142 @@
+// Copyright (c) 2018 The rust-gpio-cdev Project Developers.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+//! Wrapper for asynchronous programming using Tokio.
+
+use futures::ready;
+use futures::stream::Stream;
+use futures::task::{Context, Poll};
+use mio::event::Evented;
+use mio::unix::EventedFd;
+use mio::{PollOpt, Ready, Token};
+use tokio::io::PollEvented;
+
+use std::io;
+use std::os::unix::io::AsRawFd;
+use std::pin::Pin;
+
+use super::errors::event_err;
+use super::{LineEvent, LineEventHandle, Result};
+
+struct PollWrapper {
+    handle: LineEventHandle,
+}
+
+impl Evented for PollWrapper {
+    fn register(
+        &self,
+        poll: &mio::Poll,
+        token: Token,
+        interest: Ready,
+        opts: PollOpt,
+    ) -> io::Result<()> {
+        EventedFd(&self.handle.file.as_raw_fd()).register(poll, token, interest, opts)
+    }
+
+    fn reregister(
+        &self,
+        poll: &mio::Poll,
+        token: Token,
+        interest: Ready,
+        opts: PollOpt,
+    ) -> io::Result<()> {
+        EventedFd(&self.handle.file.as_raw_fd()).reregister(poll, token, interest, opts)
+    }
+
+    fn deregister(&self, poll: &mio::Poll) -> io::Result<()> {
+        EventedFd(&self.handle.file.as_raw_fd()).deregister(poll)
+    }
+}
+
+/// Wrapper around a `LineEventHandle` which implements a `futures::stream::Stream` for interrupts.
+///
+/// # Example
+///
+/// The following example waits for state changes on an input line.
+///
+/// ```no_run
+/// # type Result<T> = std::result::Result<T, gpio_cdev::errors::Error>;
+/// use futures::stream::StreamExt;
+/// use gpio_cdev::{AsyncLineEventHandle, Chip, EventRequestFlags, LineRequestFlags};
+///
+/// async fn print_events(line: u32) -> Result<()> {
+///     let mut chip = Chip::new("/dev/gpiochip0")?;
+///     let line = chip.get_line(line)?;
+///     let mut events = AsyncLineEventHandle::new(line.events(
+///         LineRequestFlags::INPUT,
+///         EventRequestFlags::BOTH_EDGES,
+///         "gpioevents",
+///     )?)?;
+///
+///     loop {
+///         match events.next().await {
+///             Some(event) => println!("{:?}", event?),
+///             None => break,
+///         };
+///     }
+///
+///     Ok(())
+/// }
+///
+/// # #[tokio::main]
+/// # async fn main() {
+/// #     print_events(42).await.unwrap();
+/// # }
+/// ```
+pub struct AsyncLineEventHandle {
+    evented: PollEvented<PollWrapper>,
+}
+
+impl AsyncLineEventHandle {
+    /// Wraps the specified `LineEventHandle`.
+    ///
+    /// # Arguments
+    ///
+    /// * `handle` - handle to be wrapped.
+    pub fn new(handle: LineEventHandle) -> Result<AsyncLineEventHandle> {
+        // The file descriptor needs to be configured for non-blocking I/O for PollEvented to work.
+        let fd = handle.file.as_raw_fd();
+        unsafe {
+            let flags = libc::fcntl(fd, libc::F_GETFL, 0);
+            libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK);
+        }
+
+        Ok(AsyncLineEventHandle {
+            evented: PollEvented::new(PollWrapper { handle })?,
+        })
+    }
+}
+
+impl Stream for AsyncLineEventHandle {
+    type Item = Result<LineEvent>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        let ready = Ready::readable();
+        if let Err(e) = ready!(self.evented.poll_read_ready(cx, ready)) {
+            return Poll::Ready(Some(Err(e.into())));
+        }
+
+        match self.evented.get_ref().handle.read_event() {
+            Ok(Some(event)) => Poll::Ready(Some(Ok(event))),
+            Ok(None) => Poll::Ready(Some(Err(event_err(nix::Error::Sys(
+                nix::errno::Errno::EIO,
+            ))))),
+            Err(nix::Error::Sys(nix::errno::Errno::EAGAIN)) => {
+                self.evented.clear_read_ready(cx, ready)?;
+                Poll::Pending
+            }
+            Err(e) => Poll::Ready(Some(Err(event_err(e)))),
+        }
+    }
+}
+
+impl AsRef<LineEventHandle> for AsyncLineEventHandle {
+    fn as_ref(&self) -> &LineEventHandle {
+        &self.evented.get_ref().handle
+    }
+}