nuclei::sync::watch

Struct Receiver

pub struct Receiver<T: Send + Sync + 'static> { /* private fields */ }

Expand description

Receives values from the associated Senders.

Created by channel or Sender::subscribe; clone it for more consumers, each with its own record of what it has seen.

Requires T: Send + Sync + 'static, see the module docs.

Implementations§

Source §

impl<T: Send + Sync + 'static> Receiver<T>

Source

pub fn borrow(&self) -> Ref<'_, T>

Returns a reference to the most recently sent value.

Does not mark the value seen, so a later changed can return at once even though you have read it. If you wait for changes in a loop, prefer borrow_and_update.

Wait-free, and it never blocks a sender. A borrow keeps the value it points to alive; keep it short and drop it before awaiting.

use nuclei::sync::watch;

let (_tx, rx) = watch::channel("hello");
assert_eq!(*rx.borrow(), "hello");

Source

pub fn borrow_and_update(&mut self) -> Ref<'_, T>

Returns a reference to the most recently sent value and marks it seen.

Later calls to changed do not return until a value newer than this one is sent. Wait-free; see borrow for the advice on holding the reference.

use nuclei::sync::watch;

let (tx, mut rx) = watch::channel("hello");
tx.send("goodbye").unwrap();
assert_eq!(*rx.borrow_and_update(), "goodbye");
assert!(!rx.has_changed().unwrap());

Source

pub fn has_changed(&self) -> Result<bool, RecvError>

Checks whether the channel holds a value this receiver has not seen. The value is not marked seen.

It does not compare values: the same value sent twice counts as a change.

§Errors

Returns a RecvError if and only if the channel is closed.

use nuclei::sync::watch;

let (tx, mut rx) = watch::channel("hello");
tx.send("goodbye").unwrap();
assert!(rx.has_changed().unwrap());
assert_eq!(*rx.borrow_and_update(), "goodbye");
assert!(!rx.has_changed().unwrap());
drop(tx);
assert!(rx.has_changed().is_err());

Source

pub fn is_closed(&self) -> bool

Whether the channel is closed: every Sender has been dropped.

use nuclei::sync::watch;

let (tx, rx) = watch::channel(());
assert!(!rx.is_closed());
drop(tx);
assert!(rx.is_closed());

Source

pub fn mark_changed(&mut self)

Marks the current value as not seen.

Afterwards has_changed is true and changed returns at once, whether or not a new value is sent. Useful to trigger a first notification after subscribing.

Source

pub fn mark_unchanged(&mut self)

Marks the current value as seen, without reading it.

Source

pub async fn changed(&mut self) -> Result<(), RecvError>

Waits for a change notification, then marks the current value as seen.

If the current value has not been seen yet, this marks it seen and returns at once. Otherwise it waits until a sender sends a new value or the last sender is dropped.

§Errors

Returns a RecvError if the channel is closed and the current value is seen.

§Cancel safety

Cancel safe: if the future is dropped before it completes, no value was marked seen by it.

use nuclei::sync::watch;
use std::future::Future;
use std::pin::pin;
use std::task::{Context, Poll, Waker};

let (tx, mut rx) = watch::channel("hello");
let mut cx = Context::from_waker(Waker::noop());
let mut changed = pin!(rx.changed());
assert!(changed.as_mut().poll(&mut cx).is_pending());
tx.send("goodbye").unwrap();
assert!(matches!(changed.as_mut().poll(&mut cx), Poll::Ready(Ok(()))));

Source

pub async fn wait_for( &mut self, f: impl FnMut(&T) -> bool, ) -> Result<Ref<'_, T>, RecvError>

Waits for a value that satisfies f.

Calls f on the current value first, even if it is already seen, and then on the newest value whenever something is sent. When f returns true, returns a reference to that value and marks it seen. The channel keeps only the latest value, so a burst of sends can skip values: f always sees the newest.

§Errors

Returns a RecvError if the channel is closed. Then f has been called on the last value and returned false for it.

§Cancel safety

Cancel safe: the last value f returned true for is never marked seen by a dropped call.

§Panics

If and only if f panics. Nothing the receiver shares is poisoned.

use nuclei::sync::watch;
use std::future::Future;
use std::pin::pin;
use std::task::{Context, Poll, Waker};

let (tx, mut rx) = watch::channel(0);
let mut cx = Context::from_waker(Waker::noop());
let mut wait = pin!(rx.wait_for(|v| *v == 2));
assert!(wait.as_mut().poll(&mut cx).is_pending());
tx.send(1).unwrap();
assert!(wait.as_mut().poll(&mut cx).is_pending());
tx.send(2).unwrap();
match wait.as_mut().poll(&mut cx) {
    Poll::Ready(Ok(value)) => assert_eq!(*value, 2),
    _ => panic!("the value 2 was sent"),
}

Source

pub fn same_channel(&self, other: &Self) -> bool

Whether the two receivers belong to the same channel.

use nuclei::sync::watch;

let (_tx, rx) = watch::channel(true);
let rx2 = rx.clone();
assert!(rx.same_channel(&rx2));
let (_tx3, rx3) = watch::channel(true);
assert!(!rx3.same_channel(&rx2));