From 134089af2c711e3786beb4174e0f1592d9de7752 Mon Sep 17 00:00:00 2001
From: Oleg Nosov <olegnosov1@gmail.com>
Date: Wed, 15 Jan 2020 08:55:18 +0300
Subject: [PATCH] Use `filter_map(identity)` + other fixes

---
 src/option/from_stream.rs |  3 ++-
 src/option/product.rs     |  3 ++-
 src/option/sum.rs         |  5 +++--
 src/result/from_stream.rs |  2 +-
 src/result/product.rs     | 28 +++++++++++++++-------------
 src/result/sum.rs         | 26 ++++++++++++++------------
 src/unit/from_stream.rs   |  2 +-
 src/utils.rs              |  8 ++++++++
 8 files changed, 46 insertions(+), 31 deletions(-)

diff --git a/src/option/from_stream.rs b/src/option/from_stream.rs
index f0aafea..2194f29 100644
--- a/src/option/from_stream.rs
+++ b/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;
 
diff --git a/src/option/product.rs b/src/option/product.rs
index 2238a91..f733bef 100644
--- a/src/option/product.rs
+++ b/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;
 
diff --git a/src/option/sum.rs b/src/option/sum.rs
index 0570a1e..64fdb07 100644
--- a/src/option/sum.rs
+++ b/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;
 
diff --git a/src/result/from_stream.rs b/src/result/from_stream.rs
index a8490d6..424a53b 100644
--- a/src/result/from_stream.rs
+++ b/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) => {
diff --git a/src/result/product.rs b/src/result/product.rs
index ec9d94a..ad2e014 100644
--- a/src/result/product.rs
+++ b/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
-                .scan((), |_, elem| {
-                    match elem {
-                        Ok(elem) => Some(elem),
-                        Err(err) => {
-                            found_error = Some(err);
-                            // Stop processing the stream on error
-                            None
-                        }
+            let out = <T as Product<U>>::product(stream.scan((), |(), elem| {
+                match elem {
+                    Ok(elem) => Some(elem),
+                    Err(err) => {
+                        found_error = Some(err);
+                        // Stop processing the stream on error
+                        None
                     }
-                })).await;
+                }
+            }))
+            .await;
+
             match found_error {
                 Some(err) => Err(err),
-                None => Ok(out)
+                None => Ok(out),
             }
         })
     }
diff --git a/src/result/sum.rs b/src/result/sum.rs
index ccc4240..1cf5b7a 100644
--- a/src/result/sum.rs
+++ b/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
-                .scan((), |_, elem| {
-                    match elem {
-                        Ok(elem) => Some(elem),
-                        Err(err) => {
-                            found_error = Some(err);
-                            // Stop processing the stream on error
-                            None
-                        }
+            let out = <T as Sum<U>>::sum(stream.scan((), |(), elem| {
+                match elem {
+                    Ok(elem) => Some(elem),
+                    Err(err) => {
+                        found_error = Some(err);
+                        // Stop processing the stream on error
+                        None
                     }
-                })).await;
+                }
+            }))
+            .await;
+
             match found_error {
                 Some(err) => Err(err),
-                None => Ok(out)
+                None => Ok(out),
             }
         })
     }
diff --git a/src/unit/from_stream.rs b/src/unit/from_stream.rs
index da216e2..7b78438 100644
--- a/src/unit/from_stream.rs
+++ b/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))
     }
 }
diff --git a/src/utils.rs b/src/utils.rs
index ef50ed0..780e733 100644
--- a/src/utils.rs
+++ b/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;