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, 29 insertions, 25 deletions

diff --git a/tokio/src/time/delay.rs b/tokio/src/time/delay.rs
index 83ab3c8f..e3b605e7 100644
--- a/tokio/src/time/delay.rs
+++ b/tokio/src/time/delay.rs
@@ -1,25 +1,44 @@
 use crate::time::driver::Registration;
 use crate::time::{Duration, Instant};
 
-use futures_core::ready;
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{self, Poll};
 
-/// A future that completes at a specified instant in time.
+/// Wait until `deadline` is reached.
 ///
-/// Instances of `Delay` perform no work and complete with `()` once the
-/// specified deadline has been reached.
-///
-/// `Delay` has a resolution of one millisecond and should not be used for tasks
-/// that require high-resolution timers.
+/// No work is performed while awaiting on the delay to complete. The delay
+/// operates at millisecond granularity and should not be used for tasks that
+/// require high-resolution timers.
 ///
 /// # Cancellation
 ///
-/// Canceling a `Delay` is done by dropping the value. No additional cleanup or
-/// other work is required.
+/// Canceling a delay is done by dropping the returned future. No additional
+/// cleanup work is required.
+pub fn delay_until(deadline: Instant) -> Delay {
+    let registration = Registration::new(deadline, Duration::from_millis(0));
+    Delay { registration }
+}
+
+/// Wait until `duration` has elapsed.
+///
+/// Equivalent to `delay_until(Instant::now() + duration)`. An asynchronous
+/// analog to `std::thread::sleep`.
 ///
-/// [`new`]: #method.new
+/// No work is performed while awaiting on the delay to complete. The delay
+/// operates at millisecond granularity and should not be used for tasks that
+/// require high-resolution timers.
+///
+/// # Cancellation
+///
+/// Canceling a delay is done by dropping the returned future. No additional
+/// cleanup work is required.
+pub fn delay_for(duration: Duration) -> Delay {
+    delay_until(Instant::now() + duration)
+}
+
+/// Future returned by [`delay_until`](delay_until) and
+/// [`delay_for`](delay_for).
 #[derive(Debug)]
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct Delay {
@@ -30,17 +49,6 @@ pub struct Delay {
 }
 
 impl Delay {
-    /// Create a new `Delay` instance that elapses at `deadline`.
-    ///
-    /// Only millisecond level resolution is guaranteed. There is no guarantee
-    /// as to how the sub-millisecond portion of `deadline` will be handled.
-    /// `Delay` should not be used for high-resolution timer use cases.
-    pub(crate) fn new(deadline: Instant) -> Delay {
-        let registration = Registration::new(deadline, Duration::from_millis(0));
-
-        Delay { registration }
-    }
-
     pub(crate) fn new_timeout(deadline: Instant, duration: Duration) -> Delay {
         let registration = Registration::new(deadline, duration);
         Delay { registration }
@@ -69,10 +77,6 @@ impl Delay {
         self.registration.reset(deadline);
     }
 
-    pub(crate) fn reset_timeout(&mut self) {
-        self.registration.reset_timeout();
-    }
-
     /// Register the delay with the timer instance for the current execution
     /// context.
     fn register(&mut self) {