Adds Stream::cmp (#273)

* Adds cmp

* Fixes formatting

* cleans up examples

* attempts to fix rustdoc issue

* formats with cargo fmt

* Adds proper trait bounds for cmp
pull/352/head
assemblaj 5 years ago committed by Stjepan Glavina
parent a7041be6f2
commit 4b96ea1273

@ -0,0 +1,91 @@
use std::cmp::Ordering;
use std::pin::Pin;
use super::fuse::Fuse;
use crate::future::Future;
use crate::prelude::*;
use crate::stream::Stream;
use crate::task::{Context, Poll};
// Lexicographically compares the elements of this `Stream` with those
// of another using `Ord`.
#[doc(hidden)]
#[allow(missing_debug_implementations)]
pub struct CmpFuture<L: Stream, R: Stream> {
l: Fuse<L>,
r: Fuse<R>,
l_cache: Option<L::Item>,
r_cache: Option<R::Item>,
}
impl<L: Stream, R: Stream> CmpFuture<L, R> {
pin_utils::unsafe_pinned!(l: Fuse<L>);
pin_utils::unsafe_pinned!(r: Fuse<R>);
pin_utils::unsafe_unpinned!(l_cache: Option<L::Item>);
pin_utils::unsafe_unpinned!(r_cache: Option<R::Item>);
pub(super) fn new(l: L, r: R) -> Self {
CmpFuture {
l: l.fuse(),
r: r.fuse(),
l_cache: None,
r_cache: None,
}
}
}
impl<L: Stream, R: Stream> Future for CmpFuture<L, R>
where
L: Stream + Sized,
R: Stream<Item = L::Item> + Sized,
L::Item: Ord,
{
type Output = Ordering;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
loop {
// Stream that completes earliest can be considered Less, etc
let l_complete = self.l.done && self.as_mut().l_cache.is_none();
let r_complete = self.r.done && self.as_mut().r_cache.is_none();
if l_complete && r_complete {
return Poll::Ready(Ordering::Equal);
} else if l_complete {
return Poll::Ready(Ordering::Less);
} else if r_complete {
return Poll::Ready(Ordering::Greater);
}
// Get next value if possible and necesary
if !self.l.done && self.as_mut().l_cache.is_none() {
let l_next = futures_core::ready!(self.as_mut().l().poll_next(cx));
if let Some(item) = l_next {
*self.as_mut().l_cache() = Some(item);
}
}
if !self.r.done && self.as_mut().r_cache.is_none() {
let r_next = futures_core::ready!(self.as_mut().r().poll_next(cx));
if let Some(item) = r_next {
*self.as_mut().r_cache() = Some(item);
}
}
// Compare if both values are available.
if self.as_mut().l_cache.is_some() && self.as_mut().r_cache.is_some() {
let l_value = self.as_mut().l_cache().take().unwrap();
let r_value = self.as_mut().r_cache().take().unwrap();
let result = l_value.cmp(&r_value);
if let Ordering::Equal = result {
// Reset cache to prepare for next comparison
*self.as_mut().l_cache() = None;
*self.as_mut().r_cache() = None;
} else {
// Return non equal value
return Poll::Ready(result);
}
}
}
}
}

@ -24,6 +24,7 @@
mod all;
mod any;
mod chain;
mod cmp;
mod enumerate;
mod filter;
mod filter_map;
@ -53,6 +54,7 @@ mod zip;
use all::AllFuture;
use any::AnyFuture;
use cmp::CmpFuture;
use enumerate::Enumerate;
use filter_map::FilterMap;
use find::FindFuture;
@ -1270,6 +1272,43 @@ extension_trait! {
PartialCmpFuture::new(self, other)
}
#[doc = r#"
Lexicographically compares the elements of this `Stream` with those
of another using 'Ord'.
# Examples
```
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use std::cmp::Ordering;
let s1 = VecDeque::from(vec![1]);
let s2 = VecDeque::from(vec![1, 2]);
let s3 = VecDeque::from(vec![1, 2, 3]);
let s4 = VecDeque::from(vec![1, 2, 4]);
assert_eq!(s1.clone().cmp(s1.clone()).await, Ordering::Equal);
assert_eq!(s1.clone().cmp(s2.clone()).await, Ordering::Less);
assert_eq!(s2.clone().cmp(s1.clone()).await, Ordering::Greater);
assert_eq!(s3.clone().cmp(s4.clone()).await, Ordering::Less);
assert_eq!(s4.clone().cmp(s3.clone()).await, Ordering::Greater);
#
# }) }
```
"#]
fn cmp<S>(
self,
other: S
) -> impl Future<Output = Ordering> [CmpFuture<Self, S>]
where
Self: Sized + Stream,
S: Stream,
<Self as Stream>::Item: Ord
{
CmpFuture::new(self, other)
}
#[doc = r#"
Determines if the elements of this `Stream` are lexicographically

Loading…
Cancel
Save