Struct tokio::sync::mpsc::Receiver

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

Receives values from the associated Sender.

Instances are created by the channel function.

This receiver can be turned into a Stream using ReceiverStream.

Implementations

impl<T> Receiver<T>

pub async fn recv(&mut self) -> Option<T>

Receives the next value for this receiver.

This method returns None if the channel has been closed and there are no remaining messages in the channel’s buffer. This indicates that no further values can ever be received from this Receiver. The channel is closed when all senders have been dropped, or when close is called.

If there are no messages in the channel’s buffer, but the channel has not yet been closed, this method will sleep until a message is sent or the channel is closed. Note that if close is called, but there are still outstanding Permits from before it was closed, the channel is not considered closed by recv until the permits are released.

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::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(100);

    tokio::spawn(async move {
        tx.send("hello").await.unwrap();
    });

    assert_eq!(Some("hello"), rx.recv().await);
    assert_eq!(None, rx.recv().await);
}

Values are buffered:

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(100);

    tx.send("hello").await.unwrap();
    tx.send("world").await.unwrap();

    assert_eq!(Some("hello"), rx.recv().await);
    assert_eq!(Some("world"), rx.recv().await);
}

pub async fn recv_many(&mut self, buffer: &mut Vec<T>, limit: usize) -> usize

Receives the next values for this receiver and extends buffer.

This method extends buffer by no more than a fixed number of values as specified by limit. If limit is zero, the function immediately returns 0. The return value is the number of values added to buffer.

For limit > 0, if there are no messages in the channel’s queue, but the channel has not yet been closed, this method will sleep until a message is sent or the channel is closed. Note that if close is called, but there are still outstanding Permits from before it was closed, the channel is not considered closed by recv_many until the permits are released.

For non-zero values of limit, this method will never return 0 unless the channel has been closed and there are no remaining messages in the channel’s queue. This indicates that no further values can ever be received from this Receiver. The channel is closed when all senders have been dropped, or when close is called.

The capacity of buffer is increased as needed.

Cancel safety

This method is cancel safe. If recv_many 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::mpsc;

#[tokio::main]
async fn main() {
    let mut buffer: Vec<&str> = Vec::with_capacity(2);
    let limit = 2;
    let (tx, mut rx) = mpsc::channel(100);
    let tx2 = tx.clone();
    tx2.send("first").await.unwrap();
    tx2.send("second").await.unwrap();
    tx2.send("third").await.unwrap();

    // Call `recv_many` to receive up to `limit` (2) values.
    assert_eq!(2, rx.recv_many(&mut buffer, limit).await);
    assert_eq!(vec!["first", "second"], buffer);

    // If the buffer is full, the next call to `recv_many`
    // reserves additional capacity.
    assert_eq!(1, rx.recv_many(&mut buffer, 1).await);

    tokio::spawn(async move {
        tx.send("fourth").await.unwrap();
    });

    // 'tx' is dropped, but `recv_many`
    // is guaranteed not to return 0 as the channel
    // is not yet closed.
    assert_eq!(1, rx.recv_many(&mut buffer, 1).await);
    assert_eq!(vec!["first", "second", "third", "fourth"], buffer);

    // Once the last sender is dropped, the channel is
    // closed and `recv_many` returns 0, capacity unchanged.
    drop(tx2);
    assert_eq!(0, rx.recv_many(&mut buffer, limit).await);
    assert_eq!(vec!["first", "second", "third", "fourth"], buffer);
}

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

Tries to receive the next value for this receiver.

This method returns the Empty error if the channel is currently empty, but there are still outstanding senders or permits.

This method returns the Disconnected error if the channel is currently empty, and there are no outstanding senders or permits.

Unlike the poll_recv method, this method will never return an Empty error spuriously.

Examples

