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).

1 files changed, 5 insertions, 6 deletions

diff --git a/tokio/tests/time_rt.rs b/tokio/tests/time_rt.rs
index ecce72d6..2576db4a 100644
--- a/tokio/tests/time_rt.rs
+++ b/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);
 }