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

Code before:
    ///
    /// ```
    /// 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<bool, error::RecvError> {
        loop {
            if f(&self.borrow_and_update()) {
                return Ok(true);
            }

            let changed = self.changed().await;
            if changed.is_err() {
                return Ok(false);
            }
        }
    }

    /// Returns `true` if receivers belong to the same channel.
    ///
    /// # Examples
    ///
    /// ```
    /// let (tx, rx) = tokio::sync::watch::channel(true);
    /// let rx2 = rx.clone();
    /// assert!(rx.same_channel(&rx2));
    ///
    /// let (tx3, rx3) = tokio::sync::watch::channel(true);

Review comment (suggestion):
There's an extra empty line here.
```suggestion
    /// ```
    pub async fn wait_for(
```

Code after:
    /// If the channel is closed, then `wait_for` will return a `RecvError`.
    /// Once this happens, no more messages can ever be sent on the channel.
    /// When an error is returned, it is guaranteed that the closure has been
    /// 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<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;