diff --git a/src/stream/stream/min.rs b/src/stream/stream/min.rs new file mode 100644 index 0000000..1ab5606 --- /dev/null +++ b/src/stream/stream/min.rs @@ -0,0 +1,60 @@ +use std::marker::PhantomData; +use std::cmp::{Ordering, Ord}; +use std::pin::Pin; + +use pin_project_lite::pin_project; + +use crate::future::Future; +use crate::stream::Stream; +use crate::task::{Context, Poll}; + +pin_project! { + #[doc(hidden)] + #[allow(missing_debug_implementations)] + pub struct MinFuture { + #[pin] + stream: S, + _compare: PhantomData, + min: Option, + } +} + +impl MinFuture { + pub(super) fn new(stream: S) -> Self { + Self { + stream, + _compare: PhantomData, + min: None, + } + } +} + +impl Future for MinFuture +where + S: Stream, + S::Item: Ord, + F: FnMut(&S::Item, &S::Item) -> Ordering, +{ + type Output = Option; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.project(); + let next = futures_core::ready!(this.stream.poll_next(cx)); + + match next { + Some(new) => { + cx.waker().wake_by_ref(); + match this.min.take() { + None => *this.min = Some(new), + + Some(old) => match new.cmp(&old) { + Ordering::Less => *this.min = Some(new), + _ => *this.min = Some(old), + }, + } + Poll::Pending + } + None => Poll::Ready(this.min.take()), + } + } +} diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs index 3e5dfe4..71139f4 100644 --- a/src/stream/stream/mod.rs +++ b/src/stream/stream/mod.rs @@ -42,6 +42,7 @@ mod le; mod lt; mod map; mod max_by; +mod min; mod min_by; mod min_by_key; mod ne; @@ -75,6 +76,7 @@ use last::LastFuture; use le::LeFuture; use lt::LtFuture; use max_by::MaxByFuture; +use min::MinFuture; use min_by::MinByFuture; use min_by_key::MinByKeyFuture; use ne::NeFuture; @@ -766,6 +768,40 @@ extension_trait! { MinByFuture::new(self, compare) } + #[doc = r#" + Returns the element that gives the minimum value. 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 = vec![1, 2, 3].into_iter().collect(); + + let min = s.clone().min().await; + assert_eq!(min, Some(1)); + + let min = VecDeque::::new().min().await; + assert_eq!(min, None); + # + # }) } + ``` + "#] + fn min( + self, + ) -> impl Future> [MinFuture] + where + Self: Sized, + F: FnMut(&Self::Item, &Self::Item) -> Ordering, + { + MinFuture::new(self) + } + #[doc = r#" Returns the element that gives the maximum value with respect to the specified comparison function. If several elements are equally maximum,