Struct Subscriber 

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

Thin wrapper for a cache subscriber, so as to trigger side-effects when all subscribers are gone.

The current side-effect is: auto-shrinking the cache when no more subscribers are active. This is an optimisation to reduce the number of data held in memory by the cache: when no more subscribers are active, all data are reduced to the minimum.

The side-effect takes effect on Drop.

Methods from Deref<Target = Receiver<T>>

pub fn len(&self) -> usize

Returns the number of messages that were sent into the channel and that this Receiver has yet to receive.

This count includes messages that have already been overwritten in the ring buffer and are no longer readable. If len is greater than the channel’s effective capacity (the provided capacity rounded up to the next power of two), the next call to recv returns Err(RecvError::Lagged) and the next call to try_recv returns Err(TryRecvError::Lagged). For example, with channel(10) the buffer length is 16, so lagging begins once len is larger than 16.

After a successful receive (including after handling Lagged and then reading retained messages), len decreases accordingly.

Examples

use tokio::sync::broadcast;

let (tx, mut rx1) = broadcast::channel(16);

tx.send(10).unwrap();
tx.send(20).unwrap();

assert_eq!(rx1.len(), 2);
assert_eq!(rx1.recv().await.unwrap(), 10);
assert_eq!(rx1.len(), 1);
assert_eq!(rx1.recv().await.unwrap(), 20);
assert_eq!(rx1.len(), 0);

pub fn is_empty(&self) -> bool

Returns true if there aren’t any messages in the channel that the Receiver has yet to receive.

Examples

use tokio::sync::broadcast;

let (tx, mut rx1) = broadcast::channel(16);

assert!(rx1.is_empty());

tx.send(10).unwrap();
tx.send(20).unwrap();

assert!(!rx1.is_empty());
assert_eq!(rx1.recv().await.unwrap(), 10);
assert_eq!(rx1.recv().await.unwrap(), 20);
assert!(rx1.is_empty());

pub fn same_channel(&self, other: &Receiver<T>) -> bool

Returns true if receivers belong to the same channel.

Examples

use tokio::sync::broadcast;

let (tx, rx) = broadcast::channel::<()>(16);
let rx2 = tx.subscribe();

assert!(rx.same_channel(&rx2));

let (_tx3, rx3) = broadcast::channel::<()>(16);

assert!(!rx3.same_channel(&rx2));

pub fn sender_strong_count(&self) -> usize

Returns the number of Sender handles.

pub fn sender_weak_count(&self) -> usize

Returns the number of WeakSender handles.

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.

Examples

use tokio::sync::broadcast;

let (tx, rx) = broadcast::channel::<()>(10);
assert!(!rx.is_closed());

drop(tx);

assert!(rx.is_closed());

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

Re-subscribes to the channel starting from the current tail element.

This Receiver handle will receive a clone of all values sent after it has resubscribed. This will not include elements that are in the queue of the current receiver. Consider the following example.

Examples

use tokio::sync::broadcast;

let (tx, mut rx) = broadcast::channel(2);

tx.send(1).unwrap();
let mut rx2 = rx.resubscribe();
tx.send(2).unwrap();

assert_eq!(rx2.recv().await.unwrap(), 2);
assert_eq!(rx.recv().await.unwrap(), 1);

pub async fn recv(&mut self) -> Result<T, RecvError>

Receives the next value for this receiver.

Each Receiver handle will receive a clone of all values sent after it has subscribed.

Err(RecvError::Closed) is returned when all Sender halves have dropped, indicating that no further values can be sent on the channel.

If the Receiver handle falls behind, once the channel is full, newly sent values overwrite old values in the ring buffer. The next call to recv then returns Err(RecvError::Lagged(n)), where n is the number of overwritten messages the receiver missed. The receiver stays subscribed; its internal cursor is advanced to the oldest value still held by the channel. A subsequent call to recv returns that value, unless further sends overwrite it before the receiver reads it. See lagging for details.

Cancel safety

This method is cancel safe. If recv is used as a branch in tokio::select! and another branch completes first, it is guaranteed that no messages were received on this channel.

Examples

use tokio::sync::broadcast;

let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();

tokio::spawn(async move {
    assert_eq!(rx1.recv().await.unwrap(), 10);
    assert_eq!(rx1.recv().await.unwrap(), 20);
});

tokio::spawn(async move {
    assert_eq!(rx2.recv().await.unwrap(), 10);
    assert_eq!(rx2.recv().await.unwrap(), 20);
});

tx.send(10).unwrap();
tx.send(20).unwrap();

Handling lag

use tokio::sync::broadcast;
use tokio::sync::broadcast::error::RecvError;

let (tx, mut rx) = broadcast::channel(2);

tx.send(10).unwrap();
tx.send(20).unwrap();
tx.send(30).unwrap();

// One message was overwritten before this receiver could read it.
assert!(matches!(rx.recv().await, Err(RecvError::Lagged(1))));

// Resume from the oldest retained message, or abort the task instead.
assert_eq!(20, rx.recv().await.unwrap());
assert_eq!(30, rx.recv().await.unwrap());

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

Attempts to return a pending value on this receiver without awaiting.

This is useful for a flavor of “optimistic check” before deciding to await on a receiver.

Compared with recv, this function has three failure cases instead of two (one for closed, one for an empty buffer, one for a lagging receiver).

Err(TryRecvError::Closed) is returned when all Sender halves have dropped, indicating that no further values can be sent on the channel.

If the Receiver handle falls behind, once the channel is full, newly sent values overwrite old values in the ring buffer. The next call to try_recv then returns Err(TryRecvError::Lagged(n)), where n is the number of overwritten messages the receiver missed. The receiver stays subscribed; its internal cursor is advanced to the oldest value still held by the channel. A subsequent call to try_recv returns that value, unless further sends overwrite it before the receiver reads it. If there are no values to receive, Err(TryRecvError::Empty) is returned. See lagging for details.

Examples

use tokio::sync::broadcast;

let (tx, mut rx) = broadcast::channel(16);

assert!(rx.try_recv().is_err());

tx.send(10).unwrap();

let value = rx.try_recv().unwrap();
assert_eq!(10, value);

pub fn blocking_recv(&mut self) -> Result<T, RecvError>

Blocking receive to call outside of asynchronous contexts.

Panics

This function panics if called within an asynchronous execution context.

Examples

use std::thread;
use tokio::sync::broadcast;

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

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

    let _ = tx.send(10);
    sync_code.join().unwrap();
}

Trait Implementations

impl<T> Deref for Subscriber<T>

type Target = Receiver<T>

The resulting type after dereferencing.

fn deref(&self) -> &Self::Target

Dereferences the value.

impl<T> DerefMut for Subscriber<T>

fn deref_mut(&mut self) -> &mut Self::Target

Mutably dereferences the value.

impl<T> Drop for Subscriber<T>

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.