switch to smol::Timer

Author: dignifiedquire

Cargo.toml

@@ -27,7 +27,6 @@ default = [
   "crossbeam-channel",
   "crossbeam-deque",
   "crossbeam-queue",
-  "futures-timer",
   "kv-log-macro",
   "log",
   "mio",
@@ -37,7 +36,7 @@ default = [
   "smol",
 ]
 docs = ["attributes", "unstable", "default"]
-unstable = ["std", "broadcaster", "futures-timer"]
+unstable = ["std", "broadcaster"]
 attributes = ["async-attributes"]
 std = [
   "alloc",
@@ -64,7 +63,6 @@ crossbeam-queue = { version = "0.2.0", optional = true }
 crossbeam-utils = { version = "0.7.2", optional = true }
 futures-core = { version = "0.3.4", optional = true, default-features = false }
 futures-io = { version = "0.3.4", optional = true }
-futures-timer = { version = "3.0.2", optional = true }
 kv-log-macro = { version = "1.0.4", optional = true }
 log = { version = "0.4.8", features = ["kv_unstable"], optional = true }
 memchr = { version = "2.3.3", optional = true }

src/future/future/delay.rs

@@ -2,8 +2,8 @@ use std::future::Future;
 use std::pin::Pin;
 use std::time::Duration;
 
-use futures_timer::Delay;
 use pin_project_lite::pin_project;
+use smol::Timer;
 
 use crate::task::{Context, Poll};
 
@@ -14,13 +14,13 @@ pin_project! {
         #[pin]
         future: F,
         #[pin]
-        delay: Delay,
+        delay: Timer,
     }
 }
 
 impl<F> DelayFuture<F> {
     pub fn new(future: F, dur: Duration) -> DelayFuture<F> {
-        let delay = Delay::new(dur);
+        let delay = Timer::after(dur);
 
         DelayFuture { future, delay }
     }

src/future/timeout.rs

@@ -1,11 +1,11 @@
 use std::error::Error;
 use std::fmt;
+use std::future::Future;
 use std::pin::Pin;
 use std::time::Duration;
-use std::future::Future;
 
-use futures_timer::Delay;
 use pin_project_lite::pin_project;
+use smol::Timer;
 
 use crate::task::{Context, Poll};
 
@@ -33,11 +33,7 @@ pub async fn timeout<F, T>(dur: Duration, f: F) -> Result<T, TimeoutError>
 where
     F: Future<Output = T>,
 {
-    let f = TimeoutFuture {
-        future: f,
-        delay: Delay::new(dur),
-    };
-    f.await
+    TimeoutFuture::new(f, dur).await
 }
 
 pin_project! {
@@ -46,14 +42,17 @@ pin_project! {
         #[pin]
         future: F,
         #[pin]
-        delay: Delay,
+        delay: Timer,
     }
 }
 
 impl<F> TimeoutFuture<F> {
     #[allow(dead_code)]
     pub(super) fn new(future: F, dur: Duration) -> TimeoutFuture<F> {
-        TimeoutFuture { future: future, delay: Delay::new(dur) }
+        TimeoutFuture {
+            future,
+            delay: Timer::after(dur),
+        }
     }
 }
 

src/io/timeout.rs

@@ -1,10 +1,10 @@
+use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::time::Duration;
-use std::future::Future;
 
-use futures_timer::Delay;
 use pin_project_lite::pin_project;
+use smol::Timer;
 
 use crate::io;
 
@@ -37,7 +37,7 @@ where
     F: Future<Output = io::Result<T>>,
 {
     Timeout {
-        timeout: Delay::new(dur),
+        timeout: Timer::after(dur),
         future: f,
     }
     .await
@@ -53,7 +53,7 @@ pin_project! {
         #[pin]
         future: F,
         #[pin]
-        timeout: Delay,
+        timeout: Timer,
     }
 }
 

src/os/unix/net/datagram.rs

@@ -319,8 +319,8 @@ impl AsRawFd for UnixDatagram {
 
 impl FromRawFd for UnixDatagram {
     unsafe fn from_raw_fd(fd: RawFd) -> UnixDatagram {
-        let datagram = std::os::unix::net::UnixDatagram::from_raw_fd(fd);
-        datagram.into()
+        let datagram = Async::<StdUnixDatagram>::from_raw_fd(fd);
+        UnixDatagram { watcher: datagram }
     }
 }
 

src/stream/interval.rs

@@ -1,10 +1,10 @@
+use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::time::Duration;
 
-use crate::future::Future;
 use crate::stream::Stream;
-use futures_timer::Delay;
+use smol::Timer;
 
 /// Creates a new stream that yields at a set interval.
 ///
@@ -45,7 +45,7 @@ use futures_timer::Delay;
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
 pub fn interval(dur: Duration) -> Interval {
     Interval {
-        delay: Delay::new(dur),
+        delay: Timer::after(dur),
         interval: dur,
     }
 }
@@ -60,7 +60,7 @@ pub fn interval(dur: Duration) -> Interval {
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
 #[derive(Debug)]
 pub struct Interval {
-    delay: Delay,
+    delay: Timer,
     interval: Duration,
 }
 
@@ -72,7 +72,7 @@ impl Stream for Interval {
             return Poll::Pending;
         }
         let interval = self.interval;
-        self.delay.reset(interval);
+        std::mem::replace(&mut self.delay, Timer::after(interval));
         Poll::Ready(Some(()))
     }
 }

src/stream/stream/delay.rs

@@ -3,6 +3,7 @@ use core::pin::Pin;
 use core::time::Duration;
 
 use pin_project_lite::pin_project;
+use smol::Timer;
 
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
@@ -14,7 +15,7 @@ pin_project! {
         #[pin]
         stream: S,
         #[pin]
-        delay: futures_timer::Delay,
+        delay: Timer,
         delay_done: bool,
     }
 }
@@ -23,7 +24,7 @@ impl<S> Delay<S> {
     pub(super) fn new(stream: S, dur: Duration) -> Self {
         Delay {
             stream,
-            delay: futures_timer::Delay::new(dur),
+            delay: Timer::after(dur),
             delay_done: false,
         }
     }

src/stream/stream/throttle.rs

@@ -2,8 +2,8 @@ use std::future::Future;
 use std::pin::Pin;
 use std::time::Duration;
 
-use futures_timer::Delay;
 use pin_project_lite::pin_project;
+use smol::Timer;
 
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
@@ -25,7 +25,7 @@ pin_project! {
         #[pin]
         blocked: bool,
         #[pin]
-        delay: Delay,
+        delay: Timer,
     }
 }
 
@@ -35,7 +35,7 @@ impl<S: Stream> Throttle<S> {
             stream,
             duration,
             blocked: false,
-            delay: Delay::new(Duration::default()),
+            delay: Timer::after(Duration::default()),
         }
     }
 }
@@ -59,7 +59,7 @@ impl<S: Stream> Stream for Throttle<S> {
             Poll::Ready(None) => Poll::Ready(None),
             Poll::Ready(Some(v)) => {
                 *this.blocked = true;
-                this.delay.reset(*this.duration);
+                std::mem::replace(&mut *this.delay, Timer::after(*this.duration));
                 Poll::Ready(Some(v))
             }
         }

src/stream/stream/timeout.rs

@@ -4,8 +4,8 @@ use std::future::Future;
 use std::pin::Pin;
 use std::time::Duration;
 
-use futures_timer::Delay;
 use pin_project_lite::pin_project;
+use smol::Timer;
 
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
@@ -17,13 +17,13 @@ pin_project! {
         #[pin]
         stream: S,
         #[pin]
-        delay: Delay,
+        delay: Timer,
     }
 }
 
 impl<S: Stream> Timeout<S> {
     pub(crate) fn new(stream: S, dur: Duration) -> Self {
-        let delay = Delay::new(dur);
+        let delay = Timer::after(dur);
 
         Self { stream, delay }
     }