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

Code before:
    /// called on the last value, and that it returned `false` for that value.
    /// (If the closure returned `true`, then the last value would have been
    /// returned instead of the error.)
    ///
    /// Like the `borrow` method, the returned borrow holds a read lock on the
    /// inner value. This means that long-lived borrows could cause the producer
    /// half to block. It is recommended to keep the borrow as short-lived as
    /// possible. See the documentation of `borrow` for more information on
    /// this.
    ///
    /// [`Receiver::changed()`]: crate::sync::watch::Receiver::changed
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::sync::watch;
    ///
    /// #[tokio::main]
    ///
    /// async fn main() {
    ///     let (tx, _rx) = watch::channel("hello");
    ///
    ///     tx.send("goodbye").unwrap();
    ///     
    ///     // here we subscribe to a second receiver
    ///     // now in case of using `changed` we would have
    ///     // to first check the current value and then wait
    ///     // for changes or else `changed` would hang.
    ///     let mut rx2 = tx.subscribe();
    ///     
    ///     // in place of changed we have use `wait_for`
    ///     // which would automatically check the current value
    ///     // and wait for changes until the closure returns true.
    ///     assert!(rx2.wait_for(|val| *val == "goodbye").await.is_ok());
    ///     assert_eq!(*rx2.borrow(), "goodbye");
    /// }
    /// ```
    pub async fn wait_for(
        &mut self,
        mut f: impl FnMut(&T) -> bool,
    ) -> Result<(), error::RecvError> {
        loop {
            if f(&self.borrow_and_update()) {
                return Ok(());
            }

            match self.changed().await {
                Ok(()) => {}
                Err(e) => {
                    // some error occurred but we still need to call the closure
                    // to guarantee that it has been called on the last value.

Review comment (suggestion):
I noticed that the code is currently using **&self.borrow_and_update()**, which creates a reference to a clone of the value. To avoid cloning the value multiple times in the loop, we could use **self.borrow()** instead. That makes sense?

```suggestion
    pub async fn wait_for(
        &mut self,
        mut f: impl FnMut(&T) -> bool,
    ) -> Result<(), error::RecvError> {
        loop {
            let borrowed_value = self.borrow();
            if f(&*borrowed_value) {
                return Ok(());
            }

            match self.changed().await {
                Ok(()) => {}
                Err(e) => {
                    // Call the closure on the last value before returning the error.
                    if f(&*borrowed_value) {
                        return Ok(());
                    } else {
                        return Err(e);
                    }
                }
            }
        }
    }
```

Code after:
    /// # Examples
    ///
    /// ```
    /// use tokio::sync::watch;
    ///
    /// #[tokio::main]
    ///
    /// async fn main() {
    ///     let (tx, _rx) = watch::channel("hello");
    ///
    ///     tx.send("goodbye").unwrap();
    ///
    ///     // here we subscribe to a second receiver
    ///     // now in case of using `changed` we would have
    ///     // to first check the current value and then wait
    ///     // for changes or else `changed` would hang.
    ///     let mut rx2 = tx.subscribe();
    ///
    ///     // in place of changed we have use `wait_for`
    ///     // which would automatically check the current value
    ///     // and wait for changes until the closure returns true.
    ///     assert!(rx2.wait_for(|val| *val == "goodbye").await.is_ok());
    ///     assert_eq!(*rx2.borrow(), "goodbye");
    /// }
    /// ```
    pub async fn wait_for(
        &mut self,
        mut f: impl FnMut(&T) -> bool,
    ) -> Result<Ref<'_, T>, error::RecvError> {
        let mut closed = false;
        loop {
            {
                let inner = self.shared.value.read().unwrap();

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

                if (!closed || has_changed) && f(&inner) {
                    return Ok(Ref { inner, has_changed });
                }
            }

            if closed {
                return Err(error::RecvError(()));
            }

            // Wait for the value to change.
            closed = changed_impl(&self.shared, &mut self.version).await.is_err();
        }
    }