Skip to content

Commit

Permalink
Add asynchronous line event stream for Tokio.
Browse files Browse the repository at this point in the history
  • Loading branch information
mgottschlag committed May 13, 2020
1 parent 1e69adc commit f9ee6b4
Show file tree
Hide file tree
Showing 4 changed files with 202 additions and 0 deletions.
12 changes: 12 additions & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
44 changes: 44 additions & 0 deletions examples/async_tokio.rs
Original file line number Diff line number Diff line change
@@ -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();
}
125 changes: 125 additions & 0 deletions src/async_tokio.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,125 @@
// 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::Result;
use super::{LineEvent, LineEventHandle};

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> {
Ok(AsyncLineEventHandle {
evented: PollEvented::new(PollWrapper { handle })?,
})
}
}

impl Stream for AsyncLineEventHandle {
type Item = Result<LineEvent>;

fn poll_next(mut 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())));
}

Poll::Ready(self.evented.get_mut().handle.next())
}
}

impl AsRef<LineEventHandle> for AsyncLineEventHandle {
fn as_ref(&self) -> &LineEventHandle {
&self.evented.get_ref().handle
}
}
21 changes: 21 additions & 0 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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) {
Expand Down Expand Up @@ -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 {
Expand Down

0 comments on commit f9ee6b4

Please sign in to comment.