Use `filter_map(identity)` + other fixes

src/option/from_stream.rs

@@ -2,6 +2,7 @@ use std::pin::Pin;
 
 use crate::prelude::*;
 use crate::stream::{FromStream, IntoStream};
+use crate::utils::identity;
 
 impl<T, V> FromStream<Option<T>> for Option<V>
 where
@@ -28,7 +29,7 @@ where
                         false
                     }
                 })
-                .map(Option::unwrap)
+                .filter_map(identity)
                 .collect()
                 .await;
 

src/option/product.rs

@@ -2,6 +2,7 @@ use std::pin::Pin;
 
 use crate::prelude::*;
 use crate::stream::{Product, Stream};
+use crate::utils::identity;
 
 impl<T, U> Product<Option<U>> for Option<T>
 where
@@ -52,7 +53,7 @@ where
                             false
                         }
                     })
-                    .map(Option::unwrap),
+                    .filter_map(identity),
             )
             .await;
 

src/option/sum.rs

@@ -2,6 +2,7 @@ use std::pin::Pin;
 
 use crate::prelude::*;
 use crate::stream::{Stream, Sum};
+use crate::utils::identity;
 
 impl<T, U> Sum<Option<U>> for Option<T>
 where
@@ -43,11 +44,11 @@ where
                     .take_while(|elem| {
                         elem.is_some() || {
                             found_none = true;
-                            // Stop processing the stream on error
+                            // Stop processing the stream on `None`
                             false
                         }
                     })
-                    .map(Option::unwrap),
+                    .filter_map(identity),
             )
             .await;
 

src/result/from_stream.rs

@@ -21,7 +21,7 @@ where
             // if a failure occurs
             let mut found_error = None;
             let out: V = stream
-                .scan((), |_, elem| {
+                .scan((), |(), elem| {
                     match elem {
                         Ok(elem) => Some(elem),
                         Err(err) => {

src/result/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, E> Product<Result<U, E>> for Result<T, E>
 where
@@ -36,26 +36,28 @@ where
         ```
     "#]
     fn product<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Result<T, E>> + 'a>>
-        where S: Stream<Item = Result<U, E>> + 'a
+    where
+        S: Stream<Item = Result<U, E>> + 'a,
     {
         Box::pin(async move {
             // Using `scan` here because it is able to stop the stream early
             // if a failure occurs
             let mut found_error = None;
-            let out = <T as Product<U>>::product(stream
+            let out = <T as Product<U>>::product(stream.scan((), |(), elem| {
-                .scan((), |_, elem| {
+                match elem {
-                    match elem {
+                    Ok(elem) => Some(elem),
-                        Ok(elem) => Some(elem),
+                    Err(err) => {
-                        Err(err) => {
+                        found_error = Some(err);
-                            found_error = Some(err);
+                        // Stop processing the stream on error
-                            // Stop processing the stream on error
+                        None
-                            None
-                        }
                     }
-                })).await;
+                }
+            }))
+            .await;
+
             match found_error {
                 Some(err) => Err(err),
-                None => Ok(out)
+                None => Ok(out),
             }
         })
     }

src/result/sum.rs

@@ -36,26 +36,28 @@ where
         ```
     "#]
     fn sum<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Result<T, E>> + 'a>>
-        where S: Stream<Item = Result<U, E>> + 'a
+    where
+        S: Stream<Item = Result<U, E>> + 'a,
     {
         Box::pin(async move {
             // Using `scan` here because it is able to stop the stream early
             // if a failure occurs
             let mut found_error = None;
-            let out = <T as Sum<U>>::sum(stream
+            let out = <T as Sum<U>>::sum(stream.scan((), |(), elem| {
-                .scan((), |_, elem| {
+                match elem {
-                    match elem {
+                    Ok(elem) => Some(elem),
-                        Ok(elem) => Some(elem),
+                    Err(err) => {
-                        Err(err) => {
+                        found_error = Some(err);
-                            found_error = Some(err);
+                        // Stop processing the stream on error
-                            // Stop processing the stream on error
+                        None
-                            None
-                        }
                     }
-                })).await;
+                }
+            }))
+            .await;
+
             match found_error {
                 Some(err) => Err(err),
-                None => Ok(out)
+                None => Ok(out),
             }
         })
     }

src/unit/from_stream.rs

@@ -8,6 +8,6 @@ impl FromStream<()> for () {
     fn from_stream<'a, S: IntoStream<Item = ()> + 'a>(
         stream: S,
     ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
-        Box::pin(stream.into_stream().for_each(|_| ()))
+        Box::pin(stream.into_stream().for_each(drop))
     }
 }

src/utils.rs

@@ -52,6 +52,14 @@ pub fn random(n: u32) -> u32 {
     })
 }
 
+/// Returns given argument without changes.
+#[allow(dead_code)]
+#[doc(hidden)]
+#[inline(always)]
+pub(crate) fn identity<T>(arg: T) -> T {
+    arg
+}
+
 /// Add additional context to errors
 pub(crate) trait Context {
     fn context(self, message: impl Fn() -> String) -> Self;