Struct tokio::sync::broadcast::Receiver [−][src]
pub struct Receiver<T> { /* fields omitted */ }Expand description
Receiving-half of the broadcast channel.
Must not be used concurrently. Messages may be retrieved using
recv.
To turn this receiver into a Stream, you can use the BroadcastStream
wrapper.
Examples
use tokio::sync::broadcast;
#[tokio::main]
async fn main() {
let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();
tokio::spawn(async move {
assert_eq!(rx1.recv().await.unwrap(), 10);
assert_eq!(rx1.recv().await.unwrap(), 20);
});
tokio::spawn(async move {
assert_eq!(rx2.recv().await.unwrap(), 10);
assert_eq!(rx2.recv().await.unwrap(), 20);
});
tx.send(10).unwrap();
tx.send(20).unwrap();
}Implementations
Receives the next value for this receiver.
Each Receiver handle will receive a clone of all values sent
after it has subscribed.
Err(RecvError::Closed) is returned when all Sender halves have
dropped, indicating that no further values can be sent on the channel.
If the Receiver handle falls behind, once the channel is full, newly
sent values will overwrite old values. At this point, a call to recv
will return with Err(RecvError::Lagged) and the Receiver’s
internal cursor is updated to point to the oldest value still held by
the channel. A subsequent call to recv will return this value
unless it has been since overwritten.
Cancel safety
This method is cancel safe. If recv is used as the event in a
tokio::select! statement and some other branch
completes first, it is guaranteed that no messages were received on this
channel.
Examples
use tokio::sync::broadcast;
#[tokio::main]
async fn main() {
let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();
tokio::spawn(async move {
assert_eq!(rx1.recv().await.unwrap(), 10);
assert_eq!(rx1.recv().await.unwrap(), 20);
});
tokio::spawn(async move {
assert_eq!(rx2.recv().await.unwrap(), 10);
assert_eq!(rx2.recv().await.unwrap(), 20);
});
tx.send(10).unwrap();
tx.send(20).unwrap();
}Handling lag
use tokio::sync::broadcast;
#[tokio::main]
async fn main() {
let (tx, mut rx) = broadcast::channel(2);
tx.send(10).unwrap();
tx.send(20).unwrap();
tx.send(30).unwrap();
// The receiver lagged behind
assert!(rx.recv().await.is_err());
// At this point, we can abort or continue with lost messages
assert_eq!(20, rx.recv().await.unwrap());
assert_eq!(30, rx.recv().await.unwrap());
}Attempts to return a pending value on this receiver without awaiting.
This is useful for a flavor of “optimistic check” before deciding to await on a receiver.
Compared with recv, this function has three failure cases instead of two
(one for closed, one for an empty buffer, one for a lagging receiver).
Err(TryRecvError::Closed) is returned when all Sender halves have
dropped, indicating that no further values can be sent on the channel.
If the Receiver handle falls behind, once the channel is full, newly
sent values will overwrite old values. At this point, a call to recv
will return with Err(TryRecvError::Lagged) and the Receiver’s
internal cursor is updated to point to the oldest value still held by
the channel. A subsequent call to try_recv will return this value
unless it has been since overwritten. If there are no values to
receive, Err(TryRecvError::Empty) is returned.
Examples
use tokio::sync::broadcast;
#[tokio::main]
async fn main() {
let (tx, mut rx) = broadcast::channel(16);
assert!(rx.try_recv().is_err());
tx.send(10).unwrap();
let value = rx.try_recv().unwrap();
assert_eq!(10, value);
}