make Reactor::sleep return associated Reactor::Sleep type

Author: Keruspe

src/implementors/async_io.rs

@@ -1,8 +1,4 @@
-use crate::{
-    sys::AsSysFd,
-    traits::Reactor,
-    util::{IOHandle, UnitFuture},
-};
+use crate::{sys::AsSysFd, traits::Reactor, util::IOHandle};
 use async_io::{Async, Timer};
 use futures_core::Stream;
 use futures_io::{AsyncRead, AsyncWrite};
@@ -19,6 +15,7 @@ pub struct AsyncIO;
 
 impl Reactor for AsyncIO {
     type TcpStream = Async<TcpStream>;
+    type Sleep = Timer;
 
     fn register<H: Read + Write + AsSysFd + Send + 'static>(
         &self,
@@ -27,8 +24,8 @@ impl Reactor for AsyncIO {
         Async::new(IOHandle::new(socket))
     }
 
-    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> + Send + 'static {
-        UnitFuture(Timer::after(dur))
+    fn sleep(&self, dur: Duration) -> Self::Sleep {
+        Timer::after(dur)
     }
 
     fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> + Send + 'static {

src/implementors/noop.rs

@@ -9,7 +9,7 @@ use crate::{
 use futures_core::Stream;
 use futures_io::{AsyncRead, AsyncWrite};
 use std::{
-    future::Future,
+    future::{self, Future, Ready},
     io::{self, Read, Write},
     marker::PhantomData,
     net::SocketAddr,
@@ -60,6 +60,7 @@ impl Executor for Noop {
 
 impl Reactor for Noop {
     type TcpStream = DummyIO;
+    type Sleep = Ready<()>;
 
     fn register<H: Read + Write + AsSysFd + Send + 'static>(
         &self,
@@ -68,8 +69,8 @@ impl Reactor for Noop {
         Ok(DummyIO)
     }
 
-    fn sleep(&self, _dur: Duration) -> impl Future<Output = ()> + Send + 'static {
-        async {}
+    fn sleep(&self, _dur: Duration) -> Self::Sleep {
+        future::ready(())
     }
 
     fn interval(&self, _dur: Duration) -> impl Stream<Item = Instant> + Send + 'static {

src/implementors/smol.rs

@@ -4,7 +4,7 @@ use crate::{
     Runtime,
     sys::AsSysFd,
     traits::{Executor, Reactor, RuntimeKit},
-    util::{IOHandle, Task, UnitFuture},
+    util::{IOHandle, Task},
 };
 use futures_core::Stream;
 use futures_io::{AsyncRead, AsyncWrite};
@@ -58,6 +58,7 @@ impl Executor for Smol {
 
 impl Reactor for Smol {
     type TcpStream = Async<TcpStream>;
+    type Sleep = Timer;
 
     fn register<H: Read + Write + AsSysFd + Send + 'static>(
         &self,
@@ -66,8 +67,8 @@ impl Reactor for Smol {
         Async::new(IOHandle::new(socket))
     }
 
-    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> + Send + 'static {
-        UnitFuture(Timer::after(dur))
+    fn sleep(&self, dur: Duration) -> Self::Sleep {
+        Timer::after(dur)
     }
 
     fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> + Send + 'static {

src/implementors/tokio.rs

@@ -22,6 +22,7 @@ use std::{
 use tokio::{
     net::TcpStream,
     runtime::{EnterGuard, Handle, Runtime as TokioRT},
+    time::Sleep,
 };
 use tokio_stream::{StreamExt, wrappers::IntervalStream};
 
@@ -137,6 +138,7 @@ impl Executor for Tokio {
 
 impl Reactor for Tokio {
     type TcpStream = Compat<TcpStream>;
+    type Sleep = Sleep;
 
     fn register<H: Read + Write + AsSysFd + Send + 'static>(
         &self,
@@ -156,7 +158,7 @@ impl Reactor for Tokio {
         }
     }
 
-    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> + Send + 'static {
+    fn sleep(&self, dur: Duration) -> Self::Sleep {
         let _enter = self.enter();
         tokio::time::sleep(dur)
     }

src/runtime.rs

@@ -69,6 +69,7 @@ impl<RK: RuntimeKit> Executor for Runtime<RK> {
 
 impl<RK: RuntimeKit> Reactor for Runtime<RK> {
     type TcpStream = <RK as Reactor>::TcpStream;
+    type Sleep = <RK as Reactor>::Sleep;
 
     fn register<H: Read + Write + AsSysFd + Send + 'static>(
         &self,
@@ -77,7 +78,7 @@ impl<RK: RuntimeKit> Reactor for Runtime<RK> {
         self.kit.register(socket)
     }
 
-    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> + Send + 'static {
+    fn sleep(&self, dur: Duration) -> Self::Sleep {
         self.kit.sleep(dur)
     }
 

src/traits/reactor.rs

@@ -15,6 +15,9 @@ pub trait Reactor {
     /// The type representing a TCP stream (after tcp_connect) for this reactor
     type TcpStream: AsyncRead + AsyncWrite + Send + Unpin + 'static;
 
+    /// The type representing a Sleep for this reactor
+    type Sleep: Future + Send + 'static;
+
     /// Register a synchronous handle, returning an asynchronous one
     fn register<H: Read + Write + AsSysFd + Send + 'static>(
         &self,
@@ -24,7 +27,7 @@ pub trait Reactor {
         Self: Sized;
 
     /// Sleep for the given duration
-    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> + Send + 'static
+    fn sleep(&self, dur: Duration) -> Self::Sleep
     where
         Self: Sized;
 
@@ -69,6 +72,7 @@ where
     R::Target: Reactor + Sized,
 {
     type TcpStream = <<R as Deref>::Target as Reactor>::TcpStream;
+    type Sleep = <<R as Deref>::Target as Reactor>::Sleep;
 
     fn register<H: Read + Write + AsSysFd + Send + 'static>(
         &self,
@@ -77,7 +81,7 @@ where
         self.deref().register(socket)
     }
 
-    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> + Send + 'static {
+    fn sleep(&self, dur: Duration) -> Self::Sleep {
         self.deref().sleep(dur)
     }
 

src/util/future.rs

@@ -9,9 +9,6 @@ pub use block_on::*;
 mod dummy;
 pub use dummy::*;
 
-mod future;
-pub use future::*;
-
 #[cfg(feature = "async-io")]
 mod io;
 #[cfg(feature = "async-io")]

src/util/mod.rs

@@ -53,6 +53,7 @@ impl<E: Executor, R: Reactor> Executor for RuntimeParts<E, R> {
 
 impl<E: Executor, R: Reactor> Reactor for RuntimeParts<E, R> {
     type TcpStream = R::TcpStream;
+    type Sleep = R::Sleep;
 
     fn register<H: Read + Write + AsSysFd + Send + 'static>(
         &self,
@@ -61,7 +62,7 @@ impl<E: Executor, R: Reactor> Reactor for RuntimeParts<E, R> {
         self.reactor.register(socket)
     }
 
-    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> + Send + 'static {
+    fn sleep(&self, dur: Duration) -> Self::Sleep {
         self.reactor.sleep(dur)
     }