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

Code before:
    fn drop(&mut self) {
        // No synchronization necessary as this is only used as a counter and
        // not memory access.
        if 1 == self.shared.ref_count_rx.fetch_sub(1, Relaxed) {
            // This is the last `Receiver` handle, tasks waiting on `Sender::closed()`
            self.shared.notify_tx.notify_waiters();
        }
    }
}

impl<T> Sender<T> {
    /// Creates the sending-half of the [`watch`] channel.
    ///
    /// See documentation of [`watch::channel`] for errors when calling this function.
    /// Beware that attempting to send a value when there are no receivers will
    /// return an error.
    ///
    /// [`watch`]: crate::sync::watch
    /// [`watch::channel`]: crate::sync::watch
    ///
    /// # Examples
    /// ```
    /// let sender = tokio::sync::watch::Sender::new(0u8);
    /// assert!(sender.send(3).is_err())
    /// let _rec = sender.subscribe()
    /// assert!(sender.send(4).is_ok())
    pub fn new(init: T) -> Self {
        let (tx, _) = channel(init);
        tx
    }

    /// Sends a new value via the channel, notifying all receivers.
    ///
    /// This method fails if the channel is closed, which is the case when
    /// every receiver has been dropped. It is possible to reopen the channel
    /// using the [`subscribe`] method. However, when `send` fails, the value
    /// isn't made available for future receivers (but returned with the
    /// [`SendError`]).
    ///
    /// To always make a new value available for future receivers, even if no
    /// receiver currently exists, one of the other send methods
    /// ([`send_if_modified`], [`send_modify`], or [`send_replace`]) can be
    /// used instead.
    ///
    /// [`subscribe`]: Sender::subscribe
    /// [`SendError`]: error::SendError
    /// [`send_if_modified`]: Sender::send_if_modified
    /// [`send_modify`]: Sender::send_modify
    /// [`send_replace`]: Sender::send_replace
    pub fn send(&self, value: T) -> Result<(), error::SendError<T>> {
        // This is pretty much only useful as a hint anyway, so synchronization isn't critical.

Review comment (bug):
You will need to fix this test:
```

failures:

error: doctest failed, to rerun pass `-p tokio --doc`
---- tokio/src/sync/watch.rs - sync::watch::Sender<T>::new (line 882) stdout ----
error: expected `;`, found keyword `let`
 --> tokio/src/sync/watch.rs:884:33
  |
5 | assert!(sender.send(3).is_err())
  |                                 ^ help: add `;` here
6 | let _rec = sender.subscribe()
  | --- unexpected token

error: expected `;`, found `assert`
 --> tokio/src/sync/watch.rs:885:30
  |
6 | let _rec = sender.subscribe()
  |                              ^ help: add `;` here
7 | assert!(sender.send(4).is_ok())
  | ------ unexpected token

error: aborting due to 2 previous errors

Couldn't compile the test.

failures:
    tokio/src/sync/watch.rs - sync::watch::Sender<T>::new (line 882)
```
Also, please add \`\`\` at the end of the example as well.

Code after:
        notified.await;
        // loop around again in case the wake-up was spurious
    }
}

impl<T> Clone for Receiver<T> {
    fn clone(&self) -> Self {
        let version = self.version;
        let shared = self.shared.clone();

        Self::from_shared(version, shared)
    }
}

impl<T> Drop for Receiver<T> {
    fn drop(&mut self) {
        // No synchronization necessary as this is only used as a counter and
        // not memory access.
        if 1 == self.shared.ref_count_rx.fetch_sub(1, Relaxed) {
            // This is the last `Receiver` handle, tasks waiting on `Sender::closed()`
            self.shared.notify_tx.notify_waiters();
        }
    }
}

impl<T> Sender<T> {
    /// Creates the sending-half of the [`watch`] channel.
    ///
    /// See documentation of [`watch::channel`] for errors when calling this function.
    /// Beware that attempting to send a value when there are no receivers will
    /// return an error.
    ///
    /// [`watch`]: crate::sync::watch
    /// [`watch::channel`]: crate::sync::watch
    ///
    /// # Examples
    /// ```
    /// let sender = tokio::sync::watch::Sender::new(0u8);
    /// assert!(sender.send(3).is_err());
    /// let _rec = sender.subscribe();
    /// assert!(sender.send(4).is_ok());
    /// ```
    pub fn new(init: T) -> Self {
        let (tx, _) = channel(init);
        tx
    }

    /// Sends a new value via the channel, notifying all receivers.
    ///
    /// This method fails if the channel is closed, which is the case when
    /// every receiver has been dropped. It is possible to reopen the channel