Struct tokio::sync::broadcast::Receiver

pub struct Receiver<T> { /* fields omitted */ }

Receiving-half of the broadcast channel.

Must not be used concurrently. Messages may be retrieved using recv.

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

impl<T: Clone> Receiver<T>

pub async fn recv(&mut self) -> Result<T, RecvError>

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.

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());
}

pub fn try_recv(&mut self) -> Result<T, TryRecvError>

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);
}

pub fn into_stream(self) -> impl Stream<Item = Result<T, RecvError>>

Convert the receiver into a Stream.

The conversion allows using Receiver with APIs that require stream values.

Examples

use tokio::stream::StreamExt;
use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, rx) = broadcast::channel(128);

    tokio::spawn(async move {
        for i in 0..10_i32 {
            tx.send(i).unwrap();
        }
    });

    // Streams must be pinned to iterate.
    tokio::pin! {
        let stream = rx
            .into_stream()
            .filter(Result::is_ok)
            .map(Result::unwrap)
            .filter(|v| v % 2 == 0)
            .map(|v| v + 1);
    }

    while let Some(i) = stream.next().await {
        println!("{}", i);
    }
}

Trait Implementations

impl<T> Debug for Receiver<T>

fn fmt(&self, fmt: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T> Drop for Receiver<T>

fn drop(&mut self)

Executes the destructor for this type.

impl<T: Send> Send for Receiver<T>

impl<T: Send> Sync for Receiver<T>