add stream::max_by method

src/stream/stream/max_by.rs

@@ -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),
+        }
+    }
+}

src/stream/stream/mod.rs

@@ -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.