add Future::wait_until

Author: yoshuawuyts

Cargo.toml

@@ -39,6 +39,7 @@ slab = { version = "0.4.8", optional = true }
 smallvec = { version = "1.11.0", optional = true }
 
 [dev-dependencies]
+async-io = "2.3.2"
 async-std = { version = "1.12.0", features = ["attributes"] }
 criterion = { version = "0.3", features = [
     "async",

src/future/futures_ext.rs

@@ -5,6 +5,7 @@ use futures_core::Future;
 
 use super::join::tuple::Join2;
 use super::race::tuple::Race2;
+use super::WaitUntil;
 
 /// An extension trait for the `Future` trait.
 pub trait FutureExt: Future {
@@ -19,6 +20,40 @@ pub trait FutureExt: Future {
     where
         Self: Future<Output = T> + Sized,
         S2: IntoFuture<Output = T>;
+
+    /// Delay resolving the future until the given deadline.
+    ///
+    /// The underlying future will not be polled until the deadline has expired. In addition
+    /// to using a time source as a deadline, any future can be used as a
+    /// deadline too. When used in combination with a multi-consumer channel,
+    /// this method can be used to synchronize the start of multiple futures and streams.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// use async_io::Timer;
+    /// use futures_concurrency::prelude::*;
+    /// use futures_lite::future::block_on;
+    /// use std::time::{Duration, Instant};
+    ///
+    /// block_on(async {
+    ///     let now = Instant::now();
+    ///     let duration = Duration::from_millis(100);
+    ///
+    ///     async { "meow" }
+    ///         .wait_until(Timer::after(duration))
+    ///         .await;
+    ///
+    ///     assert!(now.elapsed() >= duration);
+    /// });
+    /// ```
+    fn wait_until<D>(self, deadline: D) -> WaitUntil<Self, D::IntoFuture>
+    where
+        Self: Sized,
+        D: IntoFuture,
+    {
+        WaitUntil::new(self, deadline.into_future())
+    }
 }
 
 impl<F1> FutureExt for F1

src/future/mod.rs

@@ -76,6 +76,7 @@ pub use join::Join;
 pub use race::Race;
 pub use race_ok::RaceOk;
 pub use try_join::TryJoin;
+pub use wait_until::WaitUntil;
 
 /// A growable group of futures which act as a single unit.
 #[cfg(feature = "alloc")]
@@ -86,3 +87,4 @@ pub(crate) mod join;
 pub(crate) mod race;
 pub(crate) mod race_ok;
 pub(crate) mod try_join;
+pub(crate) mod wait_until;

src/future/wait_until.rs

@@ -0,0 +1,61 @@
+use core::future::Future;
+use core::pin::Pin;
+use core::task::{ready, Context, Poll};
+
+/// Suspends a future until the specified deadline.
+///
+/// This `struct` is created by the [`wait_until`] method on [`FutureExt`]. See its
+/// documentation for more.
+///
+/// [`wait_until`]: crate::future::FutureExt::wait_until
+/// [`FutureExt`]: crate::future::FutureExt
+#[derive(Debug)]
+#[pin_project::pin_project]
+#[must_use = "futures do nothing unless polled or .awaited"]
+pub struct WaitUntil<F, D> {
+    #[pin]
+    future: F,
+    #[pin]
+    deadline: D,
+    state: State,
+}
+
+/// The internal state
+#[derive(Debug)]
+enum State {
+    Started,
+    PollFuture,
+    Completed,
+}
+
+impl<F, D> WaitUntil<F, D> {
+    pub(super) fn new(future: F, deadline: D) -> Self {
+        Self {
+            future,
+            deadline,
+            state: State::Started,
+        }
+    }
+}
+
+impl<F: Future, D: Future> Future for WaitUntil<F, D> {
+    type Output = F::Output;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut this = self.project();
+        loop {
+            match this.state {
+                State::Started => {
+                    ready!(this.deadline.as_mut().poll(cx));
+                    *this.state = State::PollFuture;
+                }
+                State::PollFuture => {
+                    let value = ready!(this.future.as_mut().poll(cx));
+                    *this.state = State::Completed;
+                    return Poll::Ready(value);
+                }
+                State::Completed => panic!("future polled after completing"),
+            }
+        }
+    }
+}