Adds stream map combinator

Author: montekki

src/stream/stream/map.rs

@@ -0,0 +1,41 @@
+use std::marker::PhantomData;
+use std::pin::Pin;
+
+use crate::stream::Stream;
+use crate::task::{Context, Poll};
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct Map<S, F, T, B> {
+ stream: S,
+ f: F,
+ __from: PhantomData<T>,
+ __to: PhantomData<B>,
+}
+
+impl<S, F, T, B> Map<S, F, T, B> {
+ pin_utils::unsafe_pinned!(stream: S);
+ pin_utils::unsafe_unpinned!(f: F);
+
+ pub(crate) fn new(stream: S, f: F) -> Self {
+ Map {
+ stream,
+ f,
+ __from: PhantomData,
+ __to: PhantomData,
+ }
+ }
+}
+
+impl<S, F, B> Stream for Map<S, F, S::Item, B>
+where
+ S: Stream,
+ F: FnMut(S::Item) -> B,
+{
+ type Item = B;
+
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
+ Poll::Ready(next.map(self.as_mut().f()))
+ }
+}

src/stream/stream/mod.rs

@@ -32,6 +32,7 @@ mod find_map;
 mod fold;
 mod fuse;
 mod inspect;
+mod map;
 mod min_by;
 mod next;
 mod nth;
@@ -57,6 +58,7 @@ 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;
@@ -334,6 +336,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.