Adds latest version of relevant files

Author: assemblaj

src/stream/stream/mod.rs

@@ -30,8 +30,10 @@ mod filter_map;
 mod find;
 mod find_map;
 mod fold;
+mod for_each;
 mod fuse;
 mod inspect;
+mod map;
 mod min_by;
 mod next;
 mod nth;
@@ -41,6 +43,7 @@ mod skip;
 mod skip_while;
 mod step_by;
 mod take;
+mod try_for_each;
 mod zip;
 
 use all::AllFuture;
@@ -50,15 +53,18 @@ use filter_map::FilterMap;
 use find::FindFuture;
 use find_map::FindMapFuture;
 use fold::FoldFuture;
+use for_each::ForEachFuture;
 use min_by::MinByFuture;
 use next::NextFuture;
 use nth::NthFuture;
 use partial_cmp::PartialCmpFuture;
+use try_for_each::TryForEeachFuture;
 
 pub use chain::Chain;
 pub use filter::Filter;
 pub use fuse::Fuse;
 pub use inspect::Inspect;
+pub use map::Map;
 pub use scan::Scan;
 pub use skip::Skip;
 pub use skip_while::SkipWhile;
@@ -336,6 +342,37 @@ extension_trait! {
             Enumerate::new(self)
         }
 
+        #[doc = r#"
+            Takes a closure and creates a stream that calls that closure on every element of this stream.
+
+            # Examples
+
+            ```
+            # fn main() { async_std::task::block_on(async {
+            #
+            use async_std::prelude::*;
+            use std::collections::VecDeque;
+
+            let s: VecDeque<_> = vec![1, 2, 3].into_iter().collect();
+            let mut s = s.map(|x| 2 * x);
+
+            assert_eq!(s.next().await, Some(2));
+            assert_eq!(s.next().await, Some(4));
+            assert_eq!(s.next().await, Some(6));
+            assert_eq!(s.next().await, None);
+
+            #
+            # }) }
+            ```
+        "#]
+        fn map<B, F>(self, f: F) -> Map<Self, F, Self::Item, B>
+        where
+            Self: Sized,
+            F: FnMut(Self::Item) -> B,
+        {
+            Map::new(self, f)
+        }
+
         #[doc = r#"
             A combinator that does something with each element in the stream, passing the value
             on.
@@ -752,6 +789,41 @@ extension_trait! {
             FoldFuture::new(self, init, f)
         }
 
+        #[doc = r#"
+            Call a closure on each element of the stream.
+
+            # Examples
+
+            ```
+            # fn main() { async_std::task::block_on(async {
+            #
+            use async_std::prelude::*;
+            use std::collections::VecDeque;
+            use std::sync::mpsc::channel;
+
+            let (tx, rx) = channel();
+
+            let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
+            let sum = s.for_each(move |x| tx.clone().send(x).unwrap()).await;
+
+            let v: Vec<_> = rx.iter().collect();
+
+            assert_eq!(v, vec![1, 2, 3]);
+            #
+            # }) }
+            ```
+        "#]
+        fn for_each<F>(
+            self,
+            f: F,
+        ) -> impl Future<Output = ()> [ForEachFuture<Self, F, Self::Item>]
+        where
+            Self: Sized,
+            F: FnMut(Self::Item),
+        {
+            ForEachFuture::new(self, f)
+        }
+
         #[doc = r#"
             Tests if any element of the stream matches a predicate.
 
@@ -923,6 +995,51 @@ extension_trait! {
             Skip::new(self, n)
         }
 
+        #[doc = r#"
+            Applies a falliable function to each element in a stream, stopping at first error and returning it.
+
+            # Examples
+
+            ```
+            # fn main() { async_std::task::block_on(async {
+            #
+            use std::collections::VecDeque;
+            use std::sync::mpsc::channel;
+            use async_std::prelude::*;
+
+            let (tx, rx) = channel();
+
+            let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
+            let s = s.try_for_each(|v| {
+                if v % 2 == 1 {
+                    tx.clone().send(v).unwrap();
+                    Ok(())
+                } else {
+                    Err("even")
+                }
+            });
+
+            let res = s.await;
+            drop(tx);
+            let values: Vec<_> = rx.iter().collect();
+
+            assert_eq!(values, vec![1]);
+            assert_eq!(res, Err("even"));
+            #
+            # }) }
+            ```
+        "#]
+        fn try_for_each<F, E>(
+            self,
+            f: F,
+        ) -> impl Future<Output = E> [TryForEeachFuture<Self, F, Self::Item, E>]
+        where
+            Self: Sized,
+            F: FnMut(Self::Item) -> Result<(), E>,
+        {
+            TryForEeachFuture::new(self, f)
+        }
+
         #[doc = r#"
             'Zips up' two streams into a single stream of pairs.
 
