chore: bump tokio to v1

Remove dependency of mio and use tokio AsyncFd instead.
This since tokio no longer exposes mio internals.

Also bump MSRV to 1.45, required due to tokio update.

Author: oll3

.travis.yml

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

Cargo.toml

@@ -12,8 +12,8 @@ license = "MIT OR Apache-2.0"
 edition = "2018"
 
 [features]
-default = []
-async-tokio = ["tokio", "futures", "mio"]
+default = ["async-tokio"]
+async-tokio = ["tokio", "futures"]
 
 [[example]]
 name = "async_tokio"
@@ -23,9 +23,8 @@ required-features = ["async-tokio"]
 bitflags = "1.0"
 libc = "0.2"
 nix = "0.14"
-tokio = { version = "0.2", features = ["io-driver", "rt-threaded", "macros"], optional = true }
+tokio = { version = "1", features = ["io-std", "rt-multi-thread", "macros", "net"], optional = true }
 futures = { version = "0.3", optional = true }
-mio = { version = "0.6", optional = true }
 
 [dev-dependencies]
 quicli = "0.2"

src/async_tokio.rs

@@ -11,48 +11,14 @@
 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 tokio::io::unix::{AsyncFd, TryIoError};
 
-use std::io;
 use std::os::unix::io::AsRawFd;
 use std::pin::Pin;
 
 use super::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
@@ -88,7 +54,7 @@ impl Evented for PollWrapper {
 /// # }
 /// ```
 pub struct AsyncLineEventHandle {
-    evented: PollEvented<PollWrapper>,
+    asyncfd: AsyncFd<LineEventHandle>,
 }
 
 impl AsyncLineEventHandle {
@@ -106,36 +72,35 @@ impl AsyncLineEventHandle {
         }
 
         Ok(AsyncLineEventHandle {
-            evented: PollEvented::new(PollWrapper { handle })?,
+            asyncfd: AsyncFd::new(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
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        loop {
+            let mut guard = ready!(self.asyncfd.poll_read_ready_mut(cx))?;
+            match guard.try_io(|inner| inner.get_mut().read_event()) {
+                Err(TryIoError { .. }) => {
+                    // Continue
+                }
+                Ok(Ok(Some(event))) => return Poll::Ready(Some(Ok(event))),
+                Ok(Ok(None)) => {
+                    return Poll::Ready(Some(Err(event_err(nix::Error::Sys(
+                        nix::errno::Errno::EIO,
+                    )))))
+                }
+                Ok(Err(err)) => return Poll::Ready(Some(Err(err.into()))),
             }
-            Err(e) => Poll::Ready(Some(Err(event_err(e)))),
         }
     }
 }
 
 impl AsRef<LineEventHandle> for AsyncLineEventHandle {
     fn as_ref(&self) -> &LineEventHandle {
-        &self.evented.get_ref().handle
+        &self.asyncfd.get_ref()
     }
 }

src/lib.rs

@@ -92,6 +92,7 @@ extern crate nix;
 use std::cmp::min;
 use std::ffi::CStr;
 use std::fs::{read_dir, File, ReadDir};
+use std::io::Read;
 use std::mem;
 use std::ops::Index;
 use std::os::unix::io::{AsRawFd, FromRawFd, RawFd};
@@ -954,11 +955,11 @@ impl LineEventHandle {
     /// This blocks while there is not another event available from the
     /// kernel for the line which matches the subscription criteria
     /// specified in the `event_flags` when the handle was created.
-    pub fn get_event(&self) -> Result<LineEvent> {
+    pub fn get_event(&mut self) -> Result<LineEvent> {
         match self.read_event() {
             Ok(Some(event)) => Ok(event),
             Ok(None) => Err(event_err(nix::Error::Sys(nix::errno::Errno::EIO))),
-            Err(e) => Err(event_err(e)),
+            Err(e) => Err(e.into()),
         }
     }
 
@@ -981,19 +982,15 @@ impl LineEventHandle {
 
     /// Helper function which returns the line event if a complete event was read, Ok(None) if not
     /// enough data was read or the error returned by `read()`.
-    ///
-    /// This function allows access to the raw `nix::Error` as required, for example, to theck
-    /// whether read() returned -EAGAIN.
-    pub(crate) fn read_event(&self) -> std::result::Result<Option<LineEvent>, nix::Error> {
+    pub(crate) fn read_event(&mut self) -> std::io::Result<Option<LineEvent>> {
         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>(),
             )
         };
-        let bytes_read = nix::unistd::read(self.file.as_raw_fd(), &mut data_as_buf)?;
-
+        let bytes_read = self.file.read(&mut data_as_buf)?;
         if bytes_read != mem::size_of::<ffi::gpioevent_data>() {
             Ok(None)
         } else {
@@ -1016,7 +1013,7 @@ impl Iterator for LineEventHandle {
         match self.read_event() {
             Ok(None) => None,
             Ok(Some(event)) => Some(Ok(event)),
-            Err(e) => Some(Err(event_err(e))),
+            Err(e) => Some(Err(e.into())),
         }
     }
 }