pub struct Receiver<T> {
shared: Arc<Shared<T>>,
next: u64,
}
Expand description
Receiving-half of the broadcast
channel.
Must not be used concurrently. Messages may be retrieved using
recv
.
To turn this receiver into a Stream
, you can use the BroadcastStream
wrapper.
§Examples
use tokio::sync::broadcast;
#[tokio::main]
async fn main() {
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();
}
Fields§
State shared with all receivers and senders.
next: u64
Next position to read from
Implementations§
source§impl<T> Receiver<T>
impl<T> Receiver<T>
sourcepub fn len(&self) -> usize
pub fn len(&self) -> usize
Returns the number of messages that were sent into the channel and that
this Receiver
has yet to receive.
If the returned value from len
is larger than the next largest power of 2
of the capacity of the channel any call to recv
will return an
Err(RecvError::Lagged)
and any call to try_recv
will return an
Err(TryRecvError::Lagged)
, e.g. if the capacity of the channel is 10,
recv
will start to return Err(RecvError::Lagged)
once len
returns
values larger than 16.
§Examples
use tokio::sync::broadcast;
#[tokio::main]
async fn main() {
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);
}
sourcepub fn is_empty(&self) -> bool
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;
#[tokio::main]
async fn main() {
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());
}
sourcepub fn same_channel(&self, other: &Self) -> bool
pub fn same_channel(&self, other: &Self) -> bool
Returns true
if receivers belong to the same channel.
§Examples
use tokio::sync::broadcast;
#[tokio::main]
async fn main() {
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));
}
sourcefn recv_ref(
&mut self,
waiter: Option<(&UnsafeCell<Waiter>, &Waker)>,
) -> Result<RecvGuard<'_, T>, TryRecvError>
fn recv_ref( &mut self, waiter: Option<(&UnsafeCell<Waiter>, &Waker)>, ) -> Result<RecvGuard<'_, T>, TryRecvError>
Locks the next value if there is one.
source§impl<T: Clone> Receiver<T>
impl<T: Clone> Receiver<T>
sourcepub fn resubscribe(&self) -> Self
pub fn resubscribe(&self) -> Self
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;
#[tokio::main]
async fn main() {
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);
}
sourcepub async fn recv(&mut self) -> Result<T, RecvError>
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 will overwrite old values. At this point, a call to recv
will return with Err(RecvError::Lagged)
and the Receiver
’s
internal cursor is updated to point to the oldest value still held by
the channel. A subsequent call to recv
will return this value
unless it has been since overwritten.
§Cancel safety
This method is cancel safe. If recv
is used as the event in a
tokio::select!
statement and some other branch
completes first, it is guaranteed that no messages were received on this
channel.
§Examples
use tokio::sync::broadcast;
#[tokio::main]
async fn main() {
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;
#[tokio::main]
async fn main() {
let (tx, mut rx) = broadcast::channel(2);
tx.send(10).unwrap();
tx.send(20).unwrap();
tx.send(30).unwrap();
// The receiver lagged behind
assert!(rx.recv().await.is_err());
// At this point, we can abort or continue with lost messages
assert_eq!(20, rx.recv().await.unwrap());
assert_eq!(30, rx.recv().await.unwrap());
}
sourcepub fn try_recv(&mut self) -> Result<T, TryRecvError>
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 will overwrite old values. At this point, a call to recv
will return with Err(TryRecvError::Lagged)
and the Receiver
’s
internal cursor is updated to point to the oldest value still held by
the channel. A subsequent call to try_recv
will return this value
unless it has been since overwritten. If there are no values to
receive, Err(TryRecvError::Empty)
is returned.
§Examples
use tokio::sync::broadcast;
#[tokio::main]
async fn main() {
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);
}
sourcepub fn blocking_recv(&mut self) -> Result<T, RecvError>
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();
}