Repository: hannobraun/inotify-rs (language: rust)

Issue:
Update mio dependency
mio version 0.6 depends on deprecated crate. Upgrade to version 0.7.


Discussion / hints:
Thanks for the the note!
And likewise, tokio 0.3, which has recently been released.

Updating both seems mostly easy, except for the fact Tokio no longer includes `PollEvented` in its public API (it no longer exposes it's dependency on mio in the public API in general).

I suppose the solution would be to manually implement `AsyncRead` and `AsyncWrite` similarly to how `PollEvented` does it?
Thanks, @ids1024. Any pull requests doing such upgrades are definitely welcome. Not sure what the best way to handle `PollEvented` would be. I still don't have significant experience with the whole async ecosystem (that work has always been driven by contributors).
I *think* this could use [`AsyncFd`](https://github.com/tokio-rs/tokio/pull/2903) which was just merged into tokio. So I guess this is waiting on the next minor Tokio release.

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 00e36f7..e0680f2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -27,7 +27,7 @@ travis-ci   = { repository = "inotify-rs/inotify" }
 
 [features]
 default = ["stream"]
-stream = ["futures-core", "mio", "tokio"]
+stream = ["futures-core", "tokio"]
 
 
 [dependencies]
@@ -35,13 +35,12 @@ bitflags     = "1"
 futures-core = { version = "0.3.1", optional = true }
 inotify-sys  = "0.1.3"
 libc         = "0.2"
-mio          = { version = "0.6", optional = true }
-tokio        = { version = "0.2.2", optional = true, features = ["io-driver"] }
+tokio        = { version = "0.3.3", optional = true, features = ["net"] }
 
 [dev-dependencies]
 tempfile     = "3.1.0"
 futures-util = "0.3.1"
-tokio        = { version = "0.2.1", features = ["macros", "rt-core"] }
+tokio        = { version = "0.3.3", features = ["macros", "rt-multi-thread"] }
 
 [[example]]
 name              = "stream"
diff --git a/src/stream.rs b/src/stream.rs
index fe0211f..20d40bc 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -1,33 +1,25 @@
 use std::{
     io,
-    ops::Deref,
+    os::unix::io::{AsRawFd, RawFd},
     pin::Pin,
     sync::Arc,
-    task::{Poll, Context},
+    task::{Context, Poll},
 };
 
-use mio::{
-    event::Evented,
-    unix::EventedFd,
-};
-use tokio::io::{AsyncRead, PollEvented};
-use futures_core::{Stream, ready};
+use futures_core::{ready, Stream};
+use tokio::io::unix::AsyncFd;
 
-use crate::events::{
-    Event,
-    EventOwned,
-};
+use crate::events::{Event, EventOwned};
 use crate::fd_guard::FdGuard;
 use crate::util::read_into_buffer;
 
-
 /// Stream of inotify events
 ///
 /// Allows for streaming events returned by [`Inotify::event_stream`].
 ///
 /// [`Inotify::event_stream`]: struct.Inotify.html#method.event_stream
 pub struct EventStream<T> {
-    fd: PollEvented<EventedFdGuard>,
+    fd: AsyncFd<ArcFdGuard>,
     buffer: T,
     buffer_pos: usize,
     unused_bytes: usize,
@@ -40,7 +32,7 @@ where
     /// Returns a new `EventStream` associated with the default reactor.
     pub(crate) fn new(fd: Arc<FdGuard>, buffer: T) -> io::Result<Self> {
         Ok(EventStream {
-            fd: PollEvented::new(EventedFdGuard(fd))?,
+            fd: AsyncFd::new(ArcFdGuard(fd))?,
             buffer: buffer,
             buffer_pos: 0,
             unused_bytes: 0,
@@ -54,15 +46,14 @@ where
 {
     type Item = io::Result<EventOwned>;
 
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>
-    {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         // Safety: safe because we never move out of `self_`.
         let self_ = unsafe { self.get_unchecked_mut() };
 
         if self_.unused_bytes == 0 {
             // Nothing usable in buffer. Need to reset and fill buffer.
-            self_.buffer_pos   = 0;
-            self_.unused_bytes = ready!(Pin::new(&mut self_.fd).poll_read(cx, self_.buffer.as_mut()))?;
+            self_.buffer_pos = 0;
+            self_.unused_bytes = ready!(read(&self_.fd, self_.buffer.as_mut(), cx))?;
         }
 
         if self_.unused_bytes == 0 {
@@ -75,62 +66,39 @@ where
         // there, and we only take complete events out. That means we have at
         // least one event in there and can call `from_buffer` to take it out.
         let (bytes_consumed, event) = Event::from_buffer(
-            Arc::downgrade(self_.fd.get_ref()),
+            Arc::downgrade(&self_.fd.get_ref().0),
             &self_.buffer.as_ref()[self_.buffer_pos..],
         );
-        self_.buffer_pos   += bytes_consumed;
+        self_.buffer_pos += bytes_consumed;
         self_.unused_bytes -= bytes_consumed;
 
         Poll::Ready(Some(Ok(event.into_owned())))
     }
 }
 
-#[derive(Clone, Debug, PartialEq)]
-struct EventedFdGuard(Arc<FdGuard>);
+// Newtype wrapper because AsRawFd isn't implemented for Arc<T> where T: AsRawFd.
+struct ArcFdGuard(Arc<FdGuard>);
 
-impl Evented for EventedFdGuard {
-    #[inline]
-    fn register(&self,
-                poll: &mio::Poll,
-                token: mio::Token,
-                interest: mio::Ready,
-                opts: mio::PollOpt)
-                -> io::Result<()>
-    {
-        EventedFd(&(self.fd)).register(poll, token, interest, opts)
-    }
-
-    #[inline]
-    fn reregister(&self,
-                  poll: &mio::Poll,
-                  token: mio::Token,
-                  interest: mio::Ready,
-                  opts: mio::PollOpt)
-                  -> io::Result<()>
-    {
-        EventedFd(&(self.fd)).reregister(poll, token, interest, opts)
-    }
-
-    #[inline]
-    fn deregister(&self, poll: &mio::Poll) -> io::Result<()> {
-        EventedFd(&self.fd).deregister(poll)
+impl AsRawFd for ArcFdGuard {
+    fn as_raw_fd(&self) -> RawFd {
+        self.0.as_raw_fd()
     }
 }
 
-impl io::Read for EventedFdGuard {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        match read_into_buffer(self.fd, buf) {
-            i if i >= 0 => Ok(i as usize),
-            _ => Err(io::Error::last_os_error()),
+fn read(fd: &AsyncFd<ArcFdGuard>, buffer: &mut [u8], cx: &mut Context) -> Poll<io::Result<usize>> {
+    let mut guard = ready!(fd.poll_read_ready(cx))?;
+    let result = guard.with_io(|| {
+        let read = read_into_buffer(fd.as_raw_fd(), buffer);
+        if read == -1 {
+            return Err(io::Error::last_os_error());
         }
-    }
-}
 
-impl Deref for EventedFdGuard {
-    type Target = Arc<FdGuard>;
-    #[inline]
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
+        Ok(read as usize)
+    });
 
+    match result {
+        Ok(read) => Poll::Ready(Ok(read)),
+        Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => Poll::Pending,
+        Err(err) => Poll::Ready(Err(err)),
+    }
 }


Test patch: