Модуль watch

Канал с несколькими производителями и несколькими потребителями, который сохраняет только последнее отправленное значение.

Этот канал полезен для отслеживания изменений значения из разных частей кодовой базы, например изменений значений конфигурации.

Использование

channel возвращает пару Sender / Receiver. Это стороны канала — производитель и потребитель. Канал создаётся с начальным значением.

Каждый Receiver независимо отслеживает последнее значение, увиденное вызывающим его кодом.

Чтобы получить текущее значение, хранящееся в канале, и отметить его как увиденное для заданного Receiver, используйте Receiver::borrow_and_update().

Чтобы получить текущее значение, не отмечая его как увиденное, используйте Receiver::borrow(). (Если значение уже отмечено как увиденное, Receiver::borrow() эквивалентен Receiver::borrow_and_update().)

Подробнее о том, когда следует использовать эти методы, см. здесь.

Уведомления об изменениях

Сторона Receiver предоставляет асинхронный метод changed. Этот метод становится готовым, когда через сторону Sender отправляется новое, ещё не просмотренное значение.

• Receiver::changed() возвращает:
  • Ok(()) при получении нового значения.
  • Err(RecvError), если канал закрыт И текущее значение просмотрено.
• Если текущее значение не просмотрено при вызове changed, то changed вернётся немедленно. Если текущее значение просмотрено, метод будет ожидать, пока через сторону Sender не будет отправлено новое сообщение или пока Sender не будет удалён.
• После завершения метод changed отмечает новое значение как просмотренное.
• При создании начальное значение считается просмотренным. Иными словами, Receiver::changed() не вернётся, пока не будет отправлено следующее значение.
• Новые экземпляры Receiver можно создать с помощью Sender::subscribe(). Текущее значение на момент создания Receiver считается просмотренным.

changed и has_changed

Сторона Receiver предоставляет два метода для проверки изменений в канале: has_changed и changed.

• has_changed — это синхронный метод, который проверяет, просмотрено ли текущее значение, и возвращает логическое значение. Этот метод не отмечает значение как просмотренное.

• changed — это асинхронный метод, который возвращается, когда в канале появляется непросмотренное значение. Этот метод отмечает значение как просмотренное.

Обратите внимание: есть два различия в поведении этих методов при возврате ошибки.

• has_changed возвращает ошибку тогда и только тогда, когда канал закрыт.
• changed возвращает ошибку, если канал закрыт И текущее значение просмотрено.

В примере ниже показано, как эти методы различаются по условиям возврата ошибки.

borrow_and_update и borrow

Если получатель собирается в цикле ожидать уведомления от changed, предпочтительно использовать Receiver::borrow_and_update(), а не Receiver::borrow(). Это позволяет избежать возможной гонки, при которой новое значение отправляется между моментом готовности changed и чтением значения. (При использовании Receiver::borrow() цикл может выполниться дважды с одним и тем же значением.)

Если получателя интересует только текущее значение и он не собирается ожидать изменений, можно использовать Receiver::borrow(). Использовать borrow может быть удобнее, поскольку это метод &self — для borrow_and_update требуется &mut self.

Примеры

Следующий пример выводит hello! world! .

use tokio::sync::watch;
use tokio::time::{Duration, sleep};

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

tokio::spawn(async move {
    // Use the equivalent of a "do-while" loop so the initial value is
    // processed before awaiting the `changed()` future.
    loop {
        println!("{}! ", *rx.borrow_and_update());
        if rx.changed().await.is_err() {
            break;
        }
    }
});

sleep(Duration::from_millis(100)).await;
tx.send("world")?;

Различие в условиях возврата ошибки у changed и has_changed.

use tokio::sync::watch;

let (tx, mut rx) = watch::channel("hello");
tx.send("goodbye").unwrap();
drop(tx);

// `has_changed` does not mark the value as seen and errors
// since the channel is closed.
assert!(rx.has_changed().is_err());

// `changed` returns Ok since the value is not already marked as seen
// even if the channel is closed.
assert!(rx.changed().await.is_ok());

// The `changed` call above marks the value as seen.
// The next `changed` call now returns an error as the channel is closed
// AND the current value is seen.
assert!(rx.changed().await.is_err());

Закрытие

Sender::is_closed и Sender::closed позволяют производителю определить, когда все дескрипторы Receiver удалены. Это означает, что дальнейший интерес к создаваемым значениям отсутствует и работу можно остановить.

Значение в канале не будет удалено, пока не будут удалены все отправители и все получатели.

Потокобезопасность

И Sender, и Receiver потокобезопасны. Их можно перемещать в другие потоки и использовать в конкурентной среде. Копии дескрипторов Receiver можно перемещать в отдельные потоки и также использовать параллельно.