Repository: tokio-rs/tokio
File: tokio/src/sync/watch.rs (Rust)

Code before:
    /// # Cancel safety
    ///
    /// This method is cancel safe. If dropped before completion, it is
    /// guaranteed that the last seen value `val` (if any) satisfies
    /// `f(val) == false`.
    ///
    /// # Panics
    ///
    /// If and only if the closure `f` panics. In that case, no resource owned
    /// or shared by this [`Receiver`] will be poisoned.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::sync::watch;
    /// use tokio::time::{sleep, Duration};
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let (tx, mut rx) = watch::channel("hello");
    ///
    ///     tokio::spawn(async move {
    ///         sleep(Duration::from_micros(100)).await;
    ///         tx.send("goodbye").unwrap();
    ///     });
    ///
    ///     assert!(rx.wait_for(|val| *val == "goodbye").await.is_ok());
    ///     assert_eq!(*rx.borrow(), "goodbye");
    /// }
    /// ```
    pub async fn wait_for(
        &mut self,
        f: impl FnMut(&T) -> bool,
    ) -> Result<Ref<'_, T>, error::RecvError> {
        cooperative(self.wait_for_inner(f)).await
    }

    async fn wait_for_inner(
        &mut self,
        mut f: impl FnMut(&T) -> bool,
    ) -> Result<Ref<'_, T>, error::RecvError> {
        let mut closed = false;
        loop {
            {
                let inner = self.shared.value.read();

                let new_version = self.shared.state.load().version();
                let has_changed = self.version != new_version;
                self.version = new_version;

                if !closed || has_changed {

Review comment (suggestion):
We like to avoid sleeps in tests to reduce the amount of time it takes to run the tests. If you make the runtime use mocked time via the `start_paused` parameter, then this test will run instantly no matter what the duration is.

https://github.com/tokio-rs/tokio/blob/10e23d1c621ab38aadf2cefba1120494cff615f0/tokio/src/runtime/task/join.rs#L190-L205

Code after:
    /// # Cancel safety
    ///
    /// This method is cancel safe. If you use it as the event in a
    /// [`tokio::select!`](crate::select) statement and some other branch
    /// completes first, then it is guaranteed that the last seen value `val`
    /// (if any) satisfies `f(val) == false`.
    ///
    /// # Panics
    ///
    /// If and only if the closure `f` panics. In that case, no resource owned
    /// or shared by this [`Receiver`] will be poisoned.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::sync::watch;
    /// use tokio::time::{sleep, Duration};
    ///
    /// #[tokio::main(flavor = "current_thread", start_paused = true)]
    /// async fn main() {
    ///     let (tx, mut rx) = watch::channel("hello");
    ///
    ///     tokio::spawn(async move {
    ///         sleep(Duration::from_secs(1)).await;
    ///         tx.send("goodbye").unwrap();
    ///     });
    ///
    ///     assert!(rx.wait_for(|val| *val == "goodbye").await.is_ok());
    ///     assert_eq!(*rx.borrow(), "goodbye");
    /// }
    /// ```
    pub async fn wait_for(
        &mut self,
        f: impl FnMut(&T) -> bool,
    ) -> Result<Ref<'_, T>, error::RecvError> {
        cooperative(self.wait_for_inner(f)).await
    }

    async fn wait_for_inner(
        &mut self,
        mut f: impl FnMut(&T) -> bool,
    ) -> Result<Ref<'_, T>, error::RecvError> {
        let mut closed = false;
        loop {
            {
                let inner = self.shared.value.read();

                let new_version = self.shared.state.load().version();
                let has_changed = self.version != new_version;
                self.version = new_version;