summaryrefslogtreecommitdiffstats
path: root/vendor/futures-channel/tests/channel.rs
diff options
context:
space:
mode:
Diffstat (limited to 'vendor/futures-channel/tests/channel.rs')
-rw-r--r--vendor/futures-channel/tests/channel.rs66
1 files changed, 66 insertions, 0 deletions
diff --git a/vendor/futures-channel/tests/channel.rs b/vendor/futures-channel/tests/channel.rs
new file mode 100644
index 000000000..5f01a8ef4
--- /dev/null
+++ b/vendor/futures-channel/tests/channel.rs
@@ -0,0 +1,66 @@
+use futures::channel::mpsc;
+use futures::executor::block_on;
+use futures::future::poll_fn;
+use futures::sink::SinkExt;
+use futures::stream::StreamExt;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::thread;
+
+#[test]
+fn sequence() {
+ let (tx, rx) = mpsc::channel(1);
+
+ let amt = 20;
+ let t = thread::spawn(move || block_on(send_sequence(amt, tx)));
+ let list: Vec<_> = block_on(rx.collect());
+ let mut list = list.into_iter();
+ for i in (1..=amt).rev() {
+ assert_eq!(list.next(), Some(i));
+ }
+ assert_eq!(list.next(), None);
+
+ t.join().unwrap();
+}
+
+async fn send_sequence(n: u32, mut sender: mpsc::Sender<u32>) {
+ for x in 0..n {
+ sender.send(n - x).await.unwrap();
+ }
+}
+
+#[test]
+fn drop_sender() {
+ let (tx, mut rx) = mpsc::channel::<u32>(1);
+ drop(tx);
+ let f = poll_fn(|cx| rx.poll_next_unpin(cx));
+ assert_eq!(block_on(f), None)
+}
+
+#[test]
+fn drop_rx() {
+ let (mut tx, rx) = mpsc::channel::<u32>(1);
+ block_on(tx.send(1)).unwrap();
+ drop(rx);
+ assert!(block_on(tx.send(1)).is_err());
+}
+
+#[test]
+fn drop_order() {
+ static DROPS: AtomicUsize = AtomicUsize::new(0);
+ let (mut tx, rx) = mpsc::channel(1);
+
+ struct A;
+
+ impl Drop for A {
+ fn drop(&mut self) {
+ DROPS.fetch_add(1, Ordering::SeqCst);
+ }
+ }
+
+ block_on(tx.send(A)).unwrap();
+ assert_eq!(DROPS.load(Ordering::SeqCst), 0);
+ drop(rx);
+ assert_eq!(DROPS.load(Ordering::SeqCst), 1);
+ assert!(block_on(tx.send(A)).is_err());
+ assert_eq!(DROPS.load(Ordering::SeqCst), 2);
+}