diff --git a/embassy-sync/src/watch.rs b/embassy-sync/src/watch.rs index 7e5e92741..3e22b1e7b 100644 --- a/embassy-sync/src/watch.rs +++ b/embassy-sync/src/watch.rs @@ -1,4 +1,4 @@ -//! A synchronization primitive for passing the latest value to **multiple** tasks. +//! A synchronization primitive for passing the latest value to **multiple** receivers. use core::cell::RefCell; use core::future::poll_fn; @@ -10,22 +10,17 @@ use crate::blocking_mutex::raw::RawMutex; use crate::blocking_mutex::Mutex; use crate::waitqueue::MultiWakerRegistration; -/// A `Watch` is a single-slot signaling primitive, which can awake `N` up to separate [`Receiver`]s. +/// The `Watch` is a single-slot signaling primitive that allows multiple receivers to concurrently await +/// changes to the value. Unlike a [`Signal`](crate::signal::Signal), `Watch` supports multiple receivers, +/// and unlike a [`PubSubChannel`](crate::pubsub::PubSubChannel), `Watch` immediately overwrites the previous +/// value when a new one is sent, without waiting for all receivers to read the previous value. /// -/// Similar to a [`Signal`](crate::signal::Signal), except `Watch` allows for multiple tasks to -/// `.await` the latest value, and all receive it. +/// This makes `Watch` particularly useful when a single task updates a value or "state", and multiple other tasks +/// need to be notified about changes to this value asynchronously. Receivers may "lose" stale values, as they are +/// always provided with the latest value. /// -/// This is similar to a [`PubSubChannel`](crate::pubsub::PubSubChannel) with a buffer size of 1, except -/// "sending" to it (calling [`Watch::write`]) will immediately overwrite the previous value instead -/// of waiting for the receivers to pop the previous value. -/// -/// `Watch` is useful when a single task is responsible for updating a value or "state", which multiple other -/// tasks are interested in getting notified about changes to the latest value of. It is therefore fine for -/// [`Receiver`]s to "lose" stale values. -/// -/// Anyone with a reference to the Watch can update or peek the value. Watches are generally declared -/// as `static`s and then borrowed as required to either [`Watch::peek`] the value or obtain a [`Receiver`] -/// with [`Watch::receiver`] which has async methods. +/// Typically, `Watch` instances are declared as `static`, and a [`Sender`] and [`Receiver`] +/// (or [`DynSender`] and/or [`DynReceiver`]) are obtained and passed to the relevant parts of the program. /// ``` /// /// use futures_executor::block_on; @@ -36,18 +31,18 @@ use crate::waitqueue::MultiWakerRegistration; /// /// static WATCH: Watch = Watch::new(); /// -/// // Obtain Receivers +/// // Obtain receivers and sender /// let mut rcv0 = WATCH.receiver().unwrap(); -/// let mut rcv1 = WATCH.receiver().unwrap(); -/// assert!(WATCH.receiver().is_err()); +/// let mut rcv1 = WATCH.dyn_receiver().unwrap(); +/// let mut snd = WATCH.sender(); /// +/// // No more receivers, and no update +/// assert!(WATCH.receiver().is_err()); /// assert_eq!(rcv1.try_changed(), None); /// -/// WATCH.write(10); -/// assert_eq!(WATCH.try_peek(), Some(10)); -/// +/// snd.send(10); /// -/// // Receive the new value +/// // Receive the new value (async or try) /// assert_eq!(rcv0.changed().await, 10); /// assert_eq!(rcv1.try_changed(), Some(10)); /// @@ -55,13 +50,14 @@ use crate::waitqueue::MultiWakerRegistration; /// assert_eq!(rcv0.try_changed(), None); /// assert_eq!(rcv1.try_changed(), None); /// -/// WATCH.write(20); +/// snd.send(20); /// -/// // Defference `between` peek `get`. +/// // Peek does not mark the value as seen /// assert_eq!(rcv0.peek().await, 20); -/// assert_eq!(rcv1.get().await, 20); -/// /// assert_eq!(rcv0.try_changed(), Some(20)); +/// +/// // Get marks the value as seen +/// assert_eq!(rcv1.get().await, 20); /// assert_eq!(rcv1.try_changed(), None); /// /// }; @@ -80,30 +76,57 @@ struct WatchState { /// A trait representing the 'inner' behavior of the `Watch`. pub trait WatchBehavior { + /// Sends a new value to the `Watch`. + fn send(&self, val: T); + + /// Clears the value of the `Watch`. + fn clear(&self); + /// Poll the `Watch` for the current value, **without** making it as seen. - fn inner_poll_peek(&self, cx: &mut Context<'_>) -> Poll; + fn poll_peek(&self, cx: &mut Context<'_>) -> Poll; /// Tries to peek the value of the `Watch`, **without** marking it as seen. - fn inner_try_peek(&self) -> Option; + fn try_peek(&self) -> Option; /// Poll the `Watch` for the current value, making it as seen. - fn inner_poll_get(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll; + fn poll_get(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll; /// Tries to get the value of the `Watch`, marking it as seen. - fn inner_try_get(&self, id: &mut u64) -> Option; + fn try_get(&self, id: &mut u64) -> Option; /// Poll the `Watch` for a changed value, marking it as seen. - fn inner_poll_changed(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll; + fn poll_changed(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll; /// Tries to retrieve the value of the `Watch` if it has changed, marking it as seen. - fn inner_try_changed(&self, id: &mut u64) -> Option; + fn try_changed(&self, id: &mut u64) -> Option; /// Checks if the `Watch` is been initialized with a value. - fn inner_contains_value(&self) -> bool; + fn contains_value(&self) -> bool; + + /// Used when a receiver is dropped to decrement the receiver count. + /// + /// ## This method should not be called by the user. + fn drop_receiver(&self); } impl WatchBehavior for Watch { - fn inner_poll_peek(&self, cx: &mut Context<'_>) -> Poll { + fn send(&self, val: T) { + self.mutex.lock(|state| { + let mut s = state.borrow_mut(); + s.data = Some(val); + s.current_id += 1; + s.wakers.wake(); + }) + } + + fn clear(&self) { + self.mutex.lock(|state| { + let mut s = state.borrow_mut(); + s.data = None; + }) + } + + fn poll_peek(&self, cx: &mut Context<'_>) -> Poll { self.mutex.lock(|state| { let mut s = state.borrow_mut(); match &s.data { @@ -116,11 +139,11 @@ impl WatchBehavior for Watch }) } - fn inner_try_peek(&self) -> Option { + fn try_peek(&self) -> Option { self.mutex.lock(|state| state.borrow().data.clone()) } - fn inner_poll_get(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll { + fn poll_get(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll { self.mutex.lock(|state| { let mut s = state.borrow_mut(); match &s.data { @@ -136,7 +159,7 @@ impl WatchBehavior for Watch }) } - fn inner_try_get(&self, id: &mut u64) -> Option { + fn try_get(&self, id: &mut u64) -> Option { self.mutex.lock(|state| { let s = state.borrow(); *id = s.current_id; @@ -144,7 +167,7 @@ impl WatchBehavior for Watch }) } - fn inner_poll_changed(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll { + fn poll_changed(&self, id: &mut u64, cx: &mut Context<'_>) -> Poll { self.mutex.lock(|state| { let mut s = state.borrow_mut(); match (&s.data, s.current_id > *id) { @@ -160,7 +183,7 @@ impl WatchBehavior for Watch }) } - fn inner_try_changed(&self, id: &mut u64) -> Option { + fn try_changed(&self, id: &mut u64) -> Option { self.mutex.lock(|state| { let s = state.borrow(); match s.current_id > *id { @@ -173,9 +196,16 @@ impl WatchBehavior for Watch }) } - fn inner_contains_value(&self) -> bool { + fn contains_value(&self) -> bool { self.mutex.lock(|state| state.borrow().data.is_some()) } + + fn drop_receiver(&self) { + self.mutex.lock(|state| { + let mut s = state.borrow_mut(); + s.receiver_count -= 1; + }) + } } #[derive(Debug)] @@ -198,14 +228,14 @@ impl<'a, M: RawMutex, T: Clone, const N: usize> Watch { } } - /// Write a new value to the `Watch`. - pub fn write(&self, val: T) { - self.mutex.lock(|state| { - let mut s = state.borrow_mut(); - s.data = Some(val); - s.current_id += 1; - s.wakers.wake(); - }) + /// Create a new [`Receiver`] for the `Watch`. + pub fn sender(&self) -> Sender<'_, M, T, N> { + Sender(Snd::new(self)) + } + + /// Create a new [`DynReceiver`] for the `Watch`. + pub fn dyn_sender(&self) -> DynSender<'_, T> { + DynSender(Snd::new(self)) } /// Create a new [`Receiver`] for the `Watch`. @@ -214,7 +244,7 @@ impl<'a, M: RawMutex, T: Clone, const N: usize> Watch { let mut s = state.borrow_mut(); if s.receiver_count < N { s.receiver_count += 1; - Ok(Receiver(Rcv::new(self))) + Ok(Receiver(Rcv::new(self, 0))) } else { Err(Error::MaximumReceiversReached) } @@ -227,29 +257,121 @@ impl<'a, M: RawMutex, T: Clone, const N: usize> Watch { let mut s = state.borrow_mut(); if s.receiver_count < N { s.receiver_count += 1; - Ok(DynReceiver(Rcv::new(self))) + Ok(DynReceiver(Rcv::new(self, 0))) } else { Err(Error::MaximumReceiversReached) } }) } +} + +/// A receiver can `.await` a change in the `Watch` value. +pub struct Snd<'a, T: Clone, W: WatchBehavior + ?Sized> { + watch: &'a W, + _phantom: PhantomData, +} + +impl<'a, T: Clone, W: WatchBehavior + ?Sized> Clone for Snd<'a, T, W> { + fn clone(&self) -> Self { + Self { + watch: self.watch, + _phantom: PhantomData, + } + } +} + +impl<'a, T: Clone, W: WatchBehavior + ?Sized> Snd<'a, T, W> { + /// Creates a new `Receiver` with a reference to the `Watch`. + fn new(watch: &'a W) -> Self { + Self { + watch, + _phantom: PhantomData, + } + } + + /// Sends a new value to the `Watch`. + pub fn send(&self, val: T) { + self.watch.send(val) + } + + /// Clears the value of the `Watch`. + /// This will cause calls to [`Rcv::get`] and [`Rcv::peek`] to be pending. + pub fn clear(&self) { + self.watch.clear() + } /// Tries to retrieve the value of the `Watch`. pub fn try_peek(&self) -> Option { - self.inner_try_peek() + self.watch.try_peek() } /// Returns true if the `Watch` contains a value. pub fn contains_value(&self) -> bool { - self.inner_contains_value() + self.watch.contains_value() } +} - /// Clears the value of the `Watch`. This will cause calls to [`Rcv::get`] and [`Rcv::peek`] to be pending. - pub fn clear(&self) { - self.mutex.lock(|state| { - let mut s = state.borrow_mut(); - s.data = None; - }) +/// A sender of a `Watch` channel. +/// +/// For a simpler type definition, consider [`DynSender`] at the expense of +/// some runtime performance due to dynamic dispatch. +pub struct Sender<'a, M: RawMutex, T: Clone, const N: usize>(Snd<'a, T, Watch>); + +impl<'a, M: RawMutex, T: Clone, const N: usize> Clone for Sender<'a, M, T, N> { + fn clone(&self) -> Self { + Self(self.0.clone()) + } +} + +impl<'a, M: RawMutex, T: Clone, const N: usize> Sender<'a, M, T, N> { + /// Converts the `Sender` into a [`DynSender`]. + pub fn as_dyn(self) -> DynSender<'a, T> { + DynSender(Snd::new(self.watch)) + } +} + +impl<'a, M: RawMutex, T: Clone, const N: usize> Into> for Sender<'a, M, T, N> { + fn into(self) -> DynSender<'a, T> { + self.as_dyn() + } +} + +impl<'a, M: RawMutex, T: Clone, const N: usize> Deref for Sender<'a, M, T, N> { + type Target = Snd<'a, T, Watch>; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl<'a, M: RawMutex, T: Clone, const N: usize> DerefMut for Sender<'a, M, T, N> { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.0 + } +} + +/// A sender which holds a **dynamic** reference to a `Watch` channel. +/// +/// This is an alternative to [`Sender`] with a simpler type definition, +pub struct DynSender<'a, T: Clone>(Snd<'a, T, dyn WatchBehavior + 'a>); + +impl<'a, T: Clone> Clone for DynSender<'a, T> { + fn clone(&self) -> Self { + Self(self.0.clone()) + } +} + +impl<'a, T: Clone> Deref for DynSender<'a, T> { + type Target = Snd<'a, T, dyn WatchBehavior + 'a>; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl<'a, T: Clone> DerefMut for DynSender<'a, T> { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.0 } } @@ -262,59 +384,83 @@ pub struct Rcv<'a, T: Clone, W: WatchBehavior + ?Sized> { impl<'a, T: Clone, W: WatchBehavior + ?Sized> Rcv<'a, T, W> { /// Creates a new `Receiver` with a reference to the `Watch`. - fn new(watch: &'a W) -> Self { + fn new(watch: &'a W, at_id: u64) -> Self { Self { watch, - at_id: 0, + at_id, _phantom: PhantomData, } } - /// Returns the current value of the `Watch` if it is initialized, **without** marking it as seen. + /// Returns the current value of the `Watch` once it is initialized, **without** marking it as seen. + /// + /// **Note**: Futures do nothing unless you `.await` or poll them. pub async fn peek(&self) -> T { - poll_fn(|cx| self.watch.inner_poll_peek(cx)).await + poll_fn(|cx| self.watch.poll_peek(cx)).await } /// Tries to peek the current value of the `Watch` without waiting, and **without** marking it as seen. pub fn try_peek(&self) -> Option { - self.watch.inner_try_peek() + self.watch.try_peek() } - /// Returns the current value of the `Watch` if it is initialized, marking it as seen. + /// Returns the current value of the `Watch` once it is initialized, marking it as seen. + /// + /// **Note**: Futures do nothing unless you `.await` or poll them. pub async fn get(&mut self) -> T { - poll_fn(|cx| self.watch.inner_poll_get(&mut self.at_id, cx)).await + poll_fn(|cx| self.watch.poll_get(&mut self.at_id, cx)).await } /// Tries to get the current value of the `Watch` without waiting, marking it as seen. pub fn try_get(&mut self) -> Option { - self.watch.inner_try_get(&mut self.at_id) + self.watch.try_get(&mut self.at_id) } /// Waits for the `Watch` to change and returns the new value, marking it as seen. + /// + /// **Note**: Futures do nothing unless you `.await` or poll them. pub async fn changed(&mut self) -> T { - poll_fn(|cx| self.watch.inner_poll_changed(&mut self.at_id, cx)).await + poll_fn(|cx| self.watch.poll_changed(&mut self.at_id, cx)).await } /// Tries to get the new value of the watch without waiting, marking it as seen. pub fn try_changed(&mut self) -> Option { - self.watch.inner_try_changed(&mut self.at_id) + self.watch.try_changed(&mut self.at_id) } /// Checks if the `Watch` contains a value. If this returns true, /// then awaiting [`Rcv::get`] and [`Rcv::peek`] will return immediately. pub fn contains_value(&self) -> bool { - self.watch.inner_contains_value() + self.watch.contains_value() + } +} + +impl<'a, T: Clone, W: WatchBehavior + ?Sized> Drop for Rcv<'a, T, W> { + fn drop(&mut self) { + self.watch.drop_receiver(); } } /// A receiver of a `Watch` channel. pub struct Receiver<'a, M: RawMutex, T: Clone, const N: usize>(Rcv<'a, T, Watch>); -/// A receiver which holds a **reference** to a `Watch` channel. -/// -/// This is an alternative to [`Receiver`] with a simpler type definition, at the expense of -/// some runtime performance due to dynamic dispatch. -pub struct DynReceiver<'a, T: Clone>(Rcv<'a, T, dyn WatchBehavior + 'a>); +impl<'a, M: RawMutex, T: Clone, const N: usize> Receiver<'a, M, T, N> { + /// Converts the `Receiver` into a [`DynReceiver`]. + pub fn as_dyn(self) -> DynReceiver<'a, T> { + // We need to increment the receiver count since the original + // receiver is being dropped, which decrements the count. + self.watch.mutex.lock(|state| { + state.borrow_mut().receiver_count += 1; + }); + DynReceiver(Rcv::new(self.0.watch, self.at_id)) + } +} + +impl<'a, M: RawMutex, T: Clone, const N: usize> Into> for Receiver<'a, M, T, N> { + fn into(self) -> DynReceiver<'a, T> { + self.as_dyn() + } +} impl<'a, M: RawMutex, T: Clone, const N: usize> Deref for Receiver<'a, M, T, N> { type Target = Rcv<'a, T, Watch>; @@ -330,6 +476,12 @@ impl<'a, M: RawMutex, T: Clone, const N: usize> DerefMut for Receiver<'a, M, T, } } +/// A receiver which holds a **dynamic** reference to a `Watch` channel. +/// +/// This is an alternative to [`Receiver`] with a simpler type definition, at the expense of +/// some runtime performance due to dynamic dispatch. +pub struct DynReceiver<'a, T: Clone>(Rcv<'a, T, dyn WatchBehavior + 'a>); + impl<'a, T: Clone> Deref for DynReceiver<'a, T> { type Target = Rcv<'a, T, dyn WatchBehavior + 'a>; @@ -348,167 +500,242 @@ impl<'a, T: Clone> DerefMut for DynReceiver<'a, T> { mod tests { use futures_executor::block_on; - use super::*; + use super::Watch; use crate::blocking_mutex::raw::CriticalSectionRawMutex; #[test] - fn multiple_writes() { + fn multiple_sends() { let f = async { - static WATCH: Watch = Watch::new(); + static WATCH: Watch = Watch::new(); - // Obtain Receivers - let mut rcv0 = WATCH.receiver().unwrap(); - let mut rcv1 = WATCH.dyn_receiver().unwrap(); + // Obtain receiver and sender + let mut rcv = WATCH.receiver().unwrap(); + let snd = WATCH.sender(); - WATCH.write(10); + // Not initialized + assert_eq!(rcv.try_changed(), None); // Receive the new value - assert_eq!(rcv0.changed().await, 10); - assert_eq!(rcv1.changed().await, 10); + snd.send(10); + assert_eq!(rcv.changed().await, 10); + + // Receive another value + snd.send(20); + assert_eq!(rcv.try_changed(), Some(20)); // No update - assert_eq!(rcv0.try_changed(), None); - assert_eq!(rcv1.try_changed(), None); - - WATCH.write(20); - - assert_eq!(rcv0.changed().await, 20); - assert_eq!(rcv1.changed().await, 20); + assert_eq!(rcv.try_changed(), None); }; block_on(f); } #[test] - fn max_receivers() { + fn receive_after_create() { let f = async { - static WATCH: Watch = Watch::new(); + static WATCH: Watch = Watch::new(); - // Obtain Receivers - let _ = WATCH.receiver().unwrap(); - let _ = WATCH.receiver().unwrap(); - assert!(WATCH.receiver().is_err()); + // Obtain sender and send value + let snd = WATCH.sender(); + snd.send(10); + + // Obtain receiver and receive value + let mut rcv = WATCH.receiver().unwrap(); + assert_eq!(rcv.try_changed(), Some(10)); }; block_on(f); } #[test] - fn receive_initial() { + fn max_receivers_drop() { let f = async { static WATCH: Watch = Watch::new(); - // Obtain Receivers + // Try to create 3 receivers (only 2 can exist at once) + let rcv0 = WATCH.receiver(); + let rcv1 = WATCH.receiver(); + let rcv2 = WATCH.receiver(); + + // Ensure the first two are successful and the third is not + assert!(rcv0.is_ok()); + assert!(rcv1.is_ok()); + assert!(rcv2.is_err()); + + // Drop the first receiver + drop(rcv0); + + // Create another receiver and ensure it is successful + let rcv3 = WATCH.receiver(); + assert!(rcv3.is_ok()); + }; + block_on(f); + } + + #[test] + fn multiple_receivers() { + let f = async { + static WATCH: Watch = Watch::new(); + + // Obtain receivers and sender let mut rcv0 = WATCH.receiver().unwrap(); let mut rcv1 = WATCH.receiver().unwrap(); + let snd = WATCH.sender(); - assert_eq!(rcv0.contains_value(), false); - + // No update for both assert_eq!(rcv0.try_changed(), None); assert_eq!(rcv1.try_changed(), None); - WATCH.write(0); - - assert_eq!(rcv0.contains_value(), true); + // Send a new value + snd.send(0); + // Both receivers receive the new value assert_eq!(rcv0.try_changed(), Some(0)); assert_eq!(rcv1.try_changed(), Some(0)); }; block_on(f); } + #[test] + fn clone_senders() { + let f = async { + // Obtain different ways to send + static WATCH: Watch = Watch::new(); + let snd0 = WATCH.sender(); + let snd1 = snd0.clone(); + + // Obtain Receiver + let mut rcv = WATCH.receiver().unwrap().as_dyn(); + + // Send a value from first sender + snd0.send(10); + assert_eq!(rcv.try_changed(), Some(10)); + + // Send a value from second sender + snd1.send(20); + assert_eq!(rcv.try_changed(), Some(20)); + }; + block_on(f); + } + #[test] fn peek_get_changed() { let f = async { static WATCH: Watch = Watch::new(); - // Obtain Receivers - let mut rcv0 = WATCH.receiver().unwrap(); + // Obtain receiver and sender + let mut rcv = WATCH.receiver().unwrap(); + let snd = WATCH.sender(); - WATCH.write(10); + // Send a value + snd.send(10); // Ensure peek does not mark as seen - assert_eq!(rcv0.peek().await, 10); - assert_eq!(rcv0.try_changed(), Some(10)); - assert_eq!(rcv0.try_changed(), None); - assert_eq!(rcv0.peek().await, 10); + assert_eq!(rcv.peek().await, 10); + assert_eq!(rcv.try_changed(), Some(10)); + assert_eq!(rcv.try_changed(), None); + assert_eq!(rcv.try_peek(), Some(10)); - WATCH.write(20); + // Send a value + snd.send(20); // Ensure get does mark as seen - assert_eq!(rcv0.get().await, 20); - assert_eq!(rcv0.try_changed(), None); - assert_eq!(rcv0.try_get(), Some(20)); + assert_eq!(rcv.get().await, 20); + assert_eq!(rcv.try_changed(), None); + assert_eq!(rcv.try_get(), Some(20)); }; block_on(f); } #[test] - fn count_ids() { + fn use_dynamics() { let f = async { static WATCH: Watch = Watch::new(); - // Obtain Receivers - let mut rcv0 = WATCH.receiver().unwrap(); - let mut rcv1 = WATCH.receiver().unwrap(); + // Obtain receiver and sender + let mut dyn_rcv = WATCH.dyn_receiver().unwrap(); + let dyn_snd = WATCH.dyn_sender(); - let get_id = || WATCH.mutex.lock(|state| state.borrow().current_id); + // Send a value + dyn_snd.send(10); - WATCH.write(10); - - assert_eq!(rcv0.changed().await, 10); - assert_eq!(rcv1.changed().await, 10); - - assert_eq!(rcv0.try_changed(), None); - assert_eq!(rcv1.try_changed(), None); - - WATCH.write(20); - WATCH.write(20); - WATCH.write(20); - - assert_eq!(rcv0.changed().await, 20); - assert_eq!(rcv1.changed().await, 20); - - assert_eq!(rcv0.try_changed(), None); - assert_eq!(rcv1.try_changed(), None); - - assert_eq!(get_id(), 4); + // Ensure the dynamic receiver receives the value + assert_eq!(dyn_rcv.try_changed(), Some(10)); + assert_eq!(dyn_rcv.try_changed(), None); }; block_on(f); } #[test] - fn peek_still_await() { + fn convert_to_dyn() { let f = async { static WATCH: Watch = Watch::new(); - // Obtain Receivers - let mut rcv0 = WATCH.receiver().unwrap(); - let mut rcv1 = WATCH.receiver().unwrap(); + // Obtain receiver and sender + let rcv = WATCH.receiver().unwrap(); + let snd = WATCH.sender(); - WATCH.write(10); + // Convert to dynamic + let mut dyn_rcv = rcv.as_dyn(); + let dyn_snd = snd.as_dyn(); - assert_eq!(rcv0.peek().await, 10); - assert_eq!(rcv1.try_peek(), Some(10)); + // Send a value + dyn_snd.send(10); - assert_eq!(rcv0.changed().await, 10); - assert_eq!(rcv1.changed().await, 10); + // Ensure the dynamic receiver receives the value + assert_eq!(dyn_rcv.try_changed(), Some(10)); + assert_eq!(dyn_rcv.try_changed(), None); }; block_on(f); } #[test] - fn peek_with_static() { + fn dynamic_receiver_count() { let f = async { static WATCH: Watch = Watch::new(); - // Obtain Receivers - let rcv0 = WATCH.receiver().unwrap(); - let rcv1 = WATCH.receiver().unwrap(); + // Obtain receiver and sender + let rcv0 = WATCH.receiver(); + let rcv1 = WATCH.receiver(); + let rcv2 = WATCH.receiver(); - WATCH.write(20); + // Ensure the first two are successful and the third is not + assert!(rcv0.is_ok()); + assert!(rcv1.is_ok()); + assert!(rcv2.is_err()); - assert_eq!(rcv0.peek().await, 20); - assert_eq!(rcv1.peek().await, 20); - assert_eq!(WATCH.try_peek(), Some(20)); + // Convert to dynamic + let dyn_rcv0 = rcv0.unwrap().as_dyn(); + + // Drop the (now dynamic) receiver + drop(dyn_rcv0); + + // Create another receiver and ensure it is successful + let rcv3 = WATCH.receiver(); + let rcv4 = WATCH.receiver(); + assert!(rcv3.is_ok()); + assert!(rcv4.is_err()); + }; + block_on(f); + } + + #[test] + fn contains_value() { + let f = async { + static WATCH: Watch = Watch::new(); + + // Obtain receiver and sender + let rcv = WATCH.receiver().unwrap(); + let snd = WATCH.sender(); + + // check if the watch contains a value + assert_eq!(rcv.contains_value(), false); + assert_eq!(snd.contains_value(), false); + + // Send a value + snd.send(10); + + // check if the watch contains a value + assert_eq!(rcv.contains_value(), true); + assert_eq!(snd.contains_value(), true); }; block_on(f); }