@@ -1064,15 +1181,15 @@ extension_trait! {
         fn partial_cmp<S>(
            self,
            other: S
-        ) -> impl Future<Output = Option<Ordering>> + '_ [PartialCmpFuture<Self, S>] 
+        ) -> impl Future<Output = Option<Ordering>>  [PartialCmpFuture<Self, S>]
         where
             Self: Sized + Stream,
-            S: Stream,             
-            Self::Item: PartialOrd<S::Item>,
+            S: Stream,
+            <Self as Stream>::Item: PartialOrd<S::Item>,
         {
             PartialCmpFuture::new(self, other)
         }
-        
+
     }
 
     impl<S: Stream + Unpin + ?Sized> Stream for Box<S> {

src/stream/stream/partial_cmp.rs

@@ -2,8 +2,8 @@ use std::cmp::Ordering;
 use std::pin::Pin;
 
 use super::fuse::Fuse;
-use crate::prelude::*;
 use crate::future::Future;
+use crate::prelude::*;
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
@@ -15,7 +15,7 @@ pub struct PartialCmpFuture<L: Stream, R: Stream> {
     l: Fuse<L>,
     r: Fuse<R>,
     l_cache: Option<L::Item>,
-    r_cache: Option<R::Item>, 
+    r_cache: Option<R::Item>,
 }
 
 impl<L: Stream, R: Stream> PartialCmpFuture<L, R> {
@@ -28,36 +28,36 @@ impl<L: Stream, R: Stream> PartialCmpFuture<L, R> {
         PartialCmpFuture {
             l: l.fuse(),
             r: r.fuse(),
-            l_cache: None, 
-            r_cache: None, 
+            l_cache: None,
+            r_cache: None,
         }
     }
 }
 
-impl<L: Stream, R: Stream> Future for PartialCmpFuture<L, R> 
-where 
+impl<L: Stream, R: Stream> Future for PartialCmpFuture<L, R>
+where
     L: Stream + Sized,
     R: Stream + Sized,
-    L::Item: PartialOrd<R::Item>
+    L::Item: PartialOrd<R::Item>,
 {
     type Output = Option<Ordering>;
 
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {        
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         loop {
-            // Short circuit logic 
+            // Short circuit logic
             // Stream that completes earliest can be considered Less, etc
             let l_complete = self.l.done && self.as_mut().l_cache.is_none();
             let r_complete = self.r.done && self.as_mut().r_cache.is_none();
 
             if l_complete && r_complete {
-                return Poll::Ready(Some(Ordering::Equal))
+                return Poll::Ready(Some(Ordering::Equal));
             } else if l_complete {
-                return Poll::Ready(Some(Ordering::Less))
+                return Poll::Ready(Some(Ordering::Less));
             } else if r_complete {
-                return Poll::Ready(Some(Ordering::Greater))
+                return Poll::Ready(Some(Ordering::Greater));
             }
 
-            // Get next value if possible and necesary 
+            // Get next value if possible and necesary
             if !self.l.done && self.as_mut().l_cache.is_none() {
                 let l_next = futures_core::ready!(self.as_mut().l().poll_next(cx));
                 if let Some(item) = l_next {
@@ -75,17 +75,17 @@ where
             // Compare if both values are available.
             if self.as_mut().l_cache.is_some() && self.as_mut().r_cache.is_some() {
                 let l_value = self.as_mut().l_cache().take().unwrap();
-                let r_value = self.as_mut().r_cache().take().unwrap(); 
+                let r_value = self.as_mut().r_cache().take().unwrap();
                 let result = l_value.partial_cmp(&r_value);
 
                 if let Some(Ordering::Equal) = result {
-                    // Reset cache to prepare for next comparison 
-                    *self.as_mut().l_cache() = None;  
+                    // Reset cache to prepare for next comparison
+                    *self.as_mut().l_cache() = None;
                     *self.as_mut().r_cache() = None;
                 } else {
-                    // Return non equal value 
+                    // Return non equal value
                     return Poll::Ready(result);
-                }            
+                }
             }
         }
     }