nuclei::sync::watch

Struct Sender

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

Expand description

Sends values to the associated Receivers.

Created by channel. Clone it for more producers. The channel closes for the receivers when the last sender is dropped.

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

Implementations§

Source §

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

Source

pub fn new(init: T) -> Self

Creates the sending half of a channel with no receiver yet.

Sending fails until subscribe creates one.

use nuclei::sync::watch::Sender;

let sender = Sender::new(0u8);
assert!(sender.send(3).is_err());
let _rx = sender.subscribe();
assert!(sender.send(4).is_ok());

Source

pub fn send(&self, value: T) -> Result<(), SendError<T>>

Sends a new value, notifying all receivers.

Fails if every receiver is gone; the value comes back in the error and is not kept for receivers that subscribe later. To always make a value available, even with no receiver, use send_replace, send_modify or send_if_modified.

The publication is wait-free; the wake of the receivers makes the whole call lock-free (see the module docs).

use nuclei::sync::watch;

let (tx, rx) = watch::channel(1);
tx.send(2).unwrap();
assert_eq!(*rx.borrow(), 2);
drop(rx);
assert_eq!(tx.send(3).unwrap_err().0, 3);

Source

pub fn send_modify<F>(&self, modify: F)
where F: FnMut(&mut T), T: Clone,

Modifies the watched value unconditionally, notifying all receivers.

Also works with no receiver. modify runs on a clone of the current value, and the clone is published by a compare-and-swap; if another sender published first, modify runs again on the new value, so it is FnMut. Prefer send_if_modified when the value is only sometimes changed.

Lock-free: it retries against other senders.

§Panics

Panics if modify (or the clone) panics. Nothing is published then and no receiver is notified: the channel keeps the value it had.

use nuclei::sync::watch;

#[derive(Clone)]
struct State {
    counter: usize,
}
let (tx, rx) = watch::channel(State { counter: 0 });
tx.send_modify(|state| state.counter += 1);
assert_eq!(rx.borrow().counter, 1);

Source

pub fn send_if_modified<F>(&self, modify: F) -> bool
where F: FnMut(&mut T) -> bool, T: Clone,

Modifies the watched value conditionally, notifying all receivers only if it was modified.

modify runs on a clone of the current value and returns true if it changed it. Only then is the clone published, by a compare-and-swap that starts over on the new value when another sender published first (so modify is FnMut and may run more than once). When it returns false nothing is published, whatever it did to its copy: unlike a lock-based channel there is no silent modification.

Returns true if a new value was published and false if modify returned false.

Lock-free: it retries against other senders.

§Panics

Panics if modify (or the clone) panics. Nothing is published then.

use nuclei::sync::watch;

let (tx, mut rx) = watch::channel(1u32);
let inc_if_odd = |n: &mut u32| {
    if *n % 2 == 1 {
        *n += 1;
        return true;
    }
    false
};
assert!(tx.send_if_modified(inc_if_odd));
assert_eq!(*rx.borrow_and_update(), 2);
assert!(!tx.send_if_modified(inc_if_odd));
assert!(!rx.has_changed().unwrap());

Source

pub fn send_replace(&self, value: T) -> T
where T: Clone,

Sends a new value, notifying all receivers, and returns a clone of the value it replaced.

Also works with no receiver. The old value is returned as a clone because receivers may still be reading it; see the module docs. The publication is wait-free; the wake of the receivers makes the whole call lock-free.

use nuclei::sync::watch;

let (tx, _rx) = watch::channel(1);
assert_eq!(tx.send_replace(2), 1);
assert_eq!(tx.send_replace(3), 2);

Source

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

Returns a reference to the most recently sent value.

A borrow from the sender is never marked changed. See Receiver::borrow for the advice on holding the reference.

use nuclei::sync::watch;

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

Source

pub fn is_closed(&self) -> bool

Whether the channel is closed for sending: every Receiver has been dropped. subscribe reopens it.

use nuclei::sync::watch;

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

Source

pub async fn closed(&self)

Completes when every receiver has been dropped.

Lets the producer stop work nobody wants. If the channel closes briefly (the last receiver is dropped and subscribe is called again), this may or may not notice.

§Cancel safety

Cancel safe.

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

let (tx, rx) = watch::channel("hello");
let mut cx = Context::from_waker(Waker::noop());
let mut closed = pin!(tx.closed());
assert_eq!(closed.as_mut().poll(&mut cx), Poll::Pending);
drop(rx);
assert_eq!(closed.as_mut().poll(&mut cx), Poll::Ready(()));

Creates a new Receiver connected to this sender.

The value that is current now counts as seen by the new receiver. It works with no other receiver, which reopens the channel.

use nuclei::sync::watch;

let (tx, _rx) = watch::channel(0u64);
tx.send(5).unwrap();
let rx = tx.subscribe();
assert_eq!(*rx.borrow(), 5);
assert!(!rx.has_changed().unwrap());

Source

pub fn receiver_count(&self) -> usize

How many receivers exist now.

use nuclei::sync::watch;

let (tx, rx) = watch::channel("hello");
assert_eq!(tx.receiver_count(), 1);
let _rx2 = rx.clone();
assert_eq!(tx.receiver_count(), 2);

Source

pub fn sender_count(&self) -> usize

How many senders exist now.

use nuclei::sync::watch;

let (tx, _rx) = watch::channel("hello");
assert_eq!(tx.sender_count(), 1);
let tx2 = tx.clone();
assert_eq!(tx2.sender_count(), 2);

Source

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

Whether the two senders belong to the same channel.

use nuclei::sync::watch;

let (tx, _rx) = watch::channel(true);
let tx2 = tx.clone();
assert!(tx.same_channel(&tx2));
let (tx3, _rx3) = watch::channel(true);
assert!(!tx3.same_channel(&tx2));