#![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::watch; use tokio_test::task::spawn; use tokio_test::{ assert_pending, assert_ready, assert_ready_eq, assert_ready_err, assert_ready_ok, }; #[test] fn single_rx_recv() { let (tx, mut rx) = watch::channel("one"); { // Not initially notified let mut t = spawn(rx.changed()); assert_pending!(t.poll()); } assert_eq!(*rx.borrow(), "one"); { let mut t = spawn(rx.changed()); assert_pending!(t.poll()); tx.send("two").unwrap(); assert!(t.is_woken()); assert_ready_ok!(t.poll()); } assert_eq!(*rx.borrow(), "two"); { let mut t = spawn(rx.changed()); assert_pending!(t.poll()); drop(tx); assert!(t.is_woken()); assert_ready_err!(t.poll()); } assert_eq!(*rx.borrow(), "two"); } #[test] fn rx_version_underflow() { let (_tx, mut rx) = watch::channel("one"); // Version starts at 2, validate we do not underflow rx.mark_changed(); rx.mark_changed(); } #[test] fn rx_mark_changed() { let (tx, mut rx) = watch::channel("one"); let mut rx2 = rx.clone(); let mut rx3 = rx.clone(); let mut rx4 = rx.clone(); { rx.mark_changed(); assert!(rx.has_changed().unwrap()); let mut t = spawn(rx.changed()); assert_ready_ok!(t.poll()); } { assert!(!rx2.has_changed().unwrap()); let mut t = spawn(rx2.changed()); assert_pending!(t.poll()); } { rx3.mark_changed(); assert_eq!(*rx3.borrow(), "one"); assert!(rx3.has_changed().unwrap()); assert_eq!(*rx3.borrow_and_update(), "one"); assert!(!rx3.has_changed().unwrap()); let mut t = spawn(rx3.changed()); assert_pending!(t.poll()); } { tx.send("two").unwrap(); assert!(rx4.has_changed().unwrap()); assert_eq!(*rx4.borrow_and_update(), "two"); rx4.mark_changed(); assert!(rx4.has_changed().unwrap()); assert_eq!(*rx4.borrow_and_update(), "two") } assert_eq!(*rx.borrow(), "two"); } #[test] fn rx_mark_unchanged() { let (tx, mut rx) = watch::channel("one"); let mut rx2 = rx.clone(); { assert!(!rx.has_changed().unwrap()); rx.mark_changed(); assert!(rx.has_changed().unwrap()); rx.mark_unchanged(); assert!(!rx.has_changed().unwrap()); let mut t = spawn(rx.changed()); assert_pending!(t.poll()); } { assert!(!rx2.has_changed().unwrap()); tx.send("two").unwrap(); assert!(rx2.has_changed().unwrap()); rx2.mark_unchanged(); assert!(!rx2.has_changed().unwrap()); assert_eq!(*rx2.borrow_and_update(), "two"); } assert_eq!(*rx.borrow(), "two"); } #[test] fn multi_rx() { let (tx, mut rx1) = watch::channel("one"); let mut rx2 = rx1.clone(); { let mut t1 = spawn(rx1.changed()); let mut t2 = spawn(rx2.changed()); assert_pending!(t1.poll()); assert_pending!(t2.poll()); } assert_eq!(*rx1.borrow(), "one"); assert_eq!(*rx2.borrow(), "one"); let mut t2 = spawn(rx2.changed()); { let mut t1 = spawn(rx1.changed()); assert_pending!(t1.poll()); assert_pending!(t2.poll()); tx.send("two").unwrap(); assert!(t1.is_woken()); assert!(t2.is_woken()); assert_ready_ok!(t1.poll()); } assert_eq!(*rx1.borrow(), "two"); { let mut t1 = spawn(rx1.changed()); assert_pending!(t1.poll()); tx.send("three").unwrap(); assert!(t1.is_woken()); assert!(t2.is_woken()); assert_ready_ok!(t1.poll()); assert_ready_ok!(t2.poll()); } assert_eq!(*rx1.borrow(), "three"); drop(t2); assert_eq!(*rx2.borrow(), "three"); { let mut t1 = spawn(rx1.changed()); let mut t2 = spawn(rx2.changed()); assert_pending!(t1.poll()); assert_pending!(t2.poll()); tx.send("four").unwrap(); assert_ready_ok!(t1.poll()); assert_ready_ok!(t2.poll()); } assert_eq!(*rx1.borrow(), "four"); assert_eq!(*rx2.borrow(), "four"); } #[test] fn rx_observes_final_value() { // Initial value let (tx, mut rx) = watch::channel("one"); drop(tx); { let mut t1 = spawn(rx.changed()); assert_ready_err!(t1.poll()); } assert_eq!(*rx.borrow(), "one"); // Sending a value let (tx, mut rx) = watch::channel("one"); tx.send("two").unwrap(); { let mut t1 = spawn(rx.changed()); assert_ready_ok!(t1.poll()); } assert_eq!(*rx.borrow(), "two"); { let mut t1 = spawn(rx.changed()); assert_pending!(t1.poll()); tx.send("three").unwrap(); drop(tx); assert!(t1.is_woken()); assert_ready_ok!(t1.poll()); } assert_eq!(*rx.borrow(), "three"); { let mut t1 = spawn(rx.changed()); assert_ready_err!(t1.poll()); } assert_eq!(*rx.borrow(), "three"); } #[test] fn poll_close() { let (tx, rx) = watch::channel("one"); { let mut t = spawn(tx.closed()); assert_pending!(t.poll()); drop(rx); assert!(t.is_woken()); assert_ready!(t.poll()); } assert!(tx.send("two").is_err()); } #[test] fn borrow_and_update() { let (tx, mut rx) = watch::channel("one"); assert!(!rx.has_changed().unwrap()); tx.send("two").unwrap(); assert!(rx.has_changed().unwrap()); assert_ready!(spawn(rx.changed()).poll()).unwrap(); assert_pending!(spawn(rx.changed()).poll()); assert!(!rx.has_changed().unwrap()); tx.send("three").unwrap(); assert!(rx.has_changed().unwrap()); assert_eq!(*rx.borrow_and_update(), "three"); assert_pending!(spawn(rx.changed()).poll()); assert!(!rx.has_changed().unwrap()); drop(tx); assert_eq!(*rx.borrow_and_update(), "three"); assert_ready!(spawn(rx.changed()).poll()).unwrap_err(); assert!(rx.has_changed().is_err()); } #[test] fn reopened_after_subscribe() { let (tx, rx) = watch::channel("one"); assert!(!tx.is_closed()); drop(rx); assert!(tx.is_closed()); let rx = tx.subscribe(); assert!(!tx.is_closed()); drop(rx); assert!(tx.is_closed()); } #[test] #[cfg(panic = "unwind")] #[cfg(not(target_family = "wasm"))] // wasm currently doesn't support unwinding fn send_modify_panic() { let (tx, mut rx) = watch::channel("one"); tx.send_modify(|old| *old = "two"); assert_eq!(*rx.borrow_and_update(), "two"); let mut rx2 = rx.clone(); assert_eq!(*rx2.borrow_and_update(), "two"); let mut task = spawn(rx2.changed()); let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { tx.send_modify(|old| { *old = "panicked"; panic!(); }) })); assert!(result.is_err()); assert_pending!(task.poll()); assert_eq!(*rx.borrow(), "panicked"); tx.send_modify(|old| *old = "three"); assert_ready_ok!(task.poll()); assert_eq!(*rx.borrow_and_update(), "three"); } #[tokio::test] async fn multiple_sender() { let (tx1, mut rx) = watch::channel(0); let tx2 = tx1.clone(); let mut t = spawn(async { rx.changed().await.unwrap(); let v1 = *rx.borrow_and_update(); rx.changed().await.unwrap(); let v2 = *rx.borrow_and_update(); (v1, v2) }); tx1.send(1).unwrap(); assert_pending!(t.poll()); tx2.send(2).unwrap(); assert_ready_eq!(t.poll(), (1, 2)); } #[tokio::test] async fn reciever_is_notified_when_last_sender_is_dropped() { let (tx1, mut rx) = watch::channel(0); let tx2 = tx1.clone(); let mut t = spawn(rx.changed()); assert_pending!(t.poll()); drop(tx1); assert!(!t.is_woken()); drop(tx2); assert!(t.is_woken()); }