Struct async_broadcast::Receiver

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

The receiving side of a channel.

Receivers can be cloned and shared among threads. When all (active) receivers associated with a channel are dropped, the channel becomes closed. You can deactivate a receiver using Receiver::deactivate if you would like the channel to remain open without keeping active receivers around.

Implementations§

source§

impl<T> Receiver<T>

source

pub fn capacity(&self) -> usize

Returns the channel capacity.

§Examples
use async_broadcast::broadcast;

let (_s, r) = broadcast::<i32>(5);
assert_eq!(r.capacity(), 5);
source

pub fn set_capacity(&mut self, new_cap: usize)

Set the channel capacity.

There are times when you need to change the channel’s capacity after creating it. If the new_cap is less than the number of messages in the channel, the oldest messages will be dropped to shrink the channel.

§Examples
use async_broadcast::{broadcast, TrySendError, TryRecvError};

let (s, mut r) = broadcast::<i32>(3);
assert_eq!(r.capacity(), 3);
s.try_broadcast(1).unwrap();
s.try_broadcast(2).unwrap();
s.try_broadcast(3).unwrap();

r.set_capacity(1);
assert_eq!(r.capacity(), 1);
assert_eq!(r.try_recv(), Err(TryRecvError::Overflowed(2)));
assert_eq!(r.try_recv().unwrap(), 3);
assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
s.try_broadcast(1).unwrap();
assert_eq!(s.try_broadcast(2), Err(TrySendError::Full(2)));

r.set_capacity(2);
assert_eq!(r.capacity(), 2);
s.try_broadcast(2).unwrap();
assert_eq!(s.try_broadcast(2), Err(TrySendError::Full(2)));
source

pub fn overflow(&self) -> bool

If overflow mode is enabled on this channel.

§Examples
use async_broadcast::broadcast;

let (_s, r) = broadcast::<i32>(5);
assert!(!r.overflow());
source

pub fn set_overflow(&mut self, overflow: bool)

Set overflow mode on the channel.

When overflow mode is set, broadcasting to the channel will succeed even if the channel is full. It achieves that by removing the oldest message from the channel.

§Examples
use async_broadcast::{broadcast, TrySendError, TryRecvError};

let (s, mut r) = broadcast::<i32>(2);
s.try_broadcast(1).unwrap();
s.try_broadcast(2).unwrap();
assert_eq!(s.try_broadcast(3), Err(TrySendError::Full(3)));
r.set_overflow(true);
assert_eq!(s.try_broadcast(3).unwrap(), Some(1));
assert_eq!(s.try_broadcast(4).unwrap(), Some(2));

assert_eq!(r.try_recv(), Err(TryRecvError::Overflowed(2)));
assert_eq!(r.try_recv().unwrap(), 3);
assert_eq!(r.try_recv().unwrap(), 4);
assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
source

pub fn await_active(&self) -> bool

If sender will wait for active receivers.

If set to false, Send will resolve immediately with a SendError. Defaults to true.

§Examples
use async_broadcast::broadcast;

let (_, r) = broadcast::<i32>(5);
assert!(r.await_active());
source

pub fn set_await_active(&mut self, await_active: bool)

Specify if sender will wait for active receivers.

If set to false, Send will resolve immediately with a SendError. Defaults to true.

§Examples
use async_broadcast::broadcast;

let (s, mut r) = broadcast::<i32>(2);
s.broadcast(1).await.unwrap();

r.set_await_active(false);
let _ = r.deactivate();
assert!(s.broadcast(2).await.is_err());
source

pub fn close(&self) -> bool

Closes the channel.

Returns true if this call has closed the channel and it was not closed already.

The remaining messages can still be received.

§Examples
use async_broadcast::{broadcast, RecvError};

let (s, mut r) = broadcast(1);
s.broadcast(1).await.unwrap();
assert!(s.close());

assert_eq!(r.recv().await.unwrap(), 1);
assert_eq!(r.recv().await, Err(RecvError::Closed));
source

pub fn is_closed(&self) -> bool

Returns true if the channel is closed.

§Examples
use async_broadcast::{broadcast, RecvError};

let (s, r) = broadcast::<()>(1);
assert!(!s.is_closed());

drop(r);
assert!(s.is_closed());
source

pub fn is_empty(&self) -> bool

Returns true if the channel is empty.

§Examples
use async_broadcast::broadcast;

let (s, r) = broadcast(1);

assert!(s.is_empty());
s.broadcast(1).await;
assert!(!s.is_empty());
source

pub fn is_full(&self) -> bool

Returns true if the channel is full.

§Examples
use async_broadcast::broadcast;

let (s, r) = broadcast(1);

assert!(!s.is_full());
s.broadcast(1).await;
assert!(s.is_full());
source

pub fn len(&self) -> usize

Returns the number of messages in the channel.

§Examples
use async_broadcast::broadcast;

let (s, r) = broadcast(2);
assert_eq!(s.len(), 0);

s.broadcast(1).await;
s.broadcast(2).await;
assert_eq!(s.len(), 2);
source

