1use futures::channel::oneshot;
2use futures::executor::{block_on, block_on_stream};
3use futures::future;
4use futures::stream::{FuturesUnordered, StreamExt};
5use futures::task::Poll;
6use futures_test::task::noop_context;
7use std::panic::{self, AssertUnwindSafe};
8use std::sync::{Arc, Barrier};
9use std::thread;
10
11#[test]
12fn basic_usage() {
13 block_on(future::lazy(move |cx| {
14 let mut queue = FuturesUnordered::new();
15 let (tx1, rx1) = oneshot::channel();
16 let (tx2, rx2) = oneshot::channel();
17 let (tx3, rx3) = oneshot::channel();
18
19 queue.push(rx1);
20 queue.push(rx2);
21 queue.push(rx3);
22
23 assert!(!queue.poll_next_unpin(cx).is_ready());
24
25 tx2.send("hello").unwrap();
26
27 assert_eq!(Poll::Ready(Some(Ok("hello"))), queue.poll_next_unpin(cx));
28 assert!(!queue.poll_next_unpin(cx).is_ready());
29
30 tx1.send("world").unwrap();
31 tx3.send("world2").unwrap();
32
33 assert_eq!(Poll::Ready(Some(Ok("world"))), queue.poll_next_unpin(cx));
34 assert_eq!(Poll::Ready(Some(Ok("world2"))), queue.poll_next_unpin(cx));
35 assert_eq!(Poll::Ready(None), queue.poll_next_unpin(cx));
36 }));
37}
38
39#[test]
40fn resolving_errors() {
41 block_on(future::lazy(move |cx| {
42 let mut queue = FuturesUnordered::new();
43 let (tx1, rx1) = oneshot::channel();
44 let (tx2, rx2) = oneshot::channel();
45 let (tx3, rx3) = oneshot::channel();
46
47 queue.push(rx1);
48 queue.push(rx2);
49 queue.push(rx3);
50
51 assert!(!queue.poll_next_unpin(cx).is_ready());
52
53 drop(tx2);
54
55 assert_eq!(Poll::Ready(Some(Err(oneshot::Canceled))), queue.poll_next_unpin(cx));
56 assert!(!queue.poll_next_unpin(cx).is_ready());
57
58 drop(tx1);
59 tx3.send("world2").unwrap();
60
61 assert_eq!(Poll::Ready(Some(Err(oneshot::Canceled))), queue.poll_next_unpin(cx));
62 assert_eq!(Poll::Ready(Some(Ok("world2"))), queue.poll_next_unpin(cx));
63 assert_eq!(Poll::Ready(None), queue.poll_next_unpin(cx));
64 }));
65}
66
67#[test]
68fn dropping_ready_queue() {
69 block_on(future::lazy(move |_| {
70 let queue = FuturesUnordered::new();
71 let (mut tx1, rx1) = oneshot::channel::<()>();
72 let (mut tx2, rx2) = oneshot::channel::<()>();
73 let (mut tx3, rx3) = oneshot::channel::<()>();
74
75 queue.push(rx1);
76 queue.push(rx2);
77 queue.push(rx3);
78
79 {
80 let cx = &mut noop_context();
81 assert!(!tx1.poll_canceled(cx).is_ready());
82 assert!(!tx2.poll_canceled(cx).is_ready());
83 assert!(!tx3.poll_canceled(cx).is_ready());
84
85 drop(queue);
86
87 assert!(tx1.poll_canceled(cx).is_ready());
88 assert!(tx2.poll_canceled(cx).is_ready());
89 assert!(tx3.poll_canceled(cx).is_ready());
90 }
91 }));
92}
93
94#[test]
95fn stress() {
96 const ITER: usize = if cfg!(miri) { 30 } else { 300 };
97
98 for i in 0..ITER {
99 let n = (i % 10) + 1;
100
101 let mut queue = FuturesUnordered::new();
102
103 for _ in 0..5 {
104 let barrier = Arc::new(Barrier::new(n + 1));
105
106 for num in 0..n {
107 let barrier = barrier.clone();
108 let (tx, rx) = oneshot::channel();
109
110 queue.push(rx);
111
112 thread::spawn(move || {
113 barrier.wait();
114 tx.send(num).unwrap();
115 });
116 }
117
118 barrier.wait();
119
120 let mut sync = block_on_stream(queue);
121
122 let mut rx: Vec<_> = (&mut sync).take(n).map(|res| res.unwrap()).collect();
123
124 assert_eq!(rx.len(), n);
125
126 rx.sort_unstable();
127
128 for (i, x) in rx.into_iter().enumerate() {
129 assert_eq!(i, x);
130 }
131
132 queue = sync.into_inner();
133 }
134 }
135}
136
137#[test]
138fn panicking_future_dropped() {
139 block_on(future::lazy(move |cx| {
140 let mut queue = FuturesUnordered::new();
141 queue.push(future::poll_fn(|_| -> Poll<Result<i32, i32>> { panic!() }));
142
143 let r = panic::catch_unwind(AssertUnwindSafe(|| queue.poll_next_unpin(cx)));
144 assert!(r.is_err());
145 assert!(queue.is_empty());
146 assert_eq!(Poll::Ready(None), queue.poll_next_unpin(cx));
147 }));
148}
149