Merge pull request #427 from yjhmelody/stream-ne

Add stream ne

Author: yoshuawuyts

src/stream/stream/mod.rs

@@ -44,6 +44,7 @@ mod map;
 mod max_by;
 mod min_by;
 mod min_by_key;
+mod ne;
 mod next;
 mod nth;
 mod partial_cmp;
@@ -75,6 +76,7 @@ use lt::LtFuture;
 use max_by::MaxByFuture;
 use min_by::MinByFuture;
 use min_by_key::MinByKeyFuture;
+use ne::NeFuture;
 use next::NextFuture;
 use nth::NthFuture;
 use partial_cmp::PartialCmpFuture;
@@ -1588,6 +1590,39 @@ extension_trait! {
             CmpFuture::new(self, other)
         }
 
+           #[doc = r#"
+            Determines if the elements of this `Stream` are lexicographically
+            not 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_ne:  VecDeque<isize> = vec![10].into_iter().collect();
+            let multi:      VecDeque<isize> = vec![1,2].into_iter().collect();
+            let multi_ne:   VecDeque<isize> = vec![1,5].into_iter().collect();
+            assert_eq!(single.clone().ne(single.clone()).await, false);
+            assert_eq!(single_ne.clone().ne(single.clone()).await, true);
+            assert_eq!(multi.clone().ne(single_ne.clone()).await, true);
+            assert_eq!(multi_ne.clone().ne(multi.clone()).await, true);
+            #
+            # }) }
+            ```
+        "#]
+        fn ne<S>(
+           self,
+           other: S
+        ) -> impl Future<Output = bool> [NeFuture<Self, S>]
+        where
+            Self: Sized + Stream,
+            S: Sized + Stream,
+            <Self as Stream>::Item: PartialEq<S::Item>,
+        {
+            NeFuture::new(self, other)
+        }
+
         #[doc = r#"
             Determines if the elements of this `Stream` are lexicographically
             greater than or equal to those of another.

src/stream/stream/ne.rs

@@ -0,0 +1,65 @@
+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 NeFuture<L: Stream, R: Stream> {
+        #[pin]
+        l: Fuse<L>,
+        #[pin]
+        r: Fuse<R>,
+    }
+}
+
+impl<L: Stream, R: Stream> NeFuture<L, R>
+where
+    L::Item: PartialEq<R::Item>,
+{
+    pub(super) fn new(l: L, r: R) -> Self {
+        Self {
+            l: l.fuse(),
+            r: r.fuse(),
+        }
+    }
+}
+
+impl<L: Stream, R: Stream> Future for NeFuture<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(false);
+            }
+
+            match (l_val, r_val) {
+                (Some(l), Some(r)) if l == r => {
+                    continue;
+                }
+                _ => {
+                    return Poll::Ready(true);
+                }
+            }
+        }
+    }
+}