Merge pull request #52 from kaojo/feature/issue_50

Migration to 'tokio' crate from crate 'tokio-core'

Author: posborne

.gitignore

@@ -11,3 +11,6 @@
 # Generated by Cargo
 Cargo.lock
 /target/
+
+# Intellij IDEA
+.idea

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [master] - Unreleased
 
+### Changed
+
+- Migrated to 'tokio' crate. 
+
 ## [0.5.3] - 2018-04-19
 
 ### Fixed

Cargo.toml

@@ -11,10 +11,10 @@ readme = "README.md"
 
 [features]
 mio-evented = ["mio"]
-tokio = ["futures", "tokio-core", "mio-evented"]
+use_tokio = ["futures", "tokio", "mio-evented"]
 
 [dependencies]
 futures = { version = "0.1", optional = true }
 nix = "0.10.0"
 mio = { version = "0.6", optional = true }
-tokio-core = { version = "0.1", optional = true }
+tokio = { version = "0.1", optional = true }

README.md

@@ -80,8 +80,8 @@ The following features are planned for the library:
 - [ ] Support for configuring whether a pin is active low/high
 - [x] Support for configuring interrupts on GPIO
 - [x] Support for polling on GPIO with configured interrupt
-- [x] Support for asynchronous polling using `mio` or `tokio-core` (requires
-      enabling the `mio-evented` or `tokio` crate features, respectively)
+- [x] Support for asynchronous polling using `mio` or `tokio` (requires
+      enabling the `mio-evented` or `use_tokio` crate features, respectively)
 
 ## Cross Compiling
 

examples/tokio.rs

@@ -1,47 +1,47 @@
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
 extern crate futures;
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
 extern crate sysfs_gpio;
-#[cfg(feature = "tokio")]
-extern crate tokio_core;
+#[cfg(feature = "use_tokio")]
+extern crate tokio;
 
-#[cfg(feature = "tokio")]
-use futures::{Future, Stream};
-#[cfg(feature = "tokio")]
-use sysfs_gpio::{Direction, Edge, Pin};
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
 use std::env;
-#[cfg(feature = "tokio")]
-use tokio_core::reactor::Core;
 
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
+use futures::{Future, lazy, Stream};
+
+#[cfg(feature = "use_tokio")]
+use sysfs_gpio::{Direction, Edge, Pin};
+
+#[cfg(feature = "use_tokio")]
 fn stream(pin_nums: Vec<u64>) -> sysfs_gpio::Result<()> {
     // 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 mut l = Core::new()?;
-    let handle = l.handle();
-    for &(i, ref pin) in pins.iter() {
-        pin.export()?;
-        pin.set_direction(Direction::In)?;
-        pin.set_edge(Edge::BothEdges)?;
-        handle.spawn(pin.get_value_stream(&handle)?
-                         .for_each(move |val| {
-                                       println!("Pin {} changed value to {}", i, val);
-                                       Ok(())
-                                   })
-                         .map_err(|_| ()));
-    }
-    // Wait forever for events
-    loop {
-        l.turn(None)
-    }
+    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(())
 }
 
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
 fn main() {
     let pins: Vec<u64> = env::args()
         .skip(1)
@@ -54,7 +54,7 @@ fn main() {
     }
 }
 
-#[cfg(not(feature = "tokio"))]
+#[cfg(not(feature = "use_tokio"))]
 fn main() {
     println!("This example requires the `tokio` feature to be enabled.");
 }

src/lib.rs

@@ -41,38 +41,36 @@
 //! }
 //! ```
 
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
 extern crate futures;
 #[cfg(feature = "mio-evented")]
 extern crate mio;
 extern crate nix;
-#[cfg(feature = "tokio")]
-extern crate tokio_core;
+#[cfg(feature = "use_tokio")]
+extern crate tokio;
 
-#[cfg(feature = "tokio")]
-use futures::{Async, Poll, Stream};
+use std::fs;
+use std::fs::File;
+use std::io::{self, SeekFrom};
+use std::io::prelude::*;
+use std::os::unix::prelude::*;
+use std::path::Path;
 
+#[cfg(feature = "use_tokio")]
+use futures::{Async, Poll, Stream};
 #[cfg(feature = "mio-evented")]
 use mio::Evented;
 #[cfg(feature = "mio-evented")]
 use mio::unix::EventedFd;
-
 #[cfg(any(target_os = "linux", target_os = "android"))]
 use nix::sys::epoll::*;
 use nix::unistd::close;
+#[cfg(feature = "use_tokio")]
+use tokio::reactor::{Handle, PollEvented};
 
-use std::io::prelude::*;
-use std::os::unix::prelude::*;
-use std::io::{self, SeekFrom};
-use std::fs;
-use std::fs::File;
-use std::path::Path;
-
-#[cfg(feature = "tokio")]
-use tokio_core::reactor::{Handle, PollEvented};
+pub use error::Error;
 
 mod error;