use tokio::sync::mpsc;
use tokio::sync::mpsc::error::TryRecvError;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(100);

    tx.send("hello").await.unwrap();

    assert_eq!(Ok("hello"), rx.try_recv());
    assert_eq!(Err(TryRecvError::Empty), rx.try_recv());

    tx.send("hello").await.unwrap();
    // Drop the last sender, closing the channel.
    drop(tx);

    assert_eq!(Ok("hello"), rx.try_recv());
    assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());
}

pub fn blocking_recv(&mut self) -> Option<T>

Blocking receive to call outside of asynchronous contexts.

This method returns None if the channel has been closed and there are no remaining messages in the channel’s buffer. This indicates that no further values can ever be received from this Receiver. The channel is closed when all senders have been dropped, or when close is called.

If there are no messages in the channel’s buffer, but the channel has not yet been closed, this method will block until a message is sent or the channel is closed.

This method is intended for use cases where you are sending from asynchronous code to synchronous code, and will work even if the sender is not using blocking_send to send the message.

Note that if close is called, but there are still outstanding Permits from before it was closed, the channel is not considered closed by blocking_recv until the permits are released.

Panics

This function panics if called within an asynchronous execution context.

Examples

use std::thread;
use tokio::runtime::Runtime;
use tokio::sync::mpsc;

fn main() {
    let (tx, mut rx) = mpsc::channel::<u8>(10);

    let sync_code = thread::spawn(move || {
        assert_eq!(Some(10), rx.blocking_recv());
    });

    Runtime::new()
        .unwrap()
        .block_on(async move {
            let _ = tx.send(10).await;
        });
    sync_code.join().unwrap()
}

pub fn close(&mut self)

Closes the receiving half of a channel without dropping it.

This prevents any further messages from being sent on the channel while still enabling the receiver to drain messages that are buffered. Any outstanding Permit values will still be able to send messages.

To guarantee that no messages are dropped, after calling close(), recv() must be called until None is returned. If there are outstanding Permit or OwnedPermit values, the recv method will not return None until those are released.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(20);

    tokio::spawn(async move {
        let mut i = 0;
        while let Ok(permit) = tx.reserve().await {
            permit.send(i);
            i += 1;
        }
    });

    rx.close();

    while let Some(msg) = rx.recv().await {
        println!("got {}", msg);
    }

    // Channel closed and no messages are lost.
}

pub fn is_closed(&self) -> bool

Checks if a channel is closed.

This method returns true if the channel has been closed. The channel is closed when all Sender have been dropped, or when Receiver::close is called.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (_tx, mut rx) = mpsc::channel::<()>(10);
    assert!(!rx.is_closed());

    rx.close();

    assert!(rx.is_closed());
}

pub fn is_empty(&self) -> bool

Checks if a channel is empty.

This method returns true if the channel has no messages.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(10);
    assert!(rx.is_empty());

    tx.send(0).await.unwrap();
    assert!(!rx.is_empty());
}

pub fn len(&self) -> usize

Returns the number of messages in the channel.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(10);
    assert_eq!(0, rx.len());

    tx.send(0).await.unwrap();
    assert_eq!(1, rx.len());
}

pub fn capacity(&self) -> usize

Returns the current capacity of the channel.

The capacity goes down when the sender sends a value by calling Sender::send or by reserving capacity with Sender::reserve. The capacity goes up when values are received. This is distinct from max_capacity, which always returns buffer capacity initially specified when calling channel.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<()>(5);

    assert_eq!(rx.capacity(), 5);

    // Making a reservation drops the capacity by one.
    let permit = tx.reserve().await.unwrap();
    assert_eq!(rx.capacity(), 4);
    assert_eq!(rx.len(), 0);

    // Sending and receiving a value increases the capacity by one.
    permit.send(());
    assert_eq!(rx.len(), 1);
    rx.recv().await.unwrap();
    assert_eq!(rx.capacity(), 5);

    // Directly sending a message drops the capacity by one.
    tx.send(()).await.unwrap();
    assert_eq!(rx.capacity(), 4);
    assert_eq!(rx.len(), 1);

    // Receiving the message increases the capacity by one.
    rx.recv().await.unwrap();
    assert_eq!(rx.capacity(), 5);
    assert_eq!(rx.len(), 0);
}

