Use take_while instead of scan for Option

Author: olegnn

src/option/from_stream.rs

@@ -17,24 +17,25 @@ where
         let stream = stream.into_stream();
 
         Box::pin(async move {
-            // Using `scan` here because it is able to stop the stream early
+            // Using `take_while` here because it is able to stop the stream early
             // if a failure occurs
-            let mut found_error = false;
+            let mut found_none = false;
             let out: V = stream
-                .scan((), |_, elem| {
-                    match elem {
-                        Some(elem) => Some(elem),
-                        None => {
-                            found_error = true;
-                            // Stop processing the stream on error
-                            None
-                        }
+                .take_while(|elem| {
+                    elem.is_some() || {
+                        found_none = true;
+                        false
                     }
                 })
+                .map(Option::unwrap)
                 .collect()
                 .await;
 
-            if found_error { None } else { Some(out) }
+            if found_none {
+                None
+            } else {
+                Some(out)
+            }
         })
     }
 }

src/option/product.rs

@@ -1,7 +1,7 @@
 use std::pin::Pin;
 
 use crate::prelude::*;
-use crate::stream::{Stream, Product};
+use crate::stream::{Product, Stream};
 
 impl<T, U> Product<Option<U>> for Option<T>
 where
@@ -36,23 +36,24 @@ where
         ```
     "#]
     fn product<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Option<T>> + 'a>>
-        where S: Stream<Item = Option<U>> + 'a
+    where
+        S: Stream<Item = Option<U>> + 'a,
     {
         Box::pin(async move {
-            // Using `scan` here because it is able to stop the stream early
+            // Using `take_while` here because it is able to stop the stream early
             // if a failure occurs
             let mut found_none = false;
-            let out = <T as Product<U>>::product(stream
-                .scan((), |_, elem| {
-                    match elem {
-                        Some(elem) => Some(elem),
-                        None => {
+            let out = <T as Product<U>>::product(
+                stream
+                    .take_while(|elem| {
+                        elem.is_some() || {
                             found_none = true;
-                            // Stop processing the stream on error
-                            None
+                            false
                         }
-                    }
-                })).await;
+                    })
+                    .map(Option::unwrap),
+            )
+            .await;
 
             if found_none {
                 None

src/option/sum.rs

@@ -31,23 +31,24 @@ where
         ```
     "#]
     fn sum<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Option<T>> + 'a>>
-        where S: Stream<Item = Option<U>> + 'a
+    where
+        S: Stream<Item = Option<U>> + 'a,
     {
         Box::pin(async move {
-            // Using `scan` here because it is able to stop the stream early
+            // Using `take_while` here because it is able to stop the stream early
             // if a failure occurs
             let mut found_none = false;
-            let out = <T as Sum<U>>::sum(stream
-                .scan((), |_, elem| {
-                    match elem {
-                        Some(elem) => Some(elem),
-                        None => {
+            let out = <T as Sum<U>>::sum(
+                stream
+                    .take_while(|elem| {
+                        elem.is_some() || {
                             found_none = true;
-                            // Stop processing the stream on error
-                            None
+                            false
                         }
-                    }
-                })).await;
+                    })
+                    .map(Option::unwrap),
+            )
+            .await;
 
             if found_none {
                 None