stream: add `StreamExt::merge` (#2091)

Provides an equivalent to stream `select()` from futures-rs. `merge`
best describes the operation (vs. `select`). `futures-rs` named the
operation "select" for historical reasons and did not rename it back to
`merge` in 0.3. The operation is most commonly named `merge` else where
as well (e.g. ReactiveX).

1 files changed, 54 insertions, 0 deletions

diff --git a/tokio/tests/stream_merge.rs b/tokio/tests/stream_merge.rs
new file mode 100644
index 00000000..f0168d72
--- /dev/null
+++ b/tokio/tests/stream_merge.rs
@@ -0,0 +1,54 @@
+use tokio::stream::{self, Stream, StreamExt};
+use tokio::sync::mpsc;
+use tokio_test::task;
+use tokio_test::{assert_pending, assert_ready};
+
+#[tokio::test]
+async fn merge_sync_streams() {
+    let mut s = stream::iter(vec![0, 2, 4, 6]).merge(stream::iter(vec![1, 3, 5]));
+
+    for i in 0..7 {
+        let rem = 7 - i;
+        assert_eq!(s.size_hint(), (rem, Some(rem)));
+        assert_eq!(Some(i), s.next().await);
+    }
+
+    assert!(s.next().await.is_none());
+}
+
+#[tokio::test]
+async fn merge_async_streams() {
+    let (tx1, rx1) = mpsc::unbounded_channel();
+    let (tx2, rx2) = mpsc::unbounded_channel();
+
+    let mut rx = task::spawn(rx1.merge(rx2));
+
+    assert_eq!(rx.size_hint(), (0, None));
+
+    assert_pending!(rx.poll_next());
+
+    tx1.send(1).unwrap();
+
+    assert!(rx.is_woken());
+    assert_eq!(Some(1), assert_ready!(rx.poll_next()));
+
+    assert_pending!(rx.poll_next());
+    tx2.send(2).unwrap();
+
+    assert!(rx.is_woken());
+    assert_eq!(Some(2), assert_ready!(rx.poll_next()));
+    assert_pending!(rx.poll_next());
+
+    drop(tx1);
+    assert!(rx.is_woken());
+    assert_pending!(rx.poll_next());
+
+    tx2.send(3).unwrap();
+    assert!(rx.is_woken());
+    assert_eq!(Some(3), assert_ready!(rx.poll_next()));
+    assert_pending!(rx.poll_next());
+
+    drop(tx2);
+    assert!(rx.is_woken());
+    assert_eq!(None, assert_ready!(rx.poll_next()));
+}