Limit futures dependency to Stream via feature flag (#1774)

In an effort to reach API stability, the `tokio` crate is shedding its
_public_ dependencies on crates that are either a) do not provide a
stable (1.0+) release with longevity guarantees or b) match the `tokio`
release cadence. Of course, implementing `std` traits fits the
requirements.

The on exception, for now, is the `Stream` trait found in `futures_core`.
It is expected that this trait will not change much and be moved into `std.
Since Tokio is not yet going reaching 1.0, I feel that it is acceptable to maintain
a dependency on this trait given how foundational it is.

Since the `Stream` implementation is optional, types that are logically
streams provide `async fn next_*` functions to obtain the next value.
Avoiding the `next()` name prevents fn conflicts with `StreamExt::next()`.

Additionally, some misc cleanup is also done:

- `tokio::io::io` -> `tokio::io::util`.
- `delay` -> `delay_until`.
- `Timeout::new` -> `timeout(...)`.
- `signal::ctrl_c()` returns a future instead of a stream.
- `{tcp,unix}::Incoming` is removed (due to lack of `Stream` trait).
- `time::Throttle` is removed (due to lack of `Stream` trait).
-  Fix: `mpsc::UnboundedSender::send(&self)` (no more conflict with `Sink` fns).

tokio/tests/fs_dir.rs

@@ -3,8 +3,6 @@
 use tokio::fs;
 use tokio_test::assert_ok;
 
-use futures_util::future;
-use futures_util::stream::TryStreamExt;
 use std::sync::{Arc, Mutex};
 use tempfile::tempdir;
 
@@ -42,7 +40,7 @@ async fn remove() {
 }
 
 #[tokio::test]
-async fn read() {
+async fn read_inherent() {
     let base_dir = tempdir().unwrap();
 
     let p = base_dir.path();
@@ -55,15 +53,44 @@ async fn read() {
     let f = files.clone();
     let p = p.to_path_buf();
 
-    let read_dir_fut = fs::read_dir(p).await.unwrap();
-    read_dir_fut
-        .try_for_each(move |e| {
-            let s = e.file_name().to_str().unwrap().to_string();
-            f.lock().unwrap().push(s);
-            future::ok(())
-        })
-        .await
-        .unwrap();
+    let mut entries = fs::read_dir(p).await.unwrap();
+
+    while let Some(e) = assert_ok!(entries.next_entry().await) {
+        let s = e.file_name().to_str().unwrap().to_string();
+        f.lock().unwrap().push(s);
+    }
+
+    let mut files = files.lock().unwrap();
+    files.sort(); // because the order is not guaranteed
+    assert_eq!(
+        *files,
+        vec!["aa".to_string(), "bb".to_string(), "cc".to_string()]
+    );
+}
+
+#[tokio::test]
+async fn read_stream() {
+    use futures::StreamExt;
+
+    let base_dir = tempdir().unwrap();
+
+    let p = base_dir.path();
+    std::fs::create_dir(p.join("aa")).unwrap();
+    std::fs::create_dir(p.join("bb")).unwrap();
+    std::fs::create_dir(p.join("cc")).unwrap();
+
+    let files = Arc::new(Mutex::new(Vec::new()));
+
+    let f = files.clone();
+    let p = p.to_path_buf();
+
+    let mut entries = fs::read_dir(p).await.unwrap();
+
+    while let Some(res) = entries.next().await {
+        let e = assert_ok!(res);
+        let s = e.file_name().to_str().unwrap().to_string();
+        f.lock().unwrap().push(s);
+    }
 
     let mut files = files.lock().unwrap();
     files.sort(); // because the order is not guaranteed

tokio/tests/fs_file_mocked.rs

@@ -1,5 +1,16 @@
 #![warn(rust_2018_idioms)]
 
+macro_rules! ready {
+    ($e:expr $(,)?) => {
+        match $e {
+            std::task::Poll::Ready(t) => t,
+            std::task::Poll::Pending => return std::task::Poll::Pending,
+        }
+    };
+}
+
+use futures::future;
+
 // Load source
 #[allow(warnings)]
 #[path = "../src/fs/file.rs"]

tokio/tests/io_lines.rs

@@ -3,10 +3,24 @@
 use tokio::io::AsyncBufReadExt;
 use tokio_test::assert_ok;
 
-use futures_util::StreamExt;
+#[tokio::test]
+async fn lines_inherent() {
+    let rd: &[u8] = b"hello\r\nworld\n\n";
+    let mut st = rd.lines();
+
+    let b = assert_ok!(st.next_line().await).unwrap();
+    assert_eq!(b, "hello");
+    let b = assert_ok!(st.next_line().await).unwrap();
+    assert_eq!(b, "world");
+    let b = assert_ok!(st.next_line().await).unwrap();
+    assert_eq!(b, "");
+    assert!(assert_ok!(st.next_line().await).is_none());
+}
 
 #[tokio::test]
-async fn lines() {
+async fn lines_stream() {
+    use futures::StreamExt;
+
     let rd: &[u8] = b"hello\r\nworld\n\n";
     let mut st = rd.lines();
 

tokio/tests/net_driver.rs

@@ -4,7 +4,7 @@ use tokio::net::driver::Reactor;
 use tokio::net::TcpListener;
 use tokio_test::{assert_ok, assert_pending};
 
-use futures_util::task::{waker_ref, ArcWake};
+use futures::task::{waker_ref, ArcWake};
 use std::future::Future;
 use std::net::TcpStream;
 use std::pin::Pin;

tokio/tests/process_issue_42.rs

@@ -5,8 +5,8 @@
 use tokio::process::Command;
 use tokio::runtime;
 
-use futures_util::future::FutureExt;
-use futures_util::stream::FuturesOrdered;
+use futures::future::FutureExt;
+use futures::stream::FuturesOrdered;
 use std::process::Stdio;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;

tokio/tests/rt_common.rs

@@ -41,7 +41,7 @@ rt_test! {
     use tokio::time;
     use tokio_test::{assert_err, assert_ok};
 
-    use futures_util::future::poll_fn;
+    use futures::future::poll_fn;
     use std::future::Future;
     use std::pin::Pin;
     use std::sync::{mpsc, Arc};
@@ -133,12 +133,12 @@ rt_test! {
             let mut txs = (0..ITER)
                 .map(|i| {
                     let (tx, rx) = oneshot::channel();
-                    let mut done_tx = done_tx.clone();
+                    let done_tx = done_tx.clone();
 
                     tokio::spawn(async move {
                         let msg = assert_ok!(rx.await);
                         assert_eq!(i, msg);
-                        assert_ok!(done_tx.try_send(msg));
+                        assert_ok!(done_tx.send(msg));
                     });
 
                     tx

tokio/tests/signal_ctrl_c.rs

@@ -6,13 +6,13 @@ mod support {
 }
 use support::signal::send_signal;
 
-use tokio::prelude::*;
 use tokio::signal;
 use tokio::sync::oneshot;
+use tokio_test::assert_ok;
 
 #[tokio::test]
 async fn ctrl_c() {
-    let ctrl_c = signal::ctrl_c().expect("failed to init ctrl_c");
+    let ctrl_c = signal::ctrl_c();
 
     let (fire, wait) = oneshot::channel();
 
@@ -24,5 +24,6 @@ async fn ctrl_c() {
     });
 
     let _ = fire.send(());
-    let _ = ctrl_c.into_future().await;
+
+    assert_ok!(ctrl_c.await);
 }

tokio/tests/signal_drop_recv.rs

@@ -6,7 +6,6 @@ mod support {
 }
 use support::signal::send_signal;
 
-use tokio::prelude::*;
 use tokio::signal::unix::{signal, SignalKind};
 
 #[tokio::test]
@@ -16,7 +15,7 @@ async fn drop_then_get_a_signal() {
     drop(sig);
 
     send_signal(libc::SIGUSR1);
-    let sig = signal(kind).expect("failed to create second signal");
+    let mut sig = signal(kind).expect("failed to create second signal");
 
-    let _ = sig.into_future().await;
+    let _ = sig.recv().await;
 }

tokio/tests/signal_drop_rt.rs

@@ -6,7 +6,6 @@ mod support {
 }
 use support::signal::send_signal;
 
-use tokio::prelude::*;
 use tokio::runtime::Runtime;
 use tokio::signal::unix::{signal, SignalKind};
 
@@ -25,7 +24,7 @@ fn dropping_loops_does_not_cause_starvation() {
     send_signal(libc::SIGUSR1);
 
     first_rt
-        .block_on(first_signal.next())
+        .block_on(first_signal.recv())
         .expect("failed to await first signal");
 
     drop(first_rt);
@@ -33,7 +32,7 @@ fn dropping_loops_does_not_cause_starvation() {
 
     send_signal(libc::SIGUSR1);
 
-    second_rt.block_on(second_signal.next());
+    second_rt.block_on(second_signal.recv());
 }
 
 fn rt() -> Runtime {

tokio/tests/signal_drop_signal.rs

@@ -6,7 +6,6 @@ mod support {
 }
 use support::signal::send_signal;
 
-use tokio::prelude::*;
 use tokio::signal::unix::{signal, SignalKind};
 
 #[tokio::test]
@@ -15,12 +14,12 @@ async fn dropping_signal_does_not_deregister_any_other_instances() {
 
     // Signals should not starve based on ordering
     let first_duplicate_signal = signal(kind).expect("failed to register first duplicate signal");
-    let sig = signal(kind).expect("failed to register signal");
+    let mut sig = signal(kind).expect("failed to register signal");
     let second_duplicate_signal = signal(kind).expect("failed to register second duplicate signal");
 
     drop(first_duplicate_signal);
     drop(second_duplicate_signal);
 
     send_signal(libc::SIGUSR1);
-    let _ = sig.into_future().await;
+    let _ = sig.recv().await;
 }

tokio/tests/signal_multi_rt.rs

@@ -6,7 +6,6 @@ mod support {
 }
 use support::signal::send_signal;
 
-use tokio::prelude::*;
 use tokio::runtime::Runtime;
 use tokio::signal::unix::{signal, SignalKind};
 
@@ -26,9 +25,9 @@ fn multi_loop() {
                 thread::spawn(move || {
                     let mut rt = rt();
                     let _ = rt.block_on(async {
-                        let signal = signal(SignalKind::hangup()).unwrap();
+                        let mut signal = signal(SignalKind::hangup()).unwrap();
                         sender.send(()).unwrap();
-                        signal.into_future().await
+                        signal.recv().await
                     });
                 })
             })

tokio/tests/signal_notify_both.rs

@@ -6,18 +6,17 @@ mod support {
 }
 use support::signal::send_signal;
 
-use tokio::prelude::*;
 use tokio::signal::unix::{signal, SignalKind};
 
-use futures::future;
-
 #[tokio::test]
 async fn notify_both() {
     let kind = SignalKind::user_defined2();
-    let signal1 = signal(kind).expect("failed to create signal1");
 
-    let signal2 = signal(kind).expect("failed to create signal2");
+    let mut signal1 = signal(kind).expect("failed to create signal1");
+    let mut signal2 = signal(kind).expect("failed to create signal2");
 
     send_signal(libc::SIGUSR2);
-    let _ = future::join(signal1.into_future(), signal2.into_future()).await;
+
+    signal1.recv().await;
+    signal2.recv().await;
 }

tokio/tests/signal_twice.rs

@@ -6,7 +6,6 @@ mod support {
 }
 use support::signal::send_signal;
 
-use tokio::prelude::*;
 use tokio::signal::unix::{signal, SignalKind};
 
 #[tokio::test]
@@ -17,9 +16,6 @@ async fn twice() {
     for _ in 0..2 {
         send_signal(libc::SIGUSR1);
 
-        let (item, sig_next) = sig.into_future().await;
-        assert_eq!(item, Some(()));
-
-        sig = sig_next;
+        assert!(sig.recv().await.is_some());
     }
 }

tokio/tests/signal_usr1.rs

@@ -6,18 +6,17 @@ mod support {
 }
 use support::signal::send_signal;
 
-use tokio::prelude::*;
 use tokio::signal::unix::{signal, SignalKind};
 use tokio_test::assert_ok;
 
 #[tokio::test]
 async fn signal_usr1() {
-    let signal = assert_ok!(
+    let mut signal = assert_ok!(
         signal(SignalKind::user_defined1()),
         "failed to create signal"
     );
 
     send_signal(libc::SIGUSR1);
 
-    let _ = signal.into_future().await;
+    signal.recv().await;
 }

tokio/tests/sync_errors.rs

@@ -6,11 +6,8 @@ fn is_error<T: ::std::error::Error + Send + Sync>() {}
 fn mpsc_error_bound() {
     use tokio::sync::mpsc::error;
 
-    is_error::<error::SendError>();
+    is_error::<error::SendError<()>>();
     is_error::<error::TrySendError<()>>();
-    is_error::<error::UnboundedRecvError>();
-    is_error::<error::UnboundedSendError>();
-    is_error::<error::UnboundedTrySendError<()>>();
 }
 
 #[test]

tokio/tests/sync_mpsc.rs

@@ -1,6 +1,7 @@
 #![warn(rust_2018_idioms)]
 
 use tokio::sync::mpsc;
+use tokio::sync::mpsc::error::TrySendError;
 use tokio_test::task;
 use tokio_test::{
     assert_err, assert_ok, assert_pending, assert_ready, assert_ready_err, assert_ready_ok,
@@ -37,6 +38,22 @@ fn send_recv_with_buffer() {
     assert!(val.is_none());
 }
 
+#[tokio::test]
+async fn send_recv_stream_with_buffer() {
+    use futures::StreamExt;
+
+    let (mut tx, mut rx) = mpsc::channel::<i32>(16);
+
+    tokio::spawn(async move {
+        assert_ok!(tx.send(1).await);
+        assert_ok!(tx.send(2).await);
+    });
+
+    assert_eq!(Some(1), rx.next().await);
+    assert_eq!(Some(2), rx.next().await);
+    assert_eq!(None, rx.next().await);
+}
+
 #[tokio::test]
 async fn async_send_recv_with_buffer() {
     let (mut tx, mut rx) = mpsc::channel(16);
@@ -51,36 +68,6 @@ async fn async_send_recv_with_buffer() {
     assert_eq!(None, rx.recv().await);
 }
 
-#[test]
-fn send_sink_recv_with_buffer() {
-    use futures_core::Stream;
-    use futures_sink::Sink;
-
-    let (tx, rx) = mpsc::channel::<i32>(16);
-
-    task::spawn(tx).enter(|cx, mut tx| {
-        assert_ready_ok!(tx.as_mut().poll_ready(cx));
-        assert_ok!(tx.as_mut().start_send(1));
-
-        assert_ready_ok!(tx.as_mut().poll_ready(cx));
-        assert_ok!(tx.as_mut().start_send(2));
-
-        assert_ready_ok!(tx.as_mut().poll_flush(cx));
-        assert_ready_ok!(tx.as_mut().poll_close(cx));
-    });
-
-    task::spawn(rx).enter(|cx, mut rx| {
-        let val = assert_ready!(rx.as_mut().poll_next(cx));
-        assert_eq!(val, Some(1));
-
-        let val = assert_ready!(rx.as_mut().poll_next(cx));
-        assert_eq!(val, Some(2));
-
-        let val = assert_ready!(rx.as_mut().poll_next(cx));
-        assert!(val.is_none());
-    });
-}
-
 #[test]
 fn start_send_past_cap() {
     let mut t1 = task::spawn(());
@@ -124,11 +111,11 @@ fn buffer_gteq_one() {
 fn send_recv_unbounded() {
     let mut t1 = task::spawn(());
 
-    let (mut tx, mut rx) = mpsc::unbounded_channel::<i32>();
+    let (tx, mut rx) = mpsc::unbounded_channel::<i32>();
 
     // Using `try_send`
-    assert_ok!(tx.try_send(1));
-    assert_ok!(tx.try_send(2));
+    assert_ok!(tx.send(1));
+    assert_ok!(tx.send(2));
 
     let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
     assert_eq!(val, Some(1));
@@ -144,11 +131,11 @@ fn send_recv_unbounded() {
 
 #[tokio::test]
 async fn async_send_recv_unbounded() {
-    let (mut tx, mut rx) = mpsc::unbounded_channel();
+    let (tx, mut rx) = mpsc::unbounded_channel();
 
     tokio::spawn(async move {
-        assert_ok!(tx.try_send(1));
-        assert_ok!(tx.try_send(2));
+        assert_ok!(tx.send(1));
+        assert_ok!(tx.send(2));
     });
 
     assert_eq!(Some(1), rx.recv().await);
@@ -156,41 +143,20 @@ async fn async_send_recv_unbounded() {
     assert_eq!(None, rx.recv().await);
 }
 
-#[test]
-fn sink_send_recv_unbounded() {
-    use futures_core::Stream;
-    use futures_sink::Sink;
-    use futures_util::pin_mut;
+#[tokio::test]
+async fn send_recv_stream_unbounded() {
+    use futures::StreamExt;
 
-    let mut t1 = task::spawn(());
+    let (tx, mut rx) = mpsc::unbounded_channel::<i32>();
 
-    let (tx, rx) = mpsc::unbounded_channel::<i32>();
-
-    t1.enter(|cx, _| {
-        pin_mut!(tx);
-
-        assert_ready_ok!(tx.as_mut().poll_ready(cx));
-        assert_ok!(tx.as_mut().start_send(1));
-
-        assert_ready_ok!(tx.as_mut().poll_ready(cx));
-        assert_ok!(tx.as_mut().start_send(2));
-
-        assert_ready_ok!(tx.as_mut().poll_flush(cx));
-        assert_ready_ok!(tx.as_mut().poll_close(cx));
+    tokio::spawn(async move {
+        assert_ok!(tx.send(1));
+        assert_ok!(tx.send(2));
     });
 
-    t1.enter(|cx, _| {
-        pin_mut!(rx);
-
-        let val = assert_ready!(rx.as_mut().poll_next(cx));
-        assert_eq!(val, Some(1));
-
-        let val = assert_ready!(rx.as_mut().poll_next(cx));
-        assert_eq!(val, Some(2));
-
-        let val = assert_ready!(rx.as_mut().poll_next(cx));
-        assert!(val.is_none());
-    });
+    assert_eq!(Some(1), rx.next().await);
+    assert_eq!(Some(2), rx.next().await);
+    assert_eq!(None, rx.next().await);
 }
 
 #[test]
@@ -223,7 +189,7 @@ fn no_t_bounds_unbounded() {
     // same with Receiver
     println!("{:?}", rx);
     // and sender should be Clone even though T isn't Clone
-    assert!(tx.clone().try_send(NoImpls).is_ok());
+    assert!(tx.clone().send(NoImpls).is_ok());
 
     let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
     assert!(val.is_some());
@@ -356,8 +322,10 @@ fn try_send_fail() {
     tx.try_send("hello").unwrap();
 
     // This should fail
-    let err = assert_err!(tx.try_send("fail"));
-    assert!(err.is_full());
+    match assert_err!(tx.try_send("fail")) {
+        TrySendError::Full(..) => {}
+        _ => panic!(),
+    }
 
     let val = assert_ready!(t1.enter(|cx, _| rx.poll_recv(cx)));
     assert_eq!(val, Some("hello"));
@@ -421,7 +389,10 @@ fn dropping_rx_closes_channel_for_try() {
 
     {
         let err = assert_err!(tx.try_send(msg.clone()));
-        assert!(err.is_closed());
+        match err {
+            TrySendError::Closed(..) => {}
+            _ => panic!(),
+        }
     }
 
     assert_eq!(1, Arc::strong_count(&msg));

tokio/tests/sync_watch.rs

@@ -4,41 +4,6 @@ use tokio::sync::watch;
 use tokio_test::task::spawn;
 use tokio_test::{assert_pending, assert_ready};
 
-#[test]
-fn single_rx_recv_ref() {
-    let (tx, mut rx) = watch::channel("one");
-
-    {
-        let mut t = spawn(rx.recv_ref());
-        let v = assert_ready!(t.poll()).unwrap();
-        assert_eq!(*v, "one");
-    }
-
-    {
-        let mut t = spawn(rx.recv_ref());
-
-        assert_pending!(t.poll());
-
-        tx.broadcast("two").unwrap();
-
-        assert!(t.is_woken());
-
-        let v = assert_ready!(t.poll()).unwrap();
-        assert_eq!(*v, "two");
-    }
-
-    {
-        let mut t = spawn(rx.recv_ref());
-
-        assert_pending!(t.poll());
-
-        drop(tx);
-
-        let res = assert_ready!(t.poll());
-        assert!(res.is_none());
-    }
-}
-
 #[test]
 fn single_rx_recv() {
     let (tx, mut rx) = watch::channel("one");
@@ -74,9 +39,161 @@ fn single_rx_recv() {
     }
 }
 
+#[test]
+fn multi_rx() {
+    let (tx, mut rx1) = watch::channel("one");
+    let mut rx2 = rx1.clone();
+
+    {
+        let mut t1 = spawn(rx1.recv());
+        let mut t2 = spawn(rx2.recv());
+
+        let res = assert_ready!(t1.poll());
+        assert_eq!(res.unwrap(), "one");
+
+        let res = assert_ready!(t2.poll());
+        assert_eq!(res.unwrap(), "one");
+    }
+
+    let mut t2 = spawn(rx2.recv());
+
+    {
+        let mut t1 = spawn(rx1.recv());
+
+        assert_pending!(t1.poll());
+        assert_pending!(t2.poll());
+
+        tx.broadcast("two").unwrap();
+
+        assert!(t1.is_woken());
+        assert!(t2.is_woken());
+
+        let res = assert_ready!(t1.poll());
+        assert_eq!(res.unwrap(), "two");
+    }
+
+    {
+        let mut t1 = spawn(rx1.recv());
+
+        assert_pending!(t1.poll());
+
+        tx.broadcast("three").unwrap();
+
+        assert!(t1.is_woken());
+        assert!(t2.is_woken());
+
+        let res = assert_ready!(t1.poll());
+        assert_eq!(res.unwrap(), "three");
+
+        let res = assert_ready!(t2.poll());
+        assert_eq!(res.unwrap(), "three");
+    }
+
+    drop(t2);
+
+    {
+        let mut t1 = spawn(rx1.recv());
+        let mut t2 = spawn(rx2.recv());
+
+        assert_pending!(t1.poll());
+        assert_pending!(t2.poll());
+
+        tx.broadcast("four").unwrap();
+
+        let res = assert_ready!(t1.poll());
+        assert_eq!(res.unwrap(), "four");
+        drop(t1);
+
+        let mut t1 = spawn(rx1.recv());
+        assert_pending!(t1.poll());
+
+        drop(tx);
+
+        assert!(t1.is_woken());
+        let res = assert_ready!(t1.poll());
+        assert!(res.is_none());
+
+        let res = assert_ready!(t2.poll());
+        assert_eq!(res.unwrap(), "four");
+
+        drop(t2);
+        let mut t2 = spawn(rx2.recv());
+        let res = assert_ready!(t2.poll());
+        assert!(res.is_none());
+    }
+}
+
+#[test]
+fn rx_observes_final_value() {
+    // Initial value
+
+    let (tx, mut rx) = watch::channel("one");
+    drop(tx);
+
+    {
+        let mut t1 = spawn(rx.recv());
+        let res = assert_ready!(t1.poll());
+        assert_eq!(res.unwrap(), "one");
+    }
+
+    {
+        let mut t1 = spawn(rx.recv());
+        let res = assert_ready!(t1.poll());
+        assert!(res.is_none());
+    }
+
+    // Sending a value
+
+    let (tx, mut rx) = watch::channel("one");
+
+    tx.broadcast("two").unwrap();
+
+    {
+        let mut t1 = spawn(rx.recv());
+        let res = assert_ready!(t1.poll());
+        assert_eq!(res.unwrap(), "two");
+    }
+
+    {
+        let mut t1 = spawn(rx.recv());
+        assert_pending!(t1.poll());
+
+        tx.broadcast("three").unwrap();
+        drop(tx);
+
+        assert!(t1.is_woken());
+
+        let res = assert_ready!(t1.poll());
+        assert_eq!(res.unwrap(), "three");
+    }
+
+    {
+        let mut t1 = spawn(rx.recv());
+        let res = assert_ready!(t1.poll());
+        assert!(res.is_none());
+    }
+}
+
+#[test]
+fn poll_close() {
+    let (mut tx, rx) = watch::channel("one");
+
+    {
+        let mut t = spawn(tx.closed());
+        assert_pending!(t.poll());
+
+        drop(rx);
+
+        assert!(t.is_woken());
+        assert_ready!(t.poll());
+    }
+
+    assert!(tx.broadcast("two").is_err());
+}
+
 #[test]
 fn stream_impl() {
-    use tokio::prelude::*;
+    use futures::StreamExt;
 
     let (tx, mut rx) = watch::channel("one");
 
@@ -110,155 +227,3 @@ fn stream_impl() {
         assert!(res.is_none());
     }
 }
-
-#[test]
-fn multi_rx() {
-    let (tx, mut rx1) = watch::channel("one");
-    let mut rx2 = rx1.clone();
-
-    {
-        let mut t1 = spawn(rx1.recv_ref());
-        let mut t2 = spawn(rx2.recv_ref());
-
-        let res = assert_ready!(t1.poll());
-        assert_eq!(*res.unwrap(), "one");
-
-        let res = assert_ready!(t2.poll());
-        assert_eq!(*res.unwrap(), "one");
-    }
-
-    let mut t2 = spawn(rx2.recv_ref());
-
-    {
-        let mut t1 = spawn(rx1.recv_ref());
-
-        assert_pending!(t1.poll());
-        assert_pending!(t2.poll());
-
-        tx.broadcast("two").unwrap();
-
-        assert!(t1.is_woken());
-        assert!(t2.is_woken());
-
-        let res = assert_ready!(t1.poll());
-        assert_eq!(*res.unwrap(), "two");
-    }
-
-    {
-        let mut t1 = spawn(rx1.recv_ref());
-
-        assert_pending!(t1.poll());
-
-        tx.broadcast("three").unwrap();
-
-        assert!(t1.is_woken());
-        assert!(t2.is_woken());
-
-        let res = assert_ready!(t1.poll());
-        assert_eq!(*res.unwrap(), "three");
-
-        let res = assert_ready!(t2.poll());
-        assert_eq!(*res.unwrap(), "three");
-    }
-
-    drop(t2);
-
-    {
-        let mut t1 = spawn(rx1.recv_ref());
-        let mut t2 = spawn(rx2.recv_ref());
-
-        assert_pending!(t1.poll());
-        assert_pending!(t2.poll());
-
-        tx.broadcast("four").unwrap();
-
-        let res = assert_ready!(t1.poll());
-        assert_eq!(*res.unwrap(), "four");
-        drop(t1);
-
-        let mut t1 = spawn(rx1.recv_ref());
-        assert_pending!(t1.poll());
-
-        drop(tx);
-
-        assert!(t1.is_woken());
-        let res = assert_ready!(t1.poll());
-        assert!(res.is_none());
-
-        let res = assert_ready!(t2.poll());
-        assert_eq!(*res.unwrap(), "four");
-
-        drop(t2);
-        let mut t2 = spawn(rx2.recv_ref());
-        let res = assert_ready!(t2.poll());
-        assert!(res.is_none());
-    }
-}
-
-#[test]
-fn rx_observes_final_value() {
-    // Initial value
-
-    let (tx, mut rx) = watch::channel("one");
-    drop(tx);
-
-    {
-        let mut t1 = spawn(rx.recv_ref());
-        let res = assert_ready!(t1.poll());
-        assert_eq!(*res.unwrap(), "one");
-    }
-
-    {
-        let mut t1 = spawn(rx.recv_ref());
-        let res = assert_ready!(t1.poll());
-        assert!(res.is_none());
-    }
-
-    // Sending a value
-
-    let (tx, mut rx) = watch::channel("one");
-
-    tx.broadcast("two").unwrap();
-
-    {
-        let mut t1 = spawn(rx.recv_ref());
-        let res = assert_ready!(t1.poll());
-        assert_eq!(*res.unwrap(), "two");
-    }
-
-    {
-        let mut t1 = spawn(rx.recv_ref());
-        assert_pending!(t1.poll());
-
-        tx.broadcast("three").unwrap();
-        drop(tx);
-
-        assert!(t1.is_woken());
-
-        let res = assert_ready!(t1.poll());
-        assert_eq!(*res.unwrap(), "three");
-    }
-
-    {
-        let mut t1 = spawn(rx.recv_ref());
-        let res = assert_ready!(t1.poll());
-        assert!(res.is_none());
-    }
-}
-
-#[test]
-fn poll_close() {
-    let (mut tx, rx) = watch::channel("one");
-
-    {
-        let mut t = spawn(tx.closed());
-        assert_pending!(t.poll());
-
-        drop(rx);
-
-        assert!(t.is_woken());
-        assert_ready!(t.poll());
-    }
-
-    assert!(tx.broadcast("two").is_err());
-}

tokio/tests/time_interval.rs

@@ -1,12 +1,14 @@
 #![warn(rust_2018_idioms)]
 
-use tokio::time::{self, Duration, Instant, Interval};
+use tokio::time::{self, Duration, Instant};
 use tokio_test::{assert_pending, assert_ready_eq, task};
 
+use std::task::Poll;
+
 #[tokio::test]
 #[should_panic]
 async fn interval_zero_duration() {
-    let _ = Interval::new(Instant::now(), ms(0));
+    let _ = time::interval_at(Instant::now(), ms(0));
 }
 
 #[tokio::test]
@@ -18,26 +20,44 @@ async fn usage() {
     // TODO: Skip this
     time::advance(ms(1)).await;
 
-    let mut int = task::spawn(Interval::new(start, ms(300)));
+    let mut i = task::spawn(time::interval_at(start, ms(300)));
 
-    assert_ready_eq!(int.poll_next(), Some(start));
-    assert_pending!(int.poll_next());
+    assert_ready_eq!(poll_next(&mut i), start);
+    assert_pending!(poll_next(&mut i));
 
     time::advance(ms(100)).await;
-    assert_pending!(int.poll_next());
+    assert_pending!(poll_next(&mut i));
 
     time::advance(ms(200)).await;
-    assert_ready_eq!(int.poll_next(), Some(start + ms(300)));
-    assert_pending!(int.poll_next());
+    assert_ready_eq!(poll_next(&mut i), start + ms(300));
+    assert_pending!(poll_next(&mut i));
 
     time::advance(ms(400)).await;
-    assert_ready_eq!(int.poll_next(), Some(start + ms(600)));
-    assert_pending!(int.poll_next());
+    assert_ready_eq!(poll_next(&mut i), start + ms(600));
+    assert_pending!(poll_next(&mut i));
 
     time::advance(ms(500)).await;
-    assert_ready_eq!(int.poll_next(), Some(start + ms(900)));
-    assert_ready_eq!(int.poll_next(), Some(start + ms(1200)));
-    assert_pending!(int.poll_next());
+    assert_ready_eq!(poll_next(&mut i), start + ms(900));
+    assert_ready_eq!(poll_next(&mut i), start + ms(1200));
+    assert_pending!(poll_next(&mut i));
+}
+
+#[tokio::test]
+async fn usage_stream() {
+    use futures::StreamExt;
+
+    let start = Instant::now();
+    let mut interval = time::interval(ms(10));
+
+    for _ in 0..3 {
+        interval.next().await.unwrap();
+    }
+
+    assert!(start.elapsed() > ms(20));
+}
+
+fn poll_next(interval: &mut task::Spawn<time::Interval>) -> Poll<Instant> {
+    interval.enter(|cx, mut interval| interval.poll_tick(cx))
 }
 
 fn ms(n: u64) -> Duration {

tokio/tests/time_rt.rs

@@ -1,6 +1,5 @@
 #![warn(rust_2018_idioms)]
 
-use tokio::prelude::*;
 use tokio::time::*;
 
 use std::sync::mpsc;
@@ -15,7 +14,7 @@ fn timer_with_threaded_runtime() {
     rt.spawn(async move {
         let when = Instant::now() + Duration::from_millis(100);
 
-        delay(when).await;
+        delay_until(when).await;
         assert!(Instant::now() >= when);
 
         tx.send(()).unwrap();
@@ -34,7 +33,7 @@ fn timer_with_current_thread_runtime() {
     rt.block_on(async move {
         let when = Instant::now() + Duration::from_millis(100);
 
-        tokio::time::delay(when).await;
+        delay_until(when).await;
         assert!(Instant::now() >= when);
 
         tx.send(()).unwrap();
@@ -68,14 +67,14 @@ async fn starving() {
     }
 
     let when = Instant::now() + Duration::from_millis(20);
-    let starve = Starve(delay(when), 0);
+    let starve = Starve(delay_until(when), 0);
 
     starve.await;
     assert!(Instant::now() >= when);
 }
 
 #[tokio::test]
-async fn timeout() {
+async fn timeout_value() {
     use tokio::sync::oneshot;
 
     let (_tx, rx) = oneshot::channel::<()>();
@@ -83,7 +82,7 @@ async fn timeout() {
     let now = Instant::now();
     let dur = Duration::from_millis(20);
 
-    let res = rx.timeout(dur).await;
+    let res = timeout(dur, rx).await;
     assert!(res.is_err());
     assert!(Instant::now() >= now + dur);
 }

tokio/tests/time_throttle.rs

@@ -1,68 +0,0 @@
-#![warn(rust_2018_idioms)]
-
-use tokio::sync::mpsc;
-use tokio::time::throttle::Throttle;
-use tokio::time::Instant;
-use tokio_test::{assert_pending, assert_ready_eq};
-
-use futures::future::poll_fn;
-use futures::StreamExt;
-use std::task::Poll;
-use std::time::Duration;
-
-#[tokio::test]
-async fn throttle() {
-    let (mut tx, rx) = mpsc::unbounded_channel();
-    let mut stream = Throttle::new(rx, ms(1));
-
-    poll_fn(|cx| {
-        assert_pending!(stream.poll_next_unpin(cx));
-        Poll::Ready(())
-    })
-    .await;
-
-    for i in 0..3 {
-        tx.try_send(i).unwrap();
-    }
-
-    drop(tx);
-
-    let mut now = Instant::now();
-
-    while let Some(_) = stream.next().await {
-        assert!(Instant::now() >= now);
-        now += ms(1);
-    }
-}
-
-#[tokio::test]
-async fn throttle_dur_0() {
-    let (mut tx, rx) = mpsc::unbounded_channel();
-    let mut stream = Throttle::new(rx, ms(0));
-
-    poll_fn(|cx| {
-        assert_pending!(stream.poll_next_unpin(cx));
-
-        for i in 0..3 {
-            tx.try_send(i).unwrap();
-        }
-
-        Poll::Ready(())
-    })
-    .await;
-
-    poll_fn(|cx| {
-        for i in 0..3 {
-            assert_ready_eq!(stream.poll_next_unpin(cx), Some(i), "i = {}", i);
-        }
-
-        assert_pending!(stream.poll_next_unpin(cx));
-
-        Poll::Ready(())
-    })
-    .await;
-}
-
-fn ms(n: u64) -> Duration {
-    Duration::from_millis(n)
-}

tokio/tests/time_timeout.rs

@@ -1,7 +1,7 @@
 #![warn(rust_2018_idioms)]
 
 use tokio::sync::oneshot;
-use tokio::time::{self, Instant, Timeout};
+use tokio::time::{self, timeout, timeout_at, Instant};
 use tokio_test::*;
 
 use futures::future::pending;
@@ -10,7 +10,7 @@ use std::time::Duration;
 #[tokio::test]
 async fn simultaneous_deadline_future_completion() {
     // Create a future that is immediately ready
-    let mut fut = task::spawn(Timeout::new_at(async {}, Instant::now()));
+    let mut fut = task::spawn(timeout_at(Instant::now(), async {}));
 
     // Ready!
     assert_ready_ok!(fut.poll());
@@ -19,7 +19,7 @@ async fn simultaneous_deadline_future_completion() {
 #[tokio::test]
 async fn completed_future_past_deadline() {
     // Wrap it with a deadline
-    let mut fut = task::spawn(Timeout::new_at(async {}, Instant::now() - ms(1000)));
+    let mut fut = task::spawn(timeout_at(Instant::now() - ms(1000), async {}));
 
     // Ready!
     assert_ready_ok!(fut.poll());
@@ -33,7 +33,7 @@ async fn future_and_deadline_in_future() {
     let (tx, rx) = oneshot::channel();
 
     // Wrap it with a deadline
-    let mut fut = task::spawn(Timeout::new_at(rx, Instant::now() + ms(100)));
+    let mut fut = task::spawn(timeout_at(Instant::now() + ms(100), rx));
 
     assert_pending!(fut.poll());
 
@@ -57,7 +57,7 @@ async fn future_and_timeout_in_future() {
     let (tx, rx) = oneshot::channel();
 
     // Wrap it with a deadline
-    let mut fut = task::spawn(Timeout::new(rx, ms(100)));
+    let mut fut = task::spawn(timeout(ms(100), rx));
 
     // Ready!
     assert_pending!(fut.poll());
@@ -80,7 +80,7 @@ async fn deadline_now_elapses() {
     time::pause();
 
     // Wrap it with a deadline
-    let mut fut = task::spawn(Timeout::new_at(pending::<()>(), Instant::now()));
+    let mut fut = task::spawn(timeout_at(Instant::now(), pending::<()>()));
 
     // Factor in jitter
     // TODO: don't require this
@@ -94,7 +94,7 @@ async fn deadline_future_elapses() {
     time::pause();
 
     // Wrap it with a deadline
-    let mut fut = task::spawn(Timeout::new_at(pending::<()>(), Instant::now() + ms(300)));
+    let mut fut = task::spawn(timeout_at(Instant::now() + ms(300), pending::<()>()));
 
     assert_pending!(fut.poll());
 
@@ -104,63 +104,6 @@ async fn deadline_future_elapses() {
     assert_ready_err!(fut.poll());
 }
 
-#[tokio::test]
-async fn stream_and_timeout_in_future() {
-    use tokio::sync::mpsc;
-
-    time::pause();
-
-    // Not yet complete
-    let (mut tx, rx) = mpsc::unbounded_channel();
-
-    // Wrap it with a deadline
-    let mut stream = task::spawn(Timeout::new(rx, ms(100)));
-
-    // Not ready
-    assert_pending!(stream.poll_next());
-
-    // Turn the timer, it runs for the elapsed time
-    time::advance(ms(90)).await;
-
-    assert_pending!(stream.poll_next());
-
-    // Complete the future
-    tx.try_send(()).unwrap();
-
-    let item = assert_ready!(stream.poll_next());
-    assert!(item.is_some());
-}
-
-#[tokio::test]
-async fn idle_stream_timesout_periodically() {
-    use tokio::sync::mpsc;
-
-    time::pause();
-
-    // Not yet complete
-    let (_tx, rx) = mpsc::unbounded_channel::<()>();
-
-    // Wrap it with a deadline
-    let mut stream = task::spawn(Timeout::new(rx, ms(100)));
-
-    // Not ready
-    assert_pending!(stream.poll_next());
-
-    // Turn the timer, it runs for the elapsed time
-    time::advance(ms(101)).await;
-
-    let v = assert_ready!(stream.poll_next()).unwrap();
-    assert_err!(v);
-
-    // Stream's timeout should reset
-    assert_pending!(stream.poll_next());
-
-    // Turn the timer, it runs for the elapsed time
-    time::advance(ms(101)).await;
-    let v = assert_ready!(stream.poll_next()).unwrap();
-    assert_err!(v);
-}
-
 fn ms(n: u64) -> Duration {
     Duration::from_millis(n)
 }