add: Add stream unzip

Author: k-nasa

src/stream/stream/mod.rs

@@ -124,13 +124,13 @@ cfg_unstable! {
 
     use count::CountFuture;
     use partition::PartitionFuture;
+    use unzip::UnzipFuture;
 
     pub use merge::Merge;
     pub use flatten::Flatten;
     pub use flat_map::FlatMap;
     pub use timeout::{TimeoutError, Timeout};
     pub use throttle::Throttle;
-    pub use unzip::UnzipFuture;
 
     mod count;
     mod merge;

src/stream/stream/unzip.rs

@@ -7,12 +7,13 @@ use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
 pin_project! {
+    #[derive(Clone, Debug)]
     #[cfg(all(feature = "default", feature = "unstable"))]
     #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-    pub struct UnzipFuture<S: Stream, FromA, FromB> {
+    pub struct UnzipFuture<S, FromA, FromB> {
         #[pin]
         stream: S,
-        res: (FromA, FromB),
+        res: Option<(FromA, FromB)>,
     }
 }
 
@@ -24,30 +25,35 @@ where
     pub(super) fn new(stream: S) -> Self {
         UnzipFuture {
             stream,
-            res: (FromA::default(), FromB::default()),
+            res: Some((FromA::default(), FromB::default())),
         }
     }
 }
 
 impl<S, A, B, FromA, FromB> Future for UnzipFuture<S, FromA, FromB>
 where
     S: Stream<Item = (A, B)>,
-    FromA: Default + Extend<A> + Copy,
-    FromB: Default + Extend<B> + Copy,
+    FromA: Default + Extend<A>,
+    FromB: Default + Extend<B>,
 {
     type Output = (FromA, FromB);
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let mut this = self.project();
-        let next = futures_core::ready!(this.stream.as_mut().poll_next(cx));
 
-        match next {
-            Some((a, b)) => {
-                this.res.0.extend(Some(a));
-                this.res.1.extend(Some(b));
-                Poll::Pending
+        loop {
+            let next = futures_core::ready!(this.stream.as_mut().poll_next(cx));
+
+            match next {
+                Some((a, b)) => {
+                    let mut res = this.res.take().unwrap();
+                    res.0.extend(Some(a));
+                    res.1.extend(Some(b));
+
+                    *this.res = Some(res);
+                }
+                None => return Poll::Ready(this.res.take().unwrap()),
             }
-            None => Poll::Ready(*this.res),
         }
     }
 }