diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs index 07cd03a..3bf95e3 100644 --- a/src/stream/stream/mod.rs +++ b/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 = vec![1].into_iter().collect(); + let single_ne: VecDeque = vec![10].into_iter().collect(); + let multi: VecDeque = vec![1,2].into_iter().collect(); + let multi_ne: VecDeque = 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( + self, + other: S + ) -> impl Future [NeFuture] + where + Self: Sized + Stream, + S: Sized + Stream, + ::Item: PartialEq, + { + NeFuture::new(self, other) + } + #[doc = r#" Determines if the elements of this `Stream` are lexicographically greater than or equal to those of another. diff --git a/src/stream/stream/ne.rs b/src/stream/stream/ne.rs new file mode 100644 index 0000000..ffeaca8 --- /dev/null +++ b/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 { + #[pin] + l: Fuse, + #[pin] + r: Fuse, + } +} + +impl NeFuture +where + L::Item: PartialEq, +{ + pub(super) fn new(l: L, r: R) -> Self { + Self { + l: l.fuse(), + r: r.fuse(), + } + } +} + +impl Future for NeFuture +where + L: Stream + Sized, + R: Stream + Sized, + L::Item: PartialEq, +{ + type Output = bool; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + 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); + } + } + } + } +}