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

Code before:
                    match result {
                        Ok(true) => {
                            return Ok(Ref { inner, has_changed });
                        }
                        Ok(false) => {
                            // Skip the value.
                        }
                        Err(panicked) => {
                            // Drop the read-lock to avoid poisoning it.
                            drop(inner);
                            // Forward the panic to the caller.
                            panic::resume_unwind(panicked);
                            // Unreachable
                        }
                    };
                }
            }

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

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

    /// 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);
    /// assert!(!rx3.same_channel(&rx2));
    /// ```
    pub fn same_channel(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.shared, &other.shared)
    }

    cfg_process_driver! {
        pub(crate) fn try_has_changed(&mut self) -> Option<Result<(), error::RecvError>> {
            maybe_changed(&self.shared, &mut self.version)
        }
    }
}

Review comment (suggestion):
This doesn't make `wait_for` cooperative, as it may exit before it reaches this call without touching the budget. You need to wrap the entire function.

Also, you may want to make a `wait_for_inner` method instead of wrapping the entire body in `cooperative(async { ... })` to avoid extra indentation.

Code after:

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

                if !closed || has_changed {
                    let result = panic::catch_unwind(panic::AssertUnwindSafe(|| f(&inner)));
                    match result {
                        Ok(true) => {
                            return Ok(Ref { inner, has_changed });
                        }
                        Ok(false) => {
                            // Skip the value.
                        }
                        Err(panicked) => {
                            // Drop the read-lock to avoid poisoning it.
                            drop(inner);
                            // Forward the panic to the caller.
                            panic::resume_unwind(panicked);
                            // Unreachable
                        }
                    };
                }
            }

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

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

    /// 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);
    /// assert!(!rx3.same_channel(&rx2));
    /// ```
    pub fn same_channel(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.shared, &other.shared)
    }

    cfg_process_driver! {