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>
impl<T> Receiver<T>
sourcepub fn capacity(&self) -> usize
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);
sourcepub fn set_capacity(&mut self, new_cap: usize)
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)));
sourcepub fn overflow(&self) -> bool
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());
sourcepub fn set_overflow(&mut self, overflow: bool)
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));
sourcepub fn await_active(&self) -> bool
pub fn await_active(&self) -> bool
sourcepub fn set_await_active(&mut self, await_active: bool)
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());
sourcepub fn close(&self) -> bool
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));
sourcepub fn is_closed(&self) -> bool
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());
sourcepub fn is_empty(&self) -> bool
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());
sourcepub fn is_full(&self) -> bool
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());
sourcepub fn len(&self) -> usize
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);
sourcepub fn receiver_count(&self) -> usize
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);
sourcepub fn inactive_receiver_count(&self) -> usize
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);
sourcepub fn sender_count(&self) -> usize
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);
sourcepub fn deactivate(self) -> InactiveReceiver<T>
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>
impl<T: Clone> Receiver<T>
sourcepub fn recv(&mut self) -> Pin<Box<Recv<'_, T>>>
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));
sourcepub fn recv_direct(&mut self) -> Recv<'_, T> ⓘ
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));
sourcepub fn try_recv(&mut self) -> Result<T, TryRecvError>
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));
sourcepub fn new_sender(&self) -> Sender<T>
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));
sourcepub fn new_receiver(&self) -> Self
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));
sourcepub fn poll_recv(
self: Pin<&mut Self>,
cx: &mut Context<'_>
) -> Poll<Option<Result<T, RecvError>>>
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>
impl<T> Clone for Receiver<T>
source§fn clone(&self) -> Self
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)
fn clone_from(&mut self, source: &Self)
source
. Read moresource§impl<T: Clone> FusedStream for Receiver<T>
impl<T: Clone> FusedStream for Receiver<T>
source§fn is_terminated(&self) -> bool
fn is_terminated(&self) -> bool
true
if the stream should no longer be polled.