Add asynchronous line event stream for Tokio.

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"

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,159 @@
+// 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::mem;
+use std::os::unix::io::AsRawFd;
+use std::pin::Pin;
+use std::slice;
+
+use super::errors::event_err;
+use super::{ffi, 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())));
+        }
+
+        // TODO: This code should not be duplicated here.
+        let mut data: ffi::gpioevent_data = unsafe { mem::zeroed() };
+        let mut data_as_buf = unsafe {
+            slice::from_raw_parts_mut(
+                &mut data as *mut ffi::gpioevent_data as *mut u8,
+                mem::size_of::<ffi::gpioevent_data>(),
+            )
+        };
+        match nix::unistd::read(
+            self.evented.get_ref().handle.file.as_raw_fd(),
+            &mut data_as_buf,
+        ) {
+            Ok(bytes_read) => {
+                if bytes_read != mem::size_of::<ffi::gpioevent_data>() {
+                    let e = nix::Error::Sys(nix::errno::Errno::EIO);
+                    Poll::Ready(Some(Err(event_err(e))))
+                } else {
+                    Poll::Ready(Some(Ok(LineEvent(data))))
+                }
+            }
+            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
+    }
+}

src/lib.rs

@@ -89,6 +89,12 @@ extern crate bitflags;
 extern crate libc;
 #[macro_use]
 extern crate nix;
+#[cfg(feature = "async-tokio")]
+extern crate futures;
+#[cfg(feature = "async-tokio")]
+extern crate mio;
+#[cfg(feature = "async-tokio")]
+extern crate tokio;
 
 use std::cmp::min;
 use std::ffi::CStr;
@@ -101,9 +107,13 @@ use std::ptr;
 use std::slice;
 use std::sync::Arc;
 
+#[cfg(feature = "async-tokio")]
+mod async_tokio;
 pub mod errors;
 mod ffi;
 
+#[cfg(feature = "async-tokio")]
+pub use crate::async_tokio::AsyncLineEventHandle;
 use errors::*;
 
 unsafe fn rstr_lcpy(dst: *mut libc::c_char, src: &str, length: usize) {
@@ -550,6 +560,17 @@ impl Line {
             file: unsafe { File::from_raw_fd(request.fd) },
         })
     }
+
+    #[cfg(feature = "async-tokio")]
+    pub fn async_events(
+        &self,
+        handle_flags: LineRequestFlags,
+        event_flags: EventRequestFlags,
+        consumer: &str,
+    ) -> Result<AsyncLineEventHandle> {
+        let events = self.events(handle_flags, event_flags, consumer)?;
+        Ok(AsyncLineEventHandle::new(events)?)
+    }
 }
 
 impl LineInfo {