Struct tokio::sync::broadcast::Receiver

source · [−]
pub struct Receiver<T> { /* private fields */ }
Expand description

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

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

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

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

Formats the value using the given formatter. Read more

Executes the destructor for this type. Read more

Auto Trait Implementations

Blanket Implementations

Gets the TypeId of self. Read more

Immutably borrows from an owned value. Read more

Mutably borrows from an owned value. Read more

Returns the argument unchanged.

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Consumes and returns the next value in the stream or None if the stream is finished. Read more

Consumes and returns the next item in the stream. If an error is encountered before the next item, the error is returned instead. Read more

Maps this stream’s items to a different type, returning a new stream of the resulting type. Read more

Combine two streams into one by interleaving the output of both as it is produced. Read more

Filters the values produced by this stream according to the provided predicate. Read more

Filters the values produced by this stream while simultaneously mapping them to a different type according to the provided closure. Read more

Creates a stream which ends after the first None. Read more

Creates a new stream of at most n items of the underlying stream. Read more

Take elements from this stream while the provided predicate resolves to true. Read more

Creates a new stream that will skip the n first items of the underlying stream. Read more

Skip elements from the underlying stream while the provided predicate resolves to true. Read more

Tests if every element of the stream matches a predicate. Read more

Tests if any element of the stream matches a predicate. Read more

Combine two streams into one by first returning all values from the first stream then all values from the second stream. Read more

A combinator that applies a function to every element in a stream producing a single, final value. Read more

Drain stream pushing all emitted values into a collection. Read more

Applies a per-item timeout to the passed stream. Read more

The type returned in the event of a conversion error.

Performs the conversion.

The type returned in the event of a conversion error.

Performs the conversion.

The type of successful values yielded by this future

The type of failures yielded by this future

Poll this TryStream as if it were a Stream. Read more