Source code
Revision control
Copy as Markdown
Other Tools
#![allow(clippy::cognitive_complexity)]
#![warn(rust_2018_idioms)]
#![cfg(feature = "sync")]
#[cfg(all(target_family = "wasm", not(target_os = "wasi")))]
use wasm_bindgen_test::wasm_bindgen_test as test;
use tokio::sync::broadcast;
use tokio_test::task;
use tokio_test::{
assert_err, assert_ok, assert_pending, assert_ready, assert_ready_err, assert_ready_ok,
};
use std::sync::Arc;
macro_rules! assert_recv {
($e:expr) => {
match $e.try_recv() {
Ok(value) => value,
Err(e) => panic!("expected recv; got = {:?}", e),
}
};
}
macro_rules! assert_empty {
($e:expr) => {
match $e.try_recv() {
Ok(value) => panic!("expected empty; got = {:?}", value),
Err(broadcast::error::TryRecvError::Empty) => {}
Err(e) => panic!("expected empty; got = {:?}", e),
}
};
}
macro_rules! assert_lagged {
($e:expr, $n:expr) => {
match assert_err!($e) {
broadcast::error::TryRecvError::Lagged(n) => {
assert_eq!(n, $n);
}
_ => panic!("did not lag"),
}
};
}
macro_rules! assert_closed {
($e:expr) => {
match assert_err!($e) {
broadcast::error::TryRecvError::Closed => {}
_ => panic!("is not closed"),
}
};
}
#[allow(unused)]
trait AssertSend: Send + Sync {}
impl AssertSend for broadcast::Sender<i32> {}
impl AssertSend for broadcast::Receiver<i32> {}
#[test]
fn send_try_recv_bounded() {
let (tx, mut rx) = broadcast::channel(16);
assert_empty!(rx);
let n = assert_ok!(tx.send("hello"));
assert_eq!(n, 1);
let val = assert_recv!(rx);
assert_eq!(val, "hello");
assert_empty!(rx);
}
#[test]
fn send_two_recv() {
let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();
assert_empty!(rx1);
assert_empty!(rx2);
let n = assert_ok!(tx.send("hello"));
assert_eq!(n, 2);
let val = assert_recv!(rx1);
assert_eq!(val, "hello");
let val = assert_recv!(rx2);
assert_eq!(val, "hello");
assert_empty!(rx1);
assert_empty!(rx2);
}
#[test]
fn send_recv_bounded() {
let (tx, mut rx) = broadcast::channel(16);
let mut recv = task::spawn(rx.recv());
assert_pending!(recv.poll());
assert_ok!(tx.send("hello"));
assert!(recv.is_woken());
let val = assert_ready_ok!(recv.poll());
assert_eq!(val, "hello");
}
#[test]
fn send_two_recv_bounded() {
let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();
let mut recv1 = task::spawn(rx1.recv());
let mut recv2 = task::spawn(rx2.recv());
assert_pending!(recv1.poll());
assert_pending!(recv2.poll());
assert_ok!(tx.send("hello"));
assert!(recv1.is_woken());
assert!(recv2.is_woken());
let val1 = assert_ready_ok!(recv1.poll());
let val2 = assert_ready_ok!(recv2.poll());
assert_eq!(val1, "hello");
assert_eq!(val2, "hello");
drop((recv1, recv2));
let mut recv1 = task::spawn(rx1.recv());
let mut recv2 = task::spawn(rx2.recv());
assert_pending!(recv1.poll());
assert_ok!(tx.send("world"));
assert!(recv1.is_woken());
assert!(!recv2.is_woken());
let val1 = assert_ready_ok!(recv1.poll());
let val2 = assert_ready_ok!(recv2.poll());
assert_eq!(val1, "world");
assert_eq!(val2, "world");
}
#[test]
fn change_tasks() {
let (tx, mut rx) = broadcast::channel(1);
let mut recv = Box::pin(rx.recv());
let mut task1 = task::spawn(&mut recv);
assert_pending!(task1.poll());
let mut task2 = task::spawn(&mut recv);
assert_pending!(task2.poll());
tx.send("hello").unwrap();
assert!(task2.is_woken());
}
#[test]
fn send_slow_rx() {
let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();
{
let mut recv2 = task::spawn(rx2.recv());
{
let mut recv1 = task::spawn(rx1.recv());
assert_pending!(recv1.poll());
assert_pending!(recv2.poll());
assert_ok!(tx.send("one"));
assert!(recv1.is_woken());
assert!(recv2.is_woken());
assert_ok!(tx.send("two"));
let val = assert_ready_ok!(recv1.poll());
assert_eq!(val, "one");
}
let val = assert_ready_ok!(task::spawn(rx1.recv()).poll());
assert_eq!(val, "two");
let mut recv1 = task::spawn(rx1.recv());
assert_pending!(recv1.poll());
assert_ok!(tx.send("three"));
assert!(recv1.is_woken());
let val = assert_ready_ok!(recv1.poll());
assert_eq!(val, "three");
let val = assert_ready_ok!(recv2.poll());
assert_eq!(val, "one");
}
let val = assert_recv!(rx2);
assert_eq!(val, "two");
let val = assert_recv!(rx2);
assert_eq!(val, "three");
}
#[test]
fn drop_rx_while_values_remain() {
let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();
assert_ok!(tx.send("one"));
assert_ok!(tx.send("two"));
assert_recv!(rx1);
assert_recv!(rx2);
drop(rx2);
drop(rx1);
}
#[test]
fn lagging_rx() {
let (tx, mut rx1) = broadcast::channel(2);
let mut rx2 = tx.subscribe();
assert_ok!(tx.send("one"));
assert_ok!(tx.send("two"));
assert_eq!("one", assert_recv!(rx1));
assert_ok!(tx.send("three"));
// Lagged too far
let x = dbg!(rx2.try_recv());
assert_lagged!(x, 1);
// Calling again gets the next value
assert_eq!("two", assert_recv!(rx2));
assert_eq!("two", assert_recv!(rx1));
assert_eq!("three", assert_recv!(rx1));
assert_ok!(tx.send("four"));
assert_ok!(tx.send("five"));
assert_lagged!(rx2.try_recv(), 1);
assert_ok!(tx.send("six"));
assert_lagged!(rx2.try_recv(), 1);
}
#[test]
fn send_no_rx() {
let (tx, _) = broadcast::channel(16);
assert_err!(tx.send("hello"));
let mut rx = tx.subscribe();
assert_ok!(tx.send("world"));
let val = assert_recv!(rx);
assert_eq!("world", val);
}
#[test]
#[should_panic]
#[cfg(not(target_family = "wasm"))] // wasm currently doesn't support unwinding
fn zero_capacity() {
broadcast::channel::<()>(0);
}
#[test]
#[should_panic]
#[cfg(not(target_family = "wasm"))] // wasm currently doesn't support unwinding
fn capacity_too_big() {
broadcast::channel::<()>(1 + (usize::MAX >> 1));
}
#[test]
#[cfg(panic = "unwind")]
#[cfg(not(target_family = "wasm"))] // wasm currently doesn't support unwinding
fn panic_in_clone() {
use std::panic::{self, AssertUnwindSafe};
#[derive(Eq, PartialEq, Debug)]
struct MyVal(usize);
impl Clone for MyVal {
fn clone(&self) -> MyVal {
assert_ne!(0, self.0);
MyVal(self.0)
}
}
let (tx, mut rx) = broadcast::channel(16);
assert_ok!(tx.send(MyVal(0)));
assert_ok!(tx.send(MyVal(1)));
let res = panic::catch_unwind(AssertUnwindSafe(|| {
let _ = rx.try_recv();
}));
assert_err!(res);
let val = assert_recv!(rx);
assert_eq!(val, MyVal(1));
}
#[test]
fn dropping_tx_notifies_rx() {
let (tx, mut rx1) = broadcast::channel::<()>(16);
let mut rx2 = tx.subscribe();
let tx2 = tx.clone();
let mut recv1 = task::spawn(rx1.recv());
let mut recv2 = task::spawn(rx2.recv());
assert_pending!(recv1.poll());
assert_pending!(recv2.poll());
drop(tx);
assert_pending!(recv1.poll());
assert_pending!(recv2.poll());
drop(tx2);
assert!(recv1.is_woken());
assert!(recv2.is_woken());
let err = assert_ready_err!(recv1.poll());
assert!(is_closed(err));
let err = assert_ready_err!(recv2.poll());
assert!(is_closed(err));
}
#[test]
fn unconsumed_messages_are_dropped() {
let (tx, rx) = broadcast::channel(16);
let msg = Arc::new(());
assert_ok!(tx.send(msg.clone()));
assert_eq!(2, Arc::strong_count(&msg));
drop(rx);
assert_eq!(1, Arc::strong_count(&msg));
}
#[test]
fn single_capacity_recvs() {
let (tx, mut rx) = broadcast::channel(1);
assert_ok!(tx.send(1));
assert_eq!(assert_recv!(rx), 1);
assert_empty!(rx);
}
#[test]
fn single_capacity_recvs_after_drop_1() {
let (tx, mut rx) = broadcast::channel(1);
assert_ok!(tx.send(1));
drop(tx);
assert_eq!(assert_recv!(rx), 1);
assert_closed!(rx.try_recv());
}
#[test]
fn single_capacity_recvs_after_drop_2() {
let (tx, mut rx) = broadcast::channel(1);
assert_ok!(tx.send(1));
assert_ok!(tx.send(2));
drop(tx);
assert_lagged!(rx.try_recv(), 1);
assert_eq!(assert_recv!(rx), 2);
assert_closed!(rx.try_recv());
}
#[test]
fn dropping_sender_does_not_overwrite() {
let (tx, mut rx) = broadcast::channel(2);
assert_ok!(tx.send(1));
assert_ok!(tx.send(2));
drop(tx);
assert_eq!(assert_recv!(rx), 1);
assert_eq!(assert_recv!(rx), 2);
assert_closed!(rx.try_recv());
}
#[test]
fn lagging_receiver_recovers_after_wrap_closed_1() {
let (tx, mut rx) = broadcast::channel(2);
assert_ok!(tx.send(1));
assert_ok!(tx.send(2));
assert_ok!(tx.send(3));
drop(tx);
assert_lagged!(rx.try_recv(), 1);
assert_eq!(assert_recv!(rx), 2);
assert_eq!(assert_recv!(rx), 3);
assert_closed!(rx.try_recv());
}
#[test]
fn lagging_receiver_recovers_after_wrap_closed_2() {
let (tx, mut rx) = broadcast::channel(2);
assert_ok!(tx.send(1));
assert_ok!(tx.send(2));
assert_ok!(tx.send(3));
assert_ok!(tx.send(4));
drop(tx);
assert_lagged!(rx.try_recv(), 2);
assert_eq!(assert_recv!(rx), 3);
assert_eq!(assert_recv!(rx), 4);
assert_closed!(rx.try_recv());
}
#[test]
fn lagging_receiver_recovers_after_wrap_open() {
let (tx, mut rx) = broadcast::channel(2);
assert_ok!(tx.send(1));
assert_ok!(tx.send(2));
assert_ok!(tx.send(3));
assert_lagged!(rx.try_recv(), 1);
assert_eq!(assert_recv!(rx), 2);
assert_eq!(assert_recv!(rx), 3);
assert_empty!(rx);
}
#[test]
fn receiver_len_with_lagged() {
let (tx, mut rx) = broadcast::channel(3);
tx.send(10).unwrap();
tx.send(20).unwrap();
tx.send(30).unwrap();
tx.send(40).unwrap();
assert_eq!(rx.len(), 4);
assert_eq!(assert_recv!(rx), 10);
tx.send(50).unwrap();
tx.send(60).unwrap();
assert_eq!(rx.len(), 5);
assert_lagged!(rx.try_recv(), 1);
}
fn is_closed(err: broadcast::error::RecvError) -> bool {
matches!(err, broadcast::error::RecvError::Closed)
}
#[test]
fn resubscribe_points_to_tail() {
let (tx, mut rx) = broadcast::channel(3);
tx.send(1).unwrap();
let mut rx_resub = rx.resubscribe();
// verify we're one behind at the start
assert_empty!(rx_resub);
assert_eq!(assert_recv!(rx), 1);
// verify we do not affect rx
tx.send(2).unwrap();
assert_eq!(assert_recv!(rx_resub), 2);
tx.send(3).unwrap();
assert_eq!(assert_recv!(rx), 2);
assert_eq!(assert_recv!(rx), 3);
assert_empty!(rx);
assert_eq!(assert_recv!(rx_resub), 3);
assert_empty!(rx_resub);
}
#[test]
fn resubscribe_lagged() {
let (tx, mut rx) = broadcast::channel(1);
tx.send(1).unwrap();
tx.send(2).unwrap();
let mut rx_resub = rx.resubscribe();
assert_lagged!(rx.try_recv(), 1);
assert_empty!(rx_resub);
assert_eq!(assert_recv!(rx), 2);
assert_empty!(rx);
assert_empty!(rx_resub);
}
#[test]
fn resubscribe_to_closed_channel() {
let (tx, rx) = tokio::sync::broadcast::channel::<u32>(2);
drop(tx);
let mut rx_resub = rx.resubscribe();
assert_closed!(rx_resub.try_recv());
}
#[test]
fn sender_len() {
let (tx, mut rx1) = broadcast::channel(4);
let mut rx2 = tx.subscribe();
assert_eq!(tx.len(), 0);
assert!(tx.is_empty());
tx.send(1).unwrap();
tx.send(2).unwrap();
tx.send(3).unwrap();
assert_eq!(tx.len(), 3);
assert!(!tx.is_empty());
assert_recv!(rx1);
assert_recv!(rx1);
assert_eq!(tx.len(), 3);
assert!(!tx.is_empty());
assert_recv!(rx2);
assert_eq!(tx.len(), 2);
assert!(!tx.is_empty());
tx.send(4).unwrap();
tx.send(5).unwrap();
tx.send(6).unwrap();
assert_eq!(tx.len(), 4);
assert!(!tx.is_empty());
}
#[test]
#[cfg(not(all(target_family = "wasm", not(target_os = "wasi"))))]
fn sender_len_random() {
use rand::Rng;
let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();
for _ in 0..1000 {
match rand::thread_rng().gen_range(0..4) {
0 => {
let _ = rx1.try_recv();
}
1 => {
let _ = rx2.try_recv();
}
_ => {
tx.send(0).unwrap();
}
}
let expected_len = usize::min(usize::max(rx1.len(), rx2.len()), 16);
assert_eq!(tx.len(), expected_len);
}
}
#[test]
fn send_in_waker_drop() {
use futures::task::ArcWake;
use std::future::Future;
use std::task::Context;
struct SendOnDrop(broadcast::Sender<()>);
impl Drop for SendOnDrop {
fn drop(&mut self) {
let _ = self.0.send(());
}
}
impl ArcWake for SendOnDrop {
fn wake_by_ref(_arc_self: &Arc<Self>) {}
}
// Test if there is no deadlock when replacing the old waker.
let (tx, mut rx) = broadcast::channel(16);
let mut fut = Box::pin(async {
let _ = rx.recv().await;
});
// Store our special waker in the receiving future.
let waker = futures::task::waker(Arc::new(SendOnDrop(tx)));
let mut cx = Context::from_waker(&waker);
assert!(fut.as_mut().poll(&mut cx).is_pending());
drop(waker);
// Second poll shouldn't deadlock.
let mut cx = Context::from_waker(futures::task::noop_waker_ref());
let _ = fut.as_mut().poll(&mut cx);
// Test if there is no deadlock when calling waker.wake().
let (tx, mut rx) = broadcast::channel(16);
let mut fut = Box::pin(async {
let _ = rx.recv().await;
});
// Store our special waker in the receiving future.
let waker = futures::task::waker(Arc::new(SendOnDrop(tx.clone())));
let mut cx = Context::from_waker(&waker);
assert!(fut.as_mut().poll(&mut cx).is_pending());
drop(waker);
// Shouldn't deadlock.
let _ = tx.send(());
}