chore: bump tokio to v1

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

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())),
         }
     }
 }