[IN PROGRESS] Update to mio 0.7 and tokio 0.3

mio-udev/Cargo.toml

@@ -13,5 +13,5 @@ edition = "2018"
 
 [dependencies]
 udev = "0.4"
-mio = "0.6"
+mio = { version = "0.7", features = ["os-util"] }
 libc = "0.2"

mio-udev/src/lib.rs

@@ -41,9 +41,9 @@ use std::ffi::OsStr;
 use std::io;
 use std::os::unix::io::{AsRawFd, RawFd};
 
-use mio::event::Evented;
-use mio::unix::EventedFd;
-use mio::{Poll, PollOpt, Ready, Token};
+use mio::event::Source;
+use mio::unix::SourceFd;
+use mio::{Interest, Registry, Token};
 
 /// Monitors for device events.
 ///
@@ -157,29 +157,33 @@ impl MonitorSocket {
     }
 }
 
-impl Evented for MonitorSocket {
+impl AsRawFd for MonitorSocket {
+    fn as_raw_fd(&self) -> RawFd {
+        self.fd()
+    }
+}
+
+impl Source for MonitorSocket {
     fn register(
-        &self,
-        poll: &Poll,
+        &mut self,
+        registry: &Registry,
         token: Token,
-        interest: Ready,
-        opts: PollOpt,
+        interest: Interest,
     ) -> io::Result<()> {
-        EventedFd(&self.fd()).register(poll, token, interest, opts)
+        SourceFd(&self.fd()).register(registry, token, interest)
     }
 
     fn reregister(
-        &self,
-        poll: &Poll,
+        &mut self,
+        registry: &Registry,
         token: Token,
-        interest: Ready,
-        opts: PollOpt,
+        interest: Interest,
     ) -> io::Result<()> {
-        EventedFd(&self.fd()).reregister(poll, token, interest, opts)
+        SourceFd(&self.fd()).reregister(registry, token, interest)
     }
 
-    fn deregister(&self, poll: &Poll) -> io::Result<()> {
-        EventedFd(&self.fd()).deregister(poll)
+    fn deregister(&mut self, registry: &Registry) -> io::Result<()> {
+        SourceFd(&self.fd()).deregister(registry)
     }
 }
 

tokio-udev/Cargo.toml

@@ -12,12 +12,12 @@ documentation = "https://docs.rs/tokio-udev"
 edition = "2018"
 
 [dependencies]
-mio = "0.6"
+mio = { version = "0.7", features = ["os-poll"] }
 mio-udev = { path = "../mio-udev", version = "0.3.0" }
 
 futures-core = "0.3"
-tokio = { version = "0.2", features = ["io-driver"] }
+tokio = { version = ">=0.3.2", features = ["net"] } 
 
 [dev-dependencies]
 futures-util = "0.3"
-tokio = { version = "0.2", features = ["macros"] }
+tokio = { version = ">=0.3.2", features = ["macros", "rt-multi-thread"] }

tokio-udev/examples/usb_hotplug.rs

@@ -22,11 +22,13 @@ async fn main() {
     let monitor = builder.listen().expect("Couldn't create MonitorSocket");
     monitor
         .for_each(|event| {
-            println!(
-                "Hotplug event: {}: {}",
-                event.event_type(),
-                event.device().syspath().display()
-            );
+            if let Ok(event) = event {
+                println!(
+                    "Hotplug event: {}: {}",
+                    event.event_type(),
+                    event.device().syspath().display()
+                );
+            }
             ready(())
         })
         .await

tokio-udev/src/lib.rs

@@ -35,14 +35,13 @@ pub use mio_udev::{
     Property,
 };
 
+use futures_core::stream::Stream;
 use std::ffi::OsStr;
 use std::io;
 use std::pin::Pin;
 use std::sync::Mutex;
 use std::task::Poll;
-
-use futures_core::stream::Stream;
-use tokio::io::PollEvented;
+use tokio::io::unix::AsyncFd;
 
 /// Monitors for device events.
 ///
@@ -133,45 +132,38 @@ unsafe impl Send for MonitorSocket {}
 unsafe impl Sync for MonitorSocket {}
 
 impl Stream for MonitorSocket {
-    type Item = mio_udev::Event;
+    type Item = Result<mio_udev::Event, io::Error>;
 
     fn poll_next(
         self: Pin<&mut Self>,
-        cx: &mut std::task::Context,
+        ctx: &mut std::task::Context,
     ) -> Poll<Option<Self::Item>> {
-        self.inner.lock().unwrap().poll_receive(cx)
+        self.inner.lock().unwrap().poll_receive(ctx)
     }
 }
 
 struct Inner {
-    io: PollEvented<mio_udev::MonitorSocket>,
+    fd: AsyncFd<mio_udev::MonitorSocket>,
 }
 
 impl Inner {
     fn new(monitor: mio_udev::MonitorSocket) -> io::Result<Inner> {
         Ok(Inner {
-            io: PollEvented::new(monitor)?,
+            fd: AsyncFd::new(monitor)?,
         })
     }
 
     fn poll_receive(
         &mut self,
-        cx: &mut std::task::Context,
-    ) -> Poll<Option<mio_udev::Event>> {
-        if let Poll::Pending =
-            self.io.poll_read_ready(cx, mio::Ready::readable())
-        {
-            return Poll::Pending;
-        }
-
-        match self.io.get_mut().next() {
-            Some(event) => Poll::Ready(Some(event)),
-            None => {
-                self.io
-                    .clear_read_ready(cx, mio::Ready::readable())
-                    .unwrap();
-                Poll::Pending
+        ctx: &mut std::task::Context,
+    ) -> Poll<Option<Result<mio_udev::Event, io::Error>>> {
+        match self.fd.poll_read_ready(ctx) {
+            Poll::Ready(Ok(mut ready_guard)) => {
+                ready_guard.clear_ready();
+                Poll::Ready(self.fd.get_mut().next().map(Ok))
             }
+            Poll::Ready(Err(err)) => Poll::Ready(Some(Err(err))),
+            Poll::Pending => Poll::Pending,
         }
     }
 }