-pub use error::Error;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct Pin {
@@ -446,29 +444,57 @@ impl Pin {
 
     /// Get a Stream of pin interrupts for this pin
     ///
-    /// The PinStream object can be used with the `tokio-core` crate. You should probably call
+    /// The PinStream object can be used with the `tokio` crate. You should probably call
     /// `set_edge()` before using this.
     ///
-    /// This method is only available when the `tokio` crate feature is enabled.
-    #[cfg(feature = "tokio")]
-    pub fn get_stream(&self, handle: &Handle) -> Result<PinStream> {
-        PinStream::init(self.clone(), handle)
+    /// This method is only available when the `use_tokio` crate feature is enabled.
+    #[cfg(feature = "use_tokio")]
+    pub fn get_stream_with_handle(&self, handle: &Handle) -> Result<PinStream> {
+        PinStream::init_with_handle(self.clone(), handle)
+    }
+
+
+    /// Get a Stream of pin interrupts for this pin
+    ///
+    /// The PinStream object can be used with the `tokio` crate. You should probably call
+    /// `set_edge()` before using this.
+    ///
+    /// This method is only available when the `use_tokio` crate feature is enabled.
+    #[cfg(feature = "use_tokio")]
+    pub fn get_stream(&self) -> Result<PinStream> {
+        PinStream::init(self.clone())
     }
 
     /// Get a Stream of pin values for this pin
     ///
-    /// The PinStream object can be used with the `tokio-core` crate. You should probably call
+    /// The PinStream object can be used with the `tokio` crate. You should probably call
     /// `set_edge(Edge::BothEdges)` before using this.
     ///
     /// Note that the values produced are the value of the pin as soon as we get to handling the
     /// interrupt in userspace.  Each time this stream produces a value, a change has occurred, but
     /// it could end up producing the same value multiple times if the value has changed back
     /// between when the interrupt occurred and when the value was read.
     ///
-    /// This method is only available when the `tokio` crate feature is enabled.
-    #[cfg(feature = "tokio")]
-    pub fn get_value_stream(&self, handle: &Handle) -> Result<PinValueStream> {
-        Ok(PinValueStream(PinStream::init(self.clone(), handle)?))
+    /// This method is only available when the `use_tokio` crate feature is enabled.
+    #[cfg(feature = "use_tokio")]
+    pub fn get_value_stream_with_handle(&self, handle: &Handle) -> Result<PinValueStream> {
+        Ok(PinValueStream(PinStream::init_with_handle(self.clone(), handle)?))
+    }
+
+    /// Get a Stream of pin values for this pin
+    ///
+    /// The PinStream object can be used with the `tokio` crate. You should probably call
+    /// `set_edge(Edge::BothEdges)` before using this.
+    ///
+    /// Note that the values produced are the value of the pin as soon as we get to handling the
+    /// interrupt in userspace.  Each time this stream produces a value, a change has occurred, but
+    /// it could end up producing the same value multiple times if the value has changed back
+    /// between when the interrupt occurred and when the value was read.
+    ///
+    /// This method is only available when the `use_tokio` crate feature is enabled.
+    #[cfg(feature = "use_tokio")]
+    pub fn get_value_stream(&self) -> Result<PinValueStream> {
+        Ok(PinValueStream(PinStream::init(self.clone())?))
     }
 }
 
@@ -538,7 +564,7 @@ impl PinPoller {
     /// making this call.  This call makes use of epoll under the
     /// covers.  To poll on multiple GPIOs or other event sources,
     /// poll asynchronously using the integration with either `mio`
-    /// or `tokio_core`.
+    /// or `tokio`.
     ///
     /// This function will return Some(value) of the pin if a change is
     /// detected or None if a timeout occurs.  Note that the value provided
@@ -613,23 +639,33 @@ impl Evented for AsyncPinPoller {
     }
 }
 
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
 pub struct PinStream {
     evented: PollEvented<AsyncPinPoller>,
     skipped_first_event: bool,
 }
 
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
 impl PinStream {
-    pub fn init(pin: Pin, handle: &Handle) -> Result<Self> {
+    pub fn init_with_handle(pin: Pin, handle: &Handle) -> Result<Self> {
         Ok(PinStream {
                evented: PollEvented::new(pin.get_async_poller()?, &handle)?,
                skipped_first_event: false,
            })
     }
 }
 
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
+impl PinStream {
+    pub fn init(pin: Pin) -> Result<Self> {
+        Ok(PinStream {
+               evented: PollEvented::new(pin.get_async_poller()?, &Handle::default())?,
+               skipped_first_event: false,
+           })
+    }
+}
+
+#[cfg(feature = "use_tokio")]
 impl Stream for PinStream {
     type Item = ();
     type Error = Error;
@@ -650,18 +686,18 @@ impl Stream for PinStream {
     }
 }
 
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
 pub struct PinValueStream(PinStream);
 
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
 impl PinValueStream {
     #[inline]
     fn get_value(&mut self) -> Result<u8> {
         get_value_from_file(&mut self.0.evented.get_mut().devfile)
     }
 }
 
-#[cfg(feature = "tokio")]
+#[cfg(feature = "use_tokio")]
 impl Stream for PinValueStream {
     type Item = u8;
     type Error = Error;