From f5efaaa7ba82e6a0707a82ffa6cda499fdb6d694 Mon Sep 17 00:00:00 2001
From: yjhmelody <465402634@qq.com>
Date: Thu, 31 Oct 2019 14:44:19 +0800
Subject: [PATCH 1/2] Add stream eq

---
 src/stream/stream/eq.rs  | 61 ++++++++++++++++++++++++++++++++++++++++
 src/stream/stream/mod.rs | 38 +++++++++++++++++++++++++
 2 files changed, 99 insertions(+)
 create mode 100644 src/stream/stream/eq.rs

diff --git a/src/stream/stream/eq.rs b/src/stream/stream/eq.rs
new file mode 100644
index 0000000..42a37d8
--- /dev/null
+++ b/src/stream/stream/eq.rs
@@ -0,0 +1,61 @@
+use std::pin::Pin;
+
+use pin_project_lite::pin_project;
+
+use super::fuse::Fuse;
+use crate::future::Future;
+use crate::prelude::*;
+use crate::stream::Stream;
+use crate::task::{Context, Poll};
+
+pin_project! {
+    // Lexicographically compares the elements of this `Stream` with those
+    // of another.
+    #[doc(hidden)]
+    #[allow(missing_debug_implementations)]
+    pub struct EqFuture<L: Stream, R: Stream> {
+        #[pin]
+        l: Fuse<L>,
+        #[pin]
+        r: Fuse<R>,
+    }
+}
+
+impl<L: Stream, R: Stream> EqFuture<L, R>
+where
+    L::Item: PartialEq<R::Item>,
+{
+    pub(super) fn new(l: L, r: R) -> Self {
+        EqFuture {
+            l: l.fuse(),
+            r: r.fuse(),
+        }
+    }
+}
+
+impl<L: Stream, R: Stream> Future for EqFuture<L, R>
+    where
+        L: Stream + Sized,
+        R: Stream + Sized,
+        L::Item: PartialEq<R::Item>,
+{
+    type Output = bool;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut this = self.project();
+
+        loop {
+            let l_val = futures_core::ready!(this.l.as_mut().poll_next(cx));
+            let r_val = futures_core::ready!(this.r.as_mut().poll_next(cx));
+
+            if this.l.done && this.r.done {
+                return Poll::Ready(true);
+            }
+
+            match (l_val, r_val) {
+                (Some(l), Some(r)) if l != r => {return Poll::Ready(false);},
+                _ => {},
+            }
+        }
+    }
+}
diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
index e819038..07cd03a 100644
--- a/src/stream/stream/mod.rs
+++ b/src/stream/stream/mod.rs
@@ -26,6 +26,7 @@ mod any;
 mod chain;
 mod cmp;
 mod enumerate;
+mod eq;
 mod filter;
 mod filter_map;
 mod find;
@@ -60,6 +61,7 @@ use all::AllFuture;
 use any::AnyFuture;
 use cmp::CmpFuture;
 use enumerate::Enumerate;
+use eq::EqFuture;
 use filter_map::FilterMap;
 use find::FindFuture;
 use find_map::FindMapFuture;
@@ -1622,6 +1624,42 @@ extension_trait! {
             GeFuture::new(self, other)
         }
 
+        #[doc = r#"
+            Determines if the elements of this `Stream` are lexicographically
+            equal to those of another.
+
+            # Examples
+
+            ```
+            # fn main() { async_std::task::block_on(async {
+            #
+            use async_std::prelude::*;
+            use std::collections::VecDeque;
+
+            let single:     VecDeque<isize> = vec![1].into_iter().collect();
+            let single_eq:  VecDeque<isize> = vec![10].into_iter().collect();
+            let multi:      VecDeque<isize> = vec![1,2].into_iter().collect();
+            let multi_eq:   VecDeque<isize> = vec![1,5].into_iter().collect();
+            assert_eq!(single.clone().eq(single.clone()).await, true);
+            assert_eq!(single_eq.clone().eq(single.clone()).await, false);
+            assert_eq!(multi.clone().eq(single_eq.clone()).await, false);
+            assert_eq!(multi_eq.clone().eq(multi.clone()).await, false);
+            #
+            # }) }
+            ```
+        "#]
+        fn eq<S>(
+           self,
+           other: S
+        ) -> impl Future<Output = bool> [EqFuture<Self, S>]
+        where
+            Self: Sized + Stream,
+            S: Sized + Stream,
+            <Self as Stream>::Item: PartialEq<S::Item>,
+        {
+            EqFuture::new(self, other)
+        }
+
         #[doc = r#"
             Determines if the elements of this `Stream` are lexicographically
             greater than those of another.

From 1ab3d901e42fb7c2e9b44303c2b4cdf5116d68b9 Mon Sep 17 00:00:00 2001
From: yjhmelody <465402634@qq.com>
Date: Thu, 31 Oct 2019 21:17:07 +0800
Subject: [PATCH 2/2] fmt code

---
 src/stream/stream/eq.rs | 14 ++++++++------
 1 file changed, 8 insertions(+), 6 deletions(-)

diff --git a/src/stream/stream/eq.rs b/src/stream/stream/eq.rs
index 42a37d8..5343c1a 100644
--- a/src/stream/stream/eq.rs
+++ b/src/stream/stream/eq.rs
@@ -34,10 +34,10 @@ where
 }
 
 impl<L: Stream, R: Stream> Future for EqFuture<L, R>
-    where
-        L: Stream + Sized,
-        R: Stream + Sized,
-        L::Item: PartialEq<R::Item>,
+where
+    L: Stream + Sized,
+    R: Stream + Sized,
+    L::Item: PartialEq<R::Item>,
 {
     type Output = bool;
 
@@ -53,8 +53,10 @@ impl<L: Stream, R: Stream> Future for EqFuture<L, R>
             }
 
             match (l_val, r_val) {
-                (Some(l), Some(r)) if l != r => {return Poll::Ready(false);},
-                _ => {},
+                (Some(l), Some(r)) if l != r => {
+                    return Poll::Ready(false);
+                }
+                _ => {}
             }
         }
     }