pub fn receiver_count(&self) -> usize

Returns the number of receivers for the channel.

This does not include inactive receivers. Use Receiver::inactive_receiver_count if you are interested in that.

§Examples
use async_broadcast::broadcast;

let (s, r) = broadcast::<()>(1);
assert_eq!(s.receiver_count(), 1);
let r = r.deactivate();
assert_eq!(s.receiver_count(), 0);

let r2 = r.activate_cloned();
assert_eq!(r.receiver_count(), 1);
assert_eq!(r.inactive_receiver_count(), 1);
source

pub fn inactive_receiver_count(&self) -> usize

Returns the number of inactive receivers for the channel.

§Examples
use async_broadcast::broadcast;

let (s, r) = broadcast::<()>(1);
assert_eq!(s.receiver_count(), 1);
let r = r.deactivate();
assert_eq!(s.receiver_count(), 0);

let r2 = r.activate_cloned();
assert_eq!(r.receiver_count(), 1);
assert_eq!(r.inactive_receiver_count(), 1);
source

pub fn sender_count(&self) -> usize

Returns the number of senders for the channel.

§Examples
use async_broadcast::broadcast;

let (s, r) = broadcast::<()>(1);
assert_eq!(s.sender_count(), 1);

let s2 = s.clone();
assert_eq!(s.sender_count(), 2);
source

pub fn deactivate(self) -> InactiveReceiver<T>

Downgrade to a InactiveReceiver.

An inactive receiver is one that can not and does not receive any messages. Its only purpose is keep the associated channel open even when there are no (active) receivers. An inactive receiver can be upgraded into a Receiver using InactiveReceiver::activate or InactiveReceiver::activate_cloned.

Sender::try_broadcast will return TrySendError::Inactive if only inactive receivers exists for the associated channel and Sender::broadcast will wait until an active receiver is available.

§Examples
use async_broadcast::{broadcast, TrySendError};

let (s, r) = broadcast(1);
let inactive = r.deactivate();
assert_eq!(s.try_broadcast(10), Err(TrySendError::Inactive(10)));

let mut r = inactive.activate();
assert_eq!(s.broadcast(10).await, Ok(None));
assert_eq!(r.recv().await, Ok(10));
source§

impl<T: Clone> Receiver<T>

source

pub fn recv(&mut self) -> Pin<Box<Recv<'_, T>>>

Receives a message from the channel.

If the channel is empty, this method waits until there is a message.

If the channel is closed, this method receives a message or returns an error if there are no more messages.

If this receiver has missed a message (only possible if overflow mode is enabled), then this method returns an error and readjusts its cursor to point to the first available message.

The future returned by this function is pinned to the heap. If the future being Unpin is not important to you, or if you just .await this future, use the [recv_direct] method instead.

§Examples
use async_broadcast::{broadcast, RecvError};

let (s, mut r1) = broadcast(1);
let mut r2 = r1.clone();

assert_eq!(s.broadcast(1).await, Ok(None));
drop(s);

assert_eq!(r1.recv().await, Ok(1));
assert_eq!(r1.recv().await, Err(RecvError::Closed));
assert_eq!(r2.recv().await, Ok(1));
assert_eq!(r2.recv().await, Err(RecvError::Closed));
source

pub fn recv_direct(&mut self) -> Recv<'_, T>

Receives a message from the channel without pinning the future to the heap.

The future returned by this method is not Unpin and must be pinned before use. This is the desired behavior if you just .await on the future. For other uses cases, use the [recv] method instead.

§Examples
use async_broadcast::{broadcast, RecvError};

let (s, mut r1) = broadcast(1);
let mut r2 = r1.clone();

assert_eq!(s.broadcast(1).await, Ok(None));
drop(s);

assert_eq!(r1.recv_direct().await, Ok(1));
assert_eq!(r1.recv_direct().await, Err(RecvError::Closed));
assert_eq!(r2.recv_direct().await, Ok(1));
assert_eq!(r2.recv_direct().await, Err(RecvError::Closed));
source

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

Attempts to receive a message from the channel.

If the channel is empty or closed, this method returns an error.

If this receiver has missed a message (only possible if overflow mode is enabled), then this method returns an error and readjusts its cursor to point to the first available message.

§Examples
use async_broadcast::{broadcast, TryRecvError};

let (s, mut r1) = broadcast(1);
let mut r2 = r1.clone();
assert_eq!(s.broadcast(1).await, Ok(None));

assert_eq!(r1.try_recv(), Ok(1));
assert_eq!(r1.try_recv(), Err(TryRecvError::Empty));
assert_eq!(r2.try_recv(), Ok(1));
assert_eq!(r2.try_recv(), Err(TryRecvError::Empty));

drop(s);
assert_eq!(r1.try_recv(), Err(TryRecvError::Closed));
assert_eq!(r2.try_recv(), Err(TryRecvError::Closed));
source

pub fn new_sender(&self) -> Sender<T>

Produce a new Sender for this channel.

This will not re-open the channel if it was closed due to all senders being dropped.

§Examples
use async_broadcast::{broadcast, RecvError};