pub fn max_capacity(&self) -> usize

Returns the maximum buffer capacity of the channel.

The maximum capacity is the buffer capacity initially specified when calling channel. This is distinct from capacity, which returns the current available buffer capacity: as messages are sent and received, the value returned by capacity will go up or down, whereas the value returned by max_capacity will remain constant.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel::<()>(5);

    // both max capacity and capacity are the same at first
    assert_eq!(rx.max_capacity(), 5);
    assert_eq!(rx.capacity(), 5);

    // Making a reservation doesn't change the max capacity.
    let permit = tx.reserve().await.unwrap();
    assert_eq!(rx.max_capacity(), 5);
    // but drops the capacity by one
    assert_eq!(rx.capacity(), 4);
}

pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>>

Polls to receive the next message on this channel.

This method returns:

• Poll::Pending if no messages are available but the channel is not closed, or if a spurious failure happens.
• Poll::Ready(Some(message)) if a message is available.
• Poll::Ready(None) if the channel has been closed and all messages sent before it was closed have been received.

When the method returns Poll::Pending, the Waker in the provided Context is scheduled to receive a wakeup when a message is sent on any receiver, or when the channel is closed. Note that on multiple calls to poll_recv or poll_recv_many, only the Waker from the Context passed to the most recent call is scheduled to receive a wakeup.

If this method returns Poll::Pending due to a spurious failure, then the Waker will be notified when the situation causing the spurious failure has been resolved. Note that receiving such a wakeup does not guarantee that the next call will succeed — it could fail with another spurious failure.

pub fn poll_recv_many( &mut self, cx: &mut Context<'_>, buffer: &mut Vec<T>, limit: usize, ) -> Poll<usize>

Polls to receive multiple messages on this channel, extending the provided buffer.

This method returns:

• Poll::Pending if no messages are available but the channel is not closed, or if a spurious failure happens.
• Poll::Ready(count) where count is the number of messages successfully received and stored in buffer. This can be less than, or equal to, limit.
• Poll::Ready(0) if limit is set to zero or when the channel is closed.

When the method returns Poll::Pending, the Waker in the provided Context is scheduled to receive a wakeup when a message is sent on any receiver, or when the channel is closed. Note that on multiple calls to poll_recv or poll_recv_many, only the Waker from the Context passed to the most recent call is scheduled to receive a wakeup.

Note that this method does not guarantee that exactly limit messages are received. Rather, if at least one message is available, it returns as many messages as it can up to the given limit. This method returns zero only if the channel is closed (or if limit is zero).

Examples

use std::task::{Context, Poll};
use std::pin::Pin;
use tokio::sync::mpsc;
use futures::Future;

struct MyReceiverFuture<'a> {
    receiver: mpsc::Receiver<i32>,
    buffer: &'a mut Vec<i32>,
    limit: usize,
}

impl<'a> Future for MyReceiverFuture<'a> {
    type Output = usize; // Number of messages received

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let MyReceiverFuture { receiver, buffer, limit } = &mut *self;

        // Now `receiver` and `buffer` are mutable references, and `limit` is copied
        match receiver.poll_recv_many(cx, *buffer, *limit) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(count) => Poll::Ready(count),
        }
    }
}

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(32);
    let mut buffer = Vec::new();

    let my_receiver_future = MyReceiverFuture {
        receiver: rx,
        buffer: &mut buffer,
        limit: 3,
    };

    for i in 0..10 {
        tx.send(i).await.unwrap();
    }

    let count = my_receiver_future.await;
    assert_eq!(count, 3);
    assert_eq!(buffer, vec![0,1,2])
}

Trait Implementations

impl<T> Debug for Receiver<T>

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

Formats the value using the given formatter.

impl<T> Unpin for Receiver<T>