tokio::sync::watch

结构体 Sender

pub struct Sender<T> { /* private fields */ }

展开描述

向关联的 Receiver 发送值。

实例由 channel 函数创建。

实现§

Source §

impl<T> Sender<T>

Source

pub fn new(init: T) -> Self

创建 watch 通道的发送端。

有关调用此函数时的错误，请参阅 watch::channel 的文档。请注意，当没有接收者时尝试发送值将返回错误。

§示例

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

Source

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

通过通道发送新值，通知所有接收者。

如果通道已关闭（即所有接收者都已被丢弃），此方法会失败。可以使用 subscribe 方法重新打开通道。但是，当 send 失败时，该值不会对未来的接收者可用（而是通过 SendError 返回）。

如果即使当前没有接收者也总是希望新值对未来的接收者可用，可以改用其他 send 方法之一（send_if_modified、send_modify 或 send_replace）。

Source

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

无条件地原地修改被监视的值，通知所有接收者。

这对于修改被监视的值很有用，无需分配新实例。此外，此方法允许在没有接收者时发送值。

如果在可变借用期间仅按条件修改值，请优先使用更通用的函数 Self::send_if_modified()，以防止对未修改的值发出不必要的更改通知。

§Panics

当 modify 闭包被调用时如果 panic，此函数会 panic。panic 时不会通知任何接收者。闭包在 panic 之前对被监视值所做的所有更改将在后续对 borrow 的调用中可见。

§示例

use tokio::sync::watch;

struct State {
    counter: usize,
}
let (state_tx, state_rx) = watch::channel(State { counter: 0 });
state_tx.send_modify(|state| state.counter += 1);
assert_eq!(state_rx.borrow().counter, 1);

Source

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

按条件原地修改被监视的值，仅在修改时通知所有接收者。

这对于修改被监视的值很有用，无需分配新实例。此外，此方法允许在没有接收者时发送值。

如果在可变借用期间值实际已被修改，则 modify 闭包必须返回 true。仅在尽管进行了可变借用但值保证未被修改时才应返回 false。

仅当闭包返回 true 时才通知接收者。如果闭包已修改值但返回 false，则会导致静默修改——也就是说，修改后的值将在后续对 borrow 的调用中可见，但接收者不会收到更改通知。

返回闭包的结果，即如果值已被修改则返回 true，否则返回 false。

§Panics

当 modify 闭包被调用时如果 panic，此函数会 panic。panic 时不会通知任何接收者。闭包在 panic 之前对被监视值所做的所有更改将在后续对 borrow 的调用中可见。

§示例

use tokio::sync::watch;

struct State {
    counter: usize,
}
let (state_tx, mut state_rx) = watch::channel(State { counter: 1 });
let inc_counter_if_odd = |state: &mut State| {
    if state.counter % 2 == 1 {
        state.counter += 1;
        return true;
    }
    false
};

assert_eq!(state_rx.borrow().counter, 1);

assert!(!state_rx.has_changed().unwrap());
assert!(state_tx.send_if_modified(inc_counter_if_odd));
assert!(state_rx.has_changed().unwrap());
assert_eq!(state_rx.borrow_and_update().counter, 2);

assert!(!state_rx.has_changed().unwrap());
assert!(!state_tx.send_if_modified(inc_counter_if_odd));
assert!(!state_rx.has_changed().unwrap());
assert_eq!(state_rx.borrow_and_update().counter, 2);

Source

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

通过通道发送新值，通知所有接收者并返回通道中的先前值。

这对于重用被监视值内部的缓冲区很有用。此外，此方法允许在没有接收者时发送值。

§示例

use tokio::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>

返回对最近发送的值的引用。

未完成的 borrow 持有内部值的读锁。这意味着长期存活的 borrow 可能导致生产者半部阻塞。建议尽可能缩短 borrow 的生命周期。此外，如果您运行在允许 !Send future 的环境中，必须确保返回的 Ref 类型不会跨 .await 点存活，否则可能导致死锁。

§示例

use tokio::sync::watch;

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

Source

pub fn is_closed(&self) -> bool

检查通道是否已关闭。当所有接收者都已丢弃时会发生这种情况。

§示例

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

drop(rx);
assert!(tx.is_closed());

Source

pub async fn closed(&self)

当所有接收者都已丢弃时完成。

这允许生产者在对所产生值的兴趣被取消时收到通知，并立即停止工作。一旦通道被关闭，重新打开它的唯一方法是调用 Sender::subscribe 以获得新的接收者。

如果通道在短时间内变为关闭状态（例如，最后一个接收者被丢弃然后调用 subscribe），那么对 closed 的这次调用可能返回，但也可能不会“注意到”通道曾在短时间内被关闭。

§Cancel safety

此方法是取消安全的。

§示例

use tokio::sync::watch;

let (tx, rx) = watch::channel("hello");

tokio::spawn(async move {
    // use `rx`
    drop(rx);
});

// Waits for `rx` to drop
tx.closed().await;
println!("the `rx` handles dropped")

创建一个连接到此 Sender 的新 Receiver。

在此次 subscribe 调用之前发送的所有消息最初都被新 Receiver 标记为已读。

即使没有其他接收者也可以调用此方法。在这种情况下，通道会被重新打开。

§示例

新通道将接收通过此 Sender 发送的消息。

use tokio::sync::watch;

let (tx, _rx) = watch::channel(0u64);

tx.send(5).unwrap();

let rx = tx.subscribe();
assert_eq!(5, *rx.borrow());

tx.send(10).unwrap();
assert_eq!(10, *rx.borrow());

通道将最近的消息视为已读，因此此测试保证通过。

use tokio::sync::watch;
use tokio::time::Duration;

let (tx, _rx) = watch::channel(0u64);
tx.send(5).unwrap();
let mut rx = tx.subscribe();

tokio::spawn(async move {
    // by spawning and sleeping, the message is sent after `main`
    // hits the call to `changed`.
    tokio::time::sleep(Duration::from_millis(10)).await;
    tx.send(100).unwrap();
});

rx.changed().await.unwrap();
assert_eq!(100, *rx.borrow());

Source

pub fn receiver_count(&self) -> usize

返回当前存在的接收者数量。

§示例

use tokio::sync::watch;

let (tx, rx1) = watch::channel("hello");

assert_eq!(1, tx.receiver_count());

let mut _rx2 = rx1.clone();

assert_eq!(2, tx.receiver_count());

Source

pub fn sender_count(&self) -> usize

返回当前存在的发送者数量。

§示例

use tokio::sync::watch;

let (tx1, rx) = watch::channel("hello");

assert_eq!(1, tx1.sender_count());

let tx2 = tx1.clone();

assert_eq!(2, tx1.sender_count());
assert_eq!(2, tx2.sender_count());

Source

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

如果发送者属于同一通道则返回 true。

§示例

let (tx, rx) = tokio::sync::watch::channel(true);
let tx2 = tx.clone();
assert!(tx.same_channel(&tx2));

let (tx3, rx3) = tokio::sync::watch::channel(true);
assert!(!tx3.same_channel(&tx2));