298 lines
7.1 KiB
Rust
298 lines
7.1 KiB
Rust
use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet, LinkedList, VecDeque};
|
|
|
|
use tokio_stream::{self as stream, StreamExt};
|
|
use tokio_test::{assert_pending, assert_ready, assert_ready_err, assert_ready_ok, task};
|
|
|
|
mod support {
|
|
pub(crate) mod mpsc;
|
|
}
|
|
|
|
use support::mpsc;
|
|
|
|
#[allow(clippy::let_unit_value)]
|
|
#[tokio::test]
|
|
async fn empty_unit() {
|
|
// Drains the stream.
|
|
let mut iter = vec![(), (), ()].into_iter();
|
|
let _: () = stream::iter(&mut iter).collect().await;
|
|
assert!(iter.next().is_none());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn empty_vec() {
|
|
let coll: Vec<u32> = stream::empty().collect().await;
|
|
assert!(coll.is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn empty_box_slice() {
|
|
let coll: Box<[u32]> = stream::empty().collect().await;
|
|
assert!(coll.is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn empty_string() {
|
|
let coll: String = stream::empty::<&str>().collect().await;
|
|
assert!(coll.is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn empty_result() {
|
|
let coll: Result<Vec<u32>, &str> = stream::empty().collect().await;
|
|
assert_eq!(Ok(vec![]), coll);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_vec_items() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
let mut fut = task::spawn(rx.collect::<Vec<i32>>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(1).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(2).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready!(fut.poll());
|
|
assert_eq!(vec![1, 2], coll);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_vecdeque_items() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
let mut fut = task::spawn(rx.collect::<VecDeque<i32>>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
let xs = [1, 2, 42, 3];
|
|
for x in xs {
|
|
tx.send(x).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
}
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready!(fut.poll());
|
|
|
|
assert_eq!(coll, VecDeque::from(xs));
|
|
assert_eq!(coll.into_iter().collect::<Vec<_>>(), xs);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_linkedlist_items() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
let mut fut = task::spawn(rx.collect::<LinkedList<i32>>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
let xs = [1, 2, 42, 3];
|
|
for x in xs {
|
|
tx.send(x).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
}
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready!(fut.poll());
|
|
|
|
assert_eq!(coll, LinkedList::from(xs));
|
|
assert_eq!(coll.into_iter().collect::<Vec<_>>(), xs);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_btreeset_items() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
let mut fut = task::spawn(rx.collect::<BTreeSet<i32>>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(2).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(1).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready!(fut.poll());
|
|
assert_eq!(BTreeSet::from([1, 2]), coll);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_btreemap_items() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
let mut fut = task::spawn(rx.collect::<BTreeMap<i32, i32>>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send((3, 4)).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send((1, 2)).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready!(fut.poll());
|
|
assert_eq!(BTreeMap::from([(1, 2), (3, 4)]), coll);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_hashset_items() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
let mut fut = task::spawn(rx.collect::<HashSet<i32>>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(1).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(2).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready!(fut.poll());
|
|
assert_eq!(HashSet::from([1, 2]), coll);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_hashmap_items() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
let mut fut = task::spawn(rx.collect::<HashMap<i32, i32>>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send((1, 2)).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send((3, 4)).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready!(fut.poll());
|
|
assert_eq!(HashMap::from([(1, 2), (3, 4)]), coll);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_binaryheap_items() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
let mut fut = task::spawn(rx.collect::<BinaryHeap<i32>>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(2).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(1).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready!(fut.poll());
|
|
assert_eq!(vec![1, 2], coll.into_sorted_vec());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_string_items() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
|
|
let mut fut = task::spawn(rx.collect::<String>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send("hello ".to_string()).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send("world".to_string()).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready!(fut.poll());
|
|
assert_eq!("hello world", coll);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_str_items() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
|
|
let mut fut = task::spawn(rx.collect::<String>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send("hello ").unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send("world").unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready!(fut.poll());
|
|
assert_eq!("hello world", coll);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_results_ok() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
|
|
let mut fut = task::spawn(rx.collect::<Result<String, &str>>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(Ok("hello ")).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(Ok("world")).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
drop(tx);
|
|
assert!(fut.is_woken());
|
|
let coll = assert_ready_ok!(fut.poll());
|
|
assert_eq!("hello world", coll);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn collect_results_err() {
|
|
let (tx, rx) = mpsc::unbounded_channel_stream();
|
|
|
|
let mut fut = task::spawn(rx.collect::<Result<String, &str>>());
|
|
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(Ok("hello ")).unwrap();
|
|
assert!(fut.is_woken());
|
|
assert_pending!(fut.poll());
|
|
|
|
tx.send(Err("oh no")).unwrap();
|
|
assert!(fut.is_woken());
|
|
let err = assert_ready_err!(fut.poll());
|
|
assert_eq!("oh no", err);
|
|
}
|