diff --git a/tokio/src/sync/watch.rs b/tokio/src/sync/watch.rs index 7d042a6f950..1cbda3c5d02 100644 --- a/tokio/src/sync/watch.rs +++ b/tokio/src/sync/watch.rs @@ -767,44 +767,49 @@ impl Receiver { /// When this function returns, the value that was passed to the closure /// when it returned `true` will be considered seen. /// - /// If the channel is closed, then `wait_for` will return a `RecvError`. + /// 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 + /// 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 + /// [`borrow`]: Receiver::borrow + /// [`RecvError`]: error::RecvError + /// + /// # Cancel safety + /// + /// This method is cancel safe. If dropped before completion, it is + /// guaranteed that the last seen value `val` 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, _rx) = watch::channel("hello"); + /// let (tx, mut rx) = watch::channel("hello"); /// - /// tx.send("goodbye").unwrap(); + /// tokio::spawn(async move { + /// sleep(Duration::from_micros(100)).await; + /// 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"); + /// assert!(rx.wait_for(|val| *val == "goodbye").await.is_ok()); + /// assert_eq!(*rx.borrow(), "goodbye"); /// } /// ``` pub async fn wait_for(