Struct async_broadcast::Sender

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

The sending side of the broadcast channel.

Senders can be cloned and shared among threads. When all senders associated with a channel are dropped, the channel becomes closed.

The channel can also be closed manually by calling Sender::close().

Implementations

impl<T> Sender<T>

pub fn capacity(&self) -> usize

Returns the channel capacity.

Examples

use async_broadcast::broadcast;

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

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 (mut s, mut r) = broadcast::<i32>(3);
assert_eq!(s.capacity(), 3);
s.try_broadcast(1).unwrap();
s.try_broadcast(2).unwrap();
s.try_broadcast(3).unwrap();

s.set_capacity(1);
assert_eq!(s.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)));

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

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!(!s.overflow());

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 (mut 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)));
s.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));

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 (s, _) = broadcast::<i32>(5);
assert!(s.await_active());

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 (mut s, mut r) = broadcast::<i32>(2);
s.broadcast(1).await.unwrap();

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

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

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

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

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

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

pub fn receiver_count(&self) -> usize

Returns the number of receivers for the channel.

This does not include inactive receivers. Use Sender::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);

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

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

pub fn new_receiver(&self) -> Receiver<T>

Produce a new Receiver for this channel.

The new receiver starts with zero messages available. This will not re-open the channel if it was closed due to all receivers being dropped.

Examples

use async_broadcast::{broadcast, RecvError};

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

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

let mut r2 = s.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));

impl<T: Clone> Sender<T>

pub fn broadcast(&self, msg: T) -> Send<'_, T>

Broadcasts a message on the channel.

If the channel is full, this method waits until there is space for a message unless:

1. overflow mode (set through Sender::set_overflow) is enabled, in which case it removes the oldest message from the channel to make room for the new message. The removed message is returned to the caller.
2. this behavior is disabled using Sender::set_await_active, in which case, it returns SendError immediately.

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

Examples

use async_broadcast::{broadcast, SendError};

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

assert_eq!(s.broadcast(1).await, Ok(None));
drop(r);
assert_eq!(s.broadcast(2).await, Err(SendError(2)));

pub fn try_broadcast(&self, msg: T) -> Result<Option<T>, TrySendError<T>>

Attempts to broadcast a message on the channel.

If the channel is full, this method returns an error unless overflow mode (set through Sender::set_overflow) is enabled. If the overflow mode is enabled, it removes the oldest message from the channel to make room for the new message. The removed message is returned to the caller.

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

Examples

use async_broadcast::{broadcast, TrySendError};

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

assert_eq!(s.try_broadcast(1), Ok(None));
assert_eq!(s.try_broadcast(2), Err(TrySendError::Full(2)));

drop(r);
assert_eq!(s.try_broadcast(3), Err(TrySendError::Closed(3)));

Examples found in repository

src/lib.rs (line 1547)

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut this = Pin::new(self);

        loop {
            let msg = this.msg.take().unwrap();
            let inner = &this.sender.inner;

            // Attempt to send a message.
            match this.sender.try_broadcast(msg) {
                Ok(msg) => {
                    let inner = inner.write();

                    if inner.queue.len() < inner.capacity {
                        // Not full still, so notify the next awaiting sender.
                        inner.send_ops.notify(1);
                    }

                    return Poll::Ready(Ok(msg));
                }
                Err(TrySendError::Closed(msg)) => return Poll::Ready(Err(SendError(msg))),
                Err(TrySendError::Full(m)) => this.msg = Some(m),
                Err(TrySendError::Inactive(m)) if inner.read().await_active => this.msg = Some(m),
                Err(TrySendError::Inactive(m)) => return Poll::Ready(Err(SendError(m))),
            }

            // Sending failed - now start listening for notifications or wait for one.
            match &mut this.listener {
                None => {
                    // Start listening and then try sending again.
                    let inner = inner.write();
                    this.listener = Some(inner.send_ops.listen());
                }
                Some(l) => {
                    // Wait for a notification.
                    ready!(Pin::new(l).poll(cx));
                    this.listener = None;
                }
            }
        }
    }

Trait Implementations

impl<T> Clone for Sender<T>

fn clone(&self) -> Self

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<T: Debug> Debug for Sender<T>

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

Formats the value using the given formatter.

impl<T> Drop for Sender<T>

fn drop(&mut self)

Executes the destructor for this type.