pub struct Receiver<T> { /* private fields */ }
Expand description
Receives values from the associated Sender
.
Instances are created by the channel
function.
To turn this receiver into a Stream
, you can use the WatchStream
wrapper.
Implementations§
Source§impl<T> Receiver<T>
impl<T> Receiver<T>
Sourcepub fn borrow(&self) -> Ref<'_, T>
pub fn borrow(&self) -> Ref<'_, T>
Returns a reference to the most recently sent value.
This method does not mark the returned value as seen, so future calls to
changed
may return immediately even if you have already seen the
value with a call to borrow
.
Outstanding borrows hold a read lock on the inner value. This means that
long-lived borrows could cause the producer half to block. It is recommended
to keep the borrow as short-lived as possible. Additionally, if you are
running in an environment that allows !Send
futures, you must ensure that
the returned Ref
type is never held alive across an .await
point,
otherwise, it can lead to a deadlock.
The priority policy of the lock is dependent on the underlying lock
implementation, and this type does not guarantee that any particular policy
will be used. In particular, a producer which is waiting to acquire the lock
in send
might or might not block concurrent calls to borrow
, e.g.:
Potential deadlock example
// Task 1 (on thread A) | // Task 2 (on thread B)
let _ref1 = rx.borrow(); |
| // will block
| let _ = tx.send(());
// may deadlock |
let _ref2 = rx.borrow(); |
For more information on when to use this method versus
borrow_and_update
, see here.
§Examples
use tokio::sync::watch;
let (_, rx) = watch::channel("hello");
assert_eq!(*rx.borrow(), "hello");
Sourcepub fn borrow_and_update(&mut self) -> Ref<'_, T>
pub fn borrow_and_update(&mut self) -> Ref<'_, T>
Returns a reference to the most recently sent value and marks that value as seen.
This method marks the current value as seen. Subsequent calls to changed
will not return immediately until the Sender
has modified the shared
value again.
Outstanding borrows hold a read lock on the inner value. This means that
long-lived borrows could cause the producer half to block. It is recommended
to keep the borrow as short-lived as possible. Additionally, if you are
running in an environment that allows !Send
futures, you must ensure that
the returned Ref
type is never held alive across an .await
point,
otherwise, it can lead to a deadlock.
The priority policy of the lock is dependent on the underlying lock
implementation, and this type does not guarantee that any particular policy
will be used. In particular, a producer which is waiting to acquire the lock
in send
might or might not block concurrent calls to borrow
, e.g.:
Potential deadlock example
// Task 1 (on thread A) | // Task 2 (on thread B)
let _ref1 = rx1.borrow_and_update(); |
| // will block
| let _ = tx.send(());
// may deadlock |
let _ref2 = rx2.borrow_and_update(); |
For more information on when to use this method versus borrow
, see
here.
Sourcepub fn has_changed(&self) -> Result<bool, RecvError>
pub fn has_changed(&self) -> Result<bool, RecvError>
Checks if this channel contains a message that this receiver has not yet seen. The new value is not marked as seen.
Although this method is called has_changed
, it does not check new
messages for equality, so this call will return true even if the new
message is equal to the old message.
Returns an error if the channel has been closed.
§Examples
use tokio::sync::watch;
#[tokio::main]
async fn main() {
let (tx, mut rx) = watch::channel("hello");
tx.send("goodbye").unwrap();
assert!(rx.has_changed().unwrap());
assert_eq!(*rx.borrow_and_update(), "goodbye");
// The value has been marked as seen
assert!(!rx.has_changed().unwrap());
drop(tx);
// The `tx` handle has been dropped
assert!(rx.has_changed().is_err());
}
Sourcepub fn mark_changed(&mut self)
pub fn mark_changed(&mut self)
Marks the state as changed.
After invoking this method has_changed()
returns true
and changed()
returns
immediately, regardless of whether a new value has been sent.
This is useful for triggering an initial change notification after subscribing to synchronize new receivers.
Sourcepub fn mark_unchanged(&mut self)
pub fn mark_unchanged(&mut self)
Marks the state as unchanged.
The current value will be considered seen by the receiver.
This is useful if you are not interested in the current value visible in the receiver.
Sourcepub async fn changed(&mut self) -> Result<(), RecvError>
pub async fn changed(&mut self) -> Result<(), RecvError>
Waits for a change notification, then marks the newest value as seen.
If the newest value in the channel has not yet been marked seen when
this method is called, the method marks that value seen and returns
immediately. If the newest value has already been marked seen, then the
method sleeps until a new message is sent by the Sender
connected to
this Receiver
, or until the Sender
is dropped.
This method returns an error if and only if the Sender
is dropped.
For more information, see Change notifications in the module-level documentation.
§Cancel safety
This method is cancel safe. If you use it as the event in a
tokio::select!
statement and some other branch
completes first, then it is guaranteed that no values have been marked
seen by this call to changed
.
§Examples
use tokio::sync::watch;
#[tokio::main]
async fn main() {
let (tx, mut rx) = watch::channel("hello");
tokio::spawn(async move {
tx.send("goodbye").unwrap();
});
assert!(rx.changed().await.is_ok());
assert_eq!(*rx.borrow_and_update(), "goodbye");
// The `tx` handle has been dropped
assert!(rx.changed().await.is_err());
}
Sourcepub async fn wait_for(
&mut self,
f: impl FnMut(&T) -> bool,
) -> Result<Ref<'_, T>, RecvError>
pub async fn wait_for( &mut self, f: impl FnMut(&T) -> bool, ) -> Result<Ref<'_, T>, RecvError>
Waits for a value that satisfies the provided condition.
This method will call the provided closure whenever something is sent on
the channel. Once the closure returns true
, this method will return a
reference to the value that was passed to the closure.
Before wait_for
starts waiting for changes, it will call the closure
on the current value. If the closure returns true
when given the
current value, then wait_for
will immediately return a reference to
the current value. This is the case even if the current value is already
considered seen.
The watch channel only keeps track of the most recent value, so if
several messages are sent faster than wait_for
is able to call the
closure, then it may skip some updates. Whenever the closure is called,
it will be called with the most recent value.
When this function returns, the value that was passed to the closure
when it returned true
will be considered seen.
If the channel is closed, then wait_for
will return a RecvError
.
Once this happens, no more messages can ever be sent on the channel.
When an error is returned, it is guaranteed that the closure has been
called on the last value, and that it returned false
for that value.
(If the closure returned true
, then the last value would have been
returned instead of the error.)
Like the borrow
method, the returned borrow holds a read lock on the
inner value. This means that long-lived borrows could cause the producer
half to block. It is recommended to keep the borrow as short-lived as
possible. See the documentation of borrow
for more information on
this.
§Examples
use tokio::sync::watch;
#[tokio::main]
async fn main() {
let (tx, _rx) = watch::channel("hello");
tx.send("goodbye").unwrap();
// here we subscribe to a second receiver
// now in case of using `changed` we would have
// to first check the current value and then wait
// for changes or else `changed` would hang.
let mut rx2 = tx.subscribe();
// in place of changed we have use `wait_for`
// which would automatically check the current value
// and wait for changes until the closure returns true.
assert!(rx2.wait_for(|val| *val == "goodbye").await.is_ok());
assert_eq!(*rx2.borrow(), "goodbye");
}
Sourcepub fn same_channel(&self, other: &Self) -> bool
pub fn same_channel(&self, other: &Self) -> bool
Returns true
if receivers belong to the same channel.
§Examples
let (tx, rx) = tokio::sync::watch::channel(true);
let rx2 = rx.clone();
assert!(rx.same_channel(&rx2));
let (tx3, rx3) = tokio::sync::watch::channel(true);
assert!(!rx3.same_channel(&rx2));