Minor improvements to All and Any adapters

futures-util/src/stream/stream/all.rs

@@ -13,7 +13,7 @@ pin_project! {
         #[pin]
         stream: St,
         f: F,
-        accum: Option<bool>,
+        done: bool,
         #[pin]
         future: Option<Fut>,
     }
@@ -27,7 +27,7 @@ where
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("All")
             .field("stream", &self.stream)
-            .field("accum", &self.accum)
+            .field("done", &self.done)
             .field("future", &self.future)
             .finish()
     }
@@ -40,7 +40,7 @@ where
     Fut: Future<Output = bool>,
 {
     pub(super) fn new(stream: St, f: F) -> Self {
-        Self { stream, f, accum: Some(true), future: None }
+        Self { stream, f, done: false, future: None }
     }
 }
 
@@ -51,7 +51,7 @@ where
     Fut: Future<Output = bool>,
 {
     fn is_terminated(&self) -> bool {
-        self.accum.is_none() && self.future.is_none()
+        self.done && self.future.is_none()
     }
 }
 
@@ -67,21 +67,22 @@ where
         let mut this = self.project();
         Poll::Ready(loop {
             if let Some(fut) = this.future.as_mut().as_pin_mut() {
-                // we're currently processing a future to produce a new accum value
-                let acc = this.accum.unwrap() && ready!(fut.poll(cx));
-                if !acc {
+                // we're currently processing a future to produce a new value
+                let res = ready!(fut.poll(cx));
+                this.future.set(None);
+                if !res {
+                    *this.done = true;
                     break false;
                 } // early exit
-                *this.accum = Some(acc);
-                this.future.set(None);
-            } else if this.accum.is_some() {
+            } else if !*this.done {
                 // we're waiting on a new item from the stream
                 match ready!(this.stream.as_mut().poll_next(cx)) {
                     Some(item) => {
                         this.future.set(Some((this.f)(item)));
                     }
                     None => {
-                        break this.accum.take().unwrap();
+                        *this.done = true;
+                        break true;
                     }
                 }
             } else {

futures-util/src/stream/stream/any.rs

@@ -13,7 +13,7 @@ pin_project! {
         #[pin]
         stream: St,
         f: F,
-        accum: Option<bool>,
+        done: bool,
         #[pin]
         future: Option<Fut>,
     }
@@ -27,7 +27,7 @@ where
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Any")
             .field("stream", &self.stream)
-            .field("accum", &self.accum)
+            .field("done", &self.done)
             .field("future", &self.future)
             .finish()
     }
@@ -40,7 +40,7 @@ where
     Fut: Future<Output = bool>,
 {
     pub(super) fn new(stream: St, f: F) -> Self {
-        Self { stream, f, accum: Some(false), future: None }
+        Self { stream, f, done: false, future: None }
     }
 }
 
@@ -51,7 +51,7 @@ where
     Fut: Future<Output = bool>,
 {
     fn is_terminated(&self) -> bool {
-        self.accum.is_none() && self.future.is_none()
+        self.done && self.future.is_none()
     }
 }
 
@@ -67,21 +67,22 @@ where
         let mut this = self.project();
         Poll::Ready(loop {
             if let Some(fut) = this.future.as_mut().as_pin_mut() {
-                // we're currently processing a future to produce a new accum value
-                let acc = this.accum.unwrap() || ready!(fut.poll(cx));
-                if acc {
+                // we're currently processing a future to produce a new value
+                let res = ready!(fut.poll(cx));
+                this.future.set(None);
+                if res {
+                    *this.done = true;
                     break true;
                 } // early exit
-                *this.accum = Some(acc);
-                this.future.set(None);
-            } else if this.accum.is_some() {
+            } else if !*this.done {
                 // we're waiting on a new item from the stream
                 match ready!(this.stream.as_mut().poll_next(cx)) {
                     Some(item) => {
                         this.future.set(Some((this.f)(item)));
                     }
                     None => {
-                        break this.accum.take().unwrap();
+                        *this.done = true;
+                        break false;
                     }
                 }
             } else {

futures/tests/stream.rs

@@ -552,3 +552,43 @@ fn select_with_strategy_doesnt_terminate_early() {
         assert_eq!(count.get(), times_should_poll + 1);
     }
 }
+
+async fn is_even(number: u8) -> bool {
+    number % 2 == 0
+}
+
+#[test]
+fn all() {
+    block_on(async {
+        let empty: [u8; 0] = [];
+        let st = stream::iter(empty);
+        let all = st.all(is_even).await;
+        assert!(all);
+
+        let st = stream::iter([2, 4, 6, 8]);
+        let all = st.all(is_even).await;
+        assert!(all);
+
+        let st = stream::iter([2, 3, 4]);
+        let all = st.all(is_even).await;
+        assert!(!all);
+    });
+}
+
+#[test]
+fn any() {
+    block_on(async {
+        let empty: [u8; 0] = [];
+        let st = stream::iter(empty);
+        let any = st.any(is_even).await;
+        assert!(!any);
+
+        let st = stream::iter([1, 2, 3]);
+        let any = st.any(is_even).await;
+        assert!(any);
+
+        let st = stream::iter([1, 3, 5]);
+        let any = st.any(is_even).await;
+        assert!(!any);
+    });
+}