From efac9598fc350f746f798b54c05a19ec30780cf2 Mon Sep 17 00:00:00 2001
From: Ibraheem Ahmed <ibrah1440@gmail.com>
Date: Mon, 10 May 2021 10:43:50 -0400
Subject: [PATCH] Add SelectAll::clear (#2430)

---
 futures-util/src/stream/select_all.rs |  5 +++++
 futures/tests/stream_select_all.rs    | 25 ++++++++++++++++++++++++-
 2 files changed, 29 insertions(+), 1 deletion(-)

diff --git a/futures-util/src/stream/select_all.rs b/futures-util/src/stream/select_all.rs
index e9ab519e5e..145099cb29 100644
--- a/futures-util/src/stream/select_all.rs
+++ b/futures-util/src/stream/select_all.rs
@@ -90,6 +90,11 @@ impl<St: Stream + Unpin> SelectAll<St> {
     pub fn iter_pin_mut(self: Pin<&mut Self>) -> IterPinMut<'_, StreamFuture<St>> {
         self.project().inner.iter_pin_mut()
     }
+
+    /// Clears the set, removing all futures.
+    pub fn clear(&mut self) {
+        self.inner.clear()
+    }
 }
 
 impl<St: Stream + Unpin> Default for SelectAll<St> {
diff --git a/futures/tests/stream_select_all.rs b/futures/tests/stream_select_all.rs
index eb711dda0c..e7e3976398 100644
--- a/futures/tests/stream_select_all.rs
+++ b/futures/tests/stream_select_all.rs
@@ -1,5 +1,5 @@
 use futures::channel::mpsc;
-use futures::executor::block_on_stream;
+use futures::executor::{block_on, block_on_stream};
 use futures::future::{self, FutureExt};
 use futures::stream::{self, select_all, FusedStream, SelectAll, StreamExt};
 use futures::task::Poll;
@@ -76,3 +76,26 @@ fn works_1() {
     drop((a_tx, b_tx, c_tx));
     assert_eq!(None, stream.next());
 }
+
+#[test]
+fn clear() {
+    let mut tasks =
+        select_all(vec![stream::iter(vec![1].into_iter()), stream::iter(vec![2].into_iter())]);
+
+    assert_eq!(block_on(tasks.next()), Some(1));
+    assert!(!tasks.is_empty());
+
+    tasks.clear();
+    assert!(tasks.is_empty());
+
+    tasks.push(stream::iter(vec![3].into_iter()));
+    assert!(!tasks.is_empty());
+
+    tasks.clear();
+    assert!(tasks.is_empty());
+
+    assert_eq!(block_on(tasks.next()), None);
+    assert!(tasks.is_terminated());
+    tasks.clear();
+    assert!(!tasks.is_terminated());
+}