add stream::max_by method

This commit is contained in:
yjhmelody 2019-10-23 16:46:11 +08:00
parent ec23632f3e
commit 4e5828e646
2 changed files with 97 additions and 0 deletions

View file

@ -0,0 +1,56 @@
use std::cmp::Ordering;
use std::pin::Pin;
use crate::future::Future;
use crate::stream::Stream;
use crate::task::{Context, Poll};
#[doc(hidden)]
#[allow(missing_debug_implementations)]
pub struct MaxByFuture<S, F, T> {
stream: S,
compare: F,
max: Option<T>,
}
impl<S, F, T> MaxByFuture<S, F, T> {
pin_utils::unsafe_pinned!(stream: S);
pin_utils::unsafe_unpinned!(compare: F);
pin_utils::unsafe_unpinned!(max: Option<T>);
pub(super) fn new(stream: S, compare: F) -> Self {
MaxByFuture {
stream,
compare,
max: None,
}
}
}
impl<S, F> Future for MaxByFuture<S, F, S::Item>
where
S: Stream + Unpin + Sized,
S::Item: Copy,
F: FnMut(&S::Item, &S::Item) -> Ordering,
{
type Output = Option<S::Item>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
match next {
Some(new) => {
cx.waker().wake_by_ref();
match self.as_mut().max().take() {
None => *self.as_mut().max() = Some(new),
Some(old) => match (&mut self.as_mut().compare())(&new, &old) {
Ordering::Greater => *self.as_mut().max() = Some(new),
_ => *self.as_mut().max() = Some(old),
},
}
Poll::Pending
}
None => Poll::Ready(self.max),
}
}
}

View file

@ -40,6 +40,7 @@ mod last;
mod le;
mod lt;
mod map;
mod max_by;
mod min_by;
mod next;
mod nth;
@ -68,6 +69,7 @@ use gt::GtFuture;
use last::LastFuture;
use le::LeFuture;
use lt::LtFuture;
use max_by::MaxByFuture;
use min_by::MinByFuture;
use next::NextFuture;
use nth::NthFuture;
@ -639,6 +641,45 @@ extension_trait! {
MinByFuture::new(self, compare)
}
#[doc = r#"
Returns the element that gives the minimum value with respect to the
specified comparison function. If several elements are equally minimum,
the first element is returned. If the stream is empty, `None` is returned.
# Examples
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let max = s.clone().max_by(|x, y| x.cmp(y)).await;
assert_eq!(max, Some(3));
let max = s.max_by(|x, y| y.cmp(x)).await;
assert_eq!(max, Some(1));
let max = VecDeque::<usize>::new().max_by(|x, y| x.cmp(y)).await;
assert_eq!(max, None);
#
# }) }
```
"#]
fn max_by<F>(
self,
compare: F,
) -> impl Future<Output = Option<Self::Item>> [MaxByFuture<Self, F, Self::Item>]
where
Self: Sized,
F: FnMut(&Self::Item, &Self::Item) -> Ordering,
{
MaxByFuture::new(self, compare)
}
#[doc = r#"
Returns the nth element of the stream.