Struct ReceiverStream 

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

A wrapper around tokio::sync::mpsc::Receiver that implements Stream.

Example

use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::StreamExt;

let (tx, rx) = mpsc::channel(2);
tx.send(10).await?;
tx.send(20).await?;
drop(tx);

let mut stream = ReceiverStream::new(rx);
assert_eq!(stream.next().await, Some(10));
assert_eq!(stream.next().await, Some(20));
assert_eq!(stream.next().await, None);

Implementations

impl<T> ReceiverStream<T>

pub fn new(recv: Receiver<T>) -> ReceiverStream<T>

Create a new ReceiverStream.

pub fn into_inner(self) -> Receiver<T>

Get back the inner Receiver.

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 no messages are dropped, after calling close(), you must receive all items from the stream until None is returned.

Trait Implementations

impl<T> AsMut<Receiver<T>> for ReceiverStream<T>

fn as_mut(&mut self) -> &mut Receiver<T>

Converts this type into a mutable reference of the (usually inferred) input type.

impl<T> AsRef<Receiver<T>> for ReceiverStream<T>

fn as_ref(&self) -> &Receiver<T>

Converts this type into a shared reference of the (usually inferred) input type.

impl<T> Debug for ReceiverStream<T>

where T: Debug,

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

Formats the value using the given formatter.

impl<T> From<Receiver<T>> for ReceiverStream<T>

fn from(recv: Receiver<T>) -> ReceiverStream<T>

Converts to this type from the input type.

impl<T> Stream for ReceiverStream<T>

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds of the stream based on the underlying receiver.

For open channels, it returns (receiver.len(), None).

For closed channels, it returns (receiver.len(), Some(used_capacity)) where used_capacity is calculated as receiver.max_capacity() - receiver.capacity(). This accounts for any Permit that is still able to send a message.

type Item = T

Values yielded by the stream.

fn poll_next( self: Pin<&mut ReceiverStream<T>>, cx: &mut Context<'_>, ) -> Poll<Option<<ReceiverStream<T> as Stream>::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.