From 07c6f1e392f9edb2b53da5f02f15751be5ab8265 Mon Sep 17 00:00:00 2001
From: zhangguyu6 <zhangguyu6@gmail.com>
Date: Sat, 2 Nov 2019 22:35:43 +0800
Subject: [PATCH] add Stream::copied

---
 src/stream/stream/copied.rs | 33 ++++++++++++++++++++++++++++++++
 src/stream/stream/mod.rs    | 38 +++++++++++++++++++++++++++++++++++++
 2 files changed, 71 insertions(+)
 create mode 100644 src/stream/stream/copied.rs

diff --git a/src/stream/stream/copied.rs b/src/stream/stream/copied.rs
new file mode 100644
index 000000000..477d59d2f
--- /dev/null
+++ b/src/stream/stream/copied.rs
@@ -0,0 +1,33 @@
+use crate::stream::Stream;
+use crate::task::{Context, Poll};
+use pin_project_lite::pin_project;
+use std::pin::Pin;
+
+pin_project! {
+    #[doc(hidden)]
+    #[allow(missing_debug_implementations)]
+    pub struct Copied<S> {
+        #[pin]
+        stream: S,
+    }
+}
+
+impl<S> Copied<S> {
+    pub(super) fn new(stream: S) -> Self {
+        Copied { stream }
+    }
+}
+
+impl<'a, S, T: 'a> Stream for Copied<S>
+where
+    S: Stream<Item = &'a T>,
+    T: Copy,
+{
+    type Item = T;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.project();
+        let next = futures_core::ready!(this.stream.poll_next(cx));
+        Poll::Ready(next.copied())
+    }
+}
diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
index e46ca9e5c..51ef29d1b 100644
--- a/src/stream/stream/mod.rs
+++ b/src/stream/stream/mod.rs
@@ -25,6 +25,7 @@ mod all;
 mod any;
 mod chain;
 mod cmp;
+mod copied;
 mod enumerate;
 mod eq;
 mod filter;
@@ -88,6 +89,7 @@ use try_fold::TryFoldFuture;
 use try_for_each::TryForEachFuture;
 
 pub use chain::Chain;
+pub use copied::Copied;
 pub use filter::Filter;
 pub use fuse::Fuse;
 pub use inspect::Inspect;
@@ -369,6 +371,42 @@ extension_trait! {
             Chain::new(self, other)
         }
 
+
+        #[doc = r#"
+            Creates an stream which copies all of its elements.
+
+            # Examples
+
+            Basic usage:
+
+            ```
+            # fn main() { async_std::task::block_on(async {
+            #
+            use async_std::prelude::*;
+            use std::collections::VecDeque;
+
+            let v: VecDeque<_> = vec![&1, &2, &3].into_iter().collect();
+            
+            let mut v_copied  = v.copied();
+
+            assert_eq!(v_copied.next().await, Some(1));
+            assert_eq!(v_copied.next().await, Some(2));
+            assert_eq!(v_copied.next().await, Some(3));
+            assert_eq!(v_copied.next().await, None);
+    
+
+            #
+            # }) }
+            ```
+        "#]
+        fn copied<'a,T>(self) -> Copied<Self>
+        where
+            Self: Sized + Stream<Item = &'a T>,
+            T : 'a + Copy,
+        {
+            Copied::new(self)
+        }
+
         #[doc = r#"
             Creates a stream that gives the current element's count as well as the next value.