let (s1, mut r) = broadcast(2);

assert_eq!(s1.broadcast(1).await, Ok(None));

let mut s2 = r.new_sender();

assert_eq!(s2.broadcast(2).await, Ok(None));
drop(s1);
drop(s2);

assert_eq!(r.recv().await, Ok(1));
assert_eq!(r.recv().await, Ok(2));
assert_eq!(r.recv().await, Err(RecvError::Closed));
source

pub fn new_receiver(&self) -> Self

Produce a new Receiver for this channel.

Unlike Receiver::clone, this method creates a new receiver that starts with zero messages available. This is slightly faster than a real clone.

§Examples
use async_broadcast::{broadcast, RecvError};

let (s, mut r1) = broadcast(2);

assert_eq!(s.broadcast(1).await, Ok(None));

let mut r2 = r1.new_receiver();

assert_eq!(s.broadcast(2).await, Ok(None));
drop(s);

assert_eq!(r1.recv().await, Ok(1));
assert_eq!(r1.recv().await, Ok(2));
assert_eq!(r1.recv().await, Err(RecvError::Closed));

assert_eq!(r2.recv().await, Ok(2));
assert_eq!(r2.recv().await, Err(RecvError::Closed));
source

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

A low level poll method that is similar to Receiver::recv() or Receiver::recv_direct(), and can be useful for building stream implementations which use a Receiver under the hood and want to know if the stream has overflowed.

Prefer to use Receiver::recv() or Receiver::recv_direct() when otherwise possible.

§Errors

If the number of messages that have been sent has overflowed the channel capacity, a RecvError::Overflowed variant is returned containing the number of items that overflowed and were lost.

§Examples

This example shows how the Receiver::poll_recv method can be used to allow a custom stream implementation to internally make use of a Receiver. This example implementation differs from the stream implementation of Receiver because it returns an error if the channel capacity overflows, which the built in Receiver stream doesn’t do.

use futures_core::Stream;
use async_broadcast::{Receiver, RecvError};
use std::{pin::Pin, task::{Poll, Context}};

struct MyStream(Receiver<i32>);

impl futures_core::Stream for MyStream {
    type Item = Result<i32, RecvError>;
    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        Pin::new(&mut self.0).poll_recv(cx)
    }
}

Trait Implementations§

source§

impl<T> Clone for Receiver<T>

source§

fn clone(&self) -> Self

Produce a clone of this Receiver that has the same messages queued.

§Examples
use async_broadcast::{broadcast, RecvError};

let (s, mut r1) = broadcast(1);

assert_eq!(s.broadcast(1).await, Ok(None));
drop(s);

let mut r2 = r1.clone();

assert_eq!(r1.recv().await, Ok(1));
assert_eq!(r1.recv().await, Err(RecvError::Closed));
assert_eq!(r2.recv().await, Ok(1));
assert_eq!(r2.recv().await, Err(RecvError::Closed));
1.0.0 · source§

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source. Read more
source§

impl<T: Debug> Debug for Receiver<T>

source§

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

Formats the value using the given formatter. Read more
source§

impl<T> Drop for Receiver<T>

source§

fn drop(&mut self)

Executes the destructor for this type. Read more
source§

impl<T: Clone> FusedStream for Receiver<T>

source§

fn is_terminated(&self) -> bool

Returns true if the stream should no longer be polled.
source§

impl<T: Clone> Stream for Receiver<T>

§

type Item = T

Values yielded by the stream.
source§

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_> ) -> Poll<Option<Self::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. Read more
source§

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

Returns the bounds on the remaining length of the stream. Read more

Auto Trait Implementations§

§

impl<T> Freeze for Receiver<T>

§

impl<T> RefUnwindSafe for Receiver<T>

§

impl<T> Send for Receiver<T>
where T: Send + Sync,

§

impl<T> Sync for Receiver<T>
where T: Send + Sync,

§

impl<T> Unpin for Receiver<T>

§

impl<T> UnwindSafe for Receiver<T>

Blanket Implementations§

source§

impl<T> Any for T
where T: 'static + ?Sized,

source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
source§

impl<T> Borrow<T> for T
where T: ?Sized,

source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
source§

impl<T> From<T> for T

source§

fn from(t: T) -> T

Returns the argument unchanged.

source§

impl<T, U> Into<U> for T
where U: From<T>,

source§

fn into(self) -> U

Calls U::from(self).

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

source§

impl<T> ToOwned for T
where T: Clone,

§

type Owned = T

The resulting type after obtaining ownership.
source§

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more
source§

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more
source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

§

type Error = Infallible

The type returned in the event of a conversion error.
source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
source§

impl<S, T, E> TryStream for S
where S: Stream<Item = Result<T, E>> + ?Sized,

§

type Ok = T

The type of successful values yielded by this future
§

type Error = E

The type of failures yielded by this future
source§

fn try_poll_next( self: Pin<&mut S>, cx: &mut Context<'_> ) -> Poll<Option<Result<<S as TryStream>::Ok, <S as TryStream>::Error>>>

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