sync: add watch::Receiver::wait_for

[carllerche]

Correct use of watch::Receiver requires first checking the current value, then waiting for changes. Forgetting the first check can lead to race conditions.

Consider a watch used to track shutdown notification. The following would result in the shutdown notification being lost:

let (tx, _rx) = watch::channel(false);

tx.send(true)?;

let rx = tx.subscribe();

rx.changed().await?; // hang forever

We could add some APIs that make it easier to wait for specific values.

Roughly:

async fn wait_for(&mut self, f: impl FnMut(&T) -> bool) {
    loop {
        if f(self.borrow_and_update()) {
            return;
        }

        self.changed().await;
    }
}

Then, using a watch to await a shutdown signal becomes:

rx.wait_for(|is_shutdown| *is_shutdown).await;

I also imagine there could be other APIs, like wait_and_map or something that takes fn (&T) -> Option<U>.

[debadree25]

Hey! I would like to give this a try (beginner here hence seeking advise here first) so if I understand correctly we would add the API here tokio/src/sync/watch.rs ?

[Darksonn]

Yes, that's correct.