sync: implement watch::Receiver::wait_for method

tokio/src/sync/tests/loom_watch.rs

@@ -2,6 +2,8 @@ use crate::sync::watch;
 
 use loom::future::block_on;
 use loom::thread;
+use std::sync::Arc;
+use std::time::Duration;
 
 #[test]
 fn smoke() {
@@ -34,3 +36,32 @@ fn smoke() {
         th.join().unwrap();
     })
 }
+
+#[test]
+fn wait_for_test() {
+    loom::model(move || {
+        let (tx, mut rx) = watch::channel(false);
+
+        let tx_arc = Arc::new(tx);
+        let tx1 = tx_arc.clone();
+        let tx2 = tx_arc.clone();
+
+        let th1 = thread::spawn(move || {
+            for _ in 0..10 {
+                tx1.send(false).unwrap();
+                std::thread::sleep(Duration::from_millis(10));
+            }
+        });
+
+        let th2 = thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(10));
+            tx2.send(true).unwrap();
+        });
+
+        let result = block_on(rx.wait_for(|x| *x));
+        assert_eq!(result.unwrap(), true);
+
+        th1.join().unwrap();
+        th2.join().unwrap();
+    });
+}

tokio/src/sync/watch.rs

@@ -610,6 +610,55 @@ impl<T> Receiver<T> {
         }
     }
 
+    /// This function is similar to [`Receiver::changed()`], but it takes a closure that is
+    /// called with a reference to the new value. If the closure returns `true`,
+    /// then the function returns immediately. Otherwise, it waits for a new
+    /// value and calls the closure again.
+    ///
+    /// [`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<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