async_std/stream/stream/
max_by.rs

1use core::cmp::Ordering;
2use core::pin::Pin;
3use core::future::Future;
4
5use pin_project_lite::pin_project;
6
7use crate::stream::Stream;
8use crate::task::{Context, Poll};
9
10pin_project! {
11    #[doc(hidden)]
12    #[allow(missing_debug_implementations)]
13    pub struct MaxByFuture<S, F, T> {
14        #[pin]
15        stream: S,
16        compare: F,
17        max: Option<T>,
18    }
19}
20
21impl<S, F, T> MaxByFuture<S, F, T> {
22    pub(super) fn new(stream: S, compare: F) -> Self {
23        Self {
24            stream,
25            compare,
26            max: None,
27        }
28    }
29}
30
31impl<S, F> Future for MaxByFuture<S, F, S::Item>
32where
33    S: Stream,
34    F: FnMut(&S::Item, &S::Item) -> Ordering,
35{
36    type Output = Option<S::Item>;
37
38    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
39        let this = self.project();
40        let next = futures_core::ready!(this.stream.poll_next(cx));
41
42        match next {
43            Some(new) => {
44                cx.waker().wake_by_ref();
45                match this.max.take() {
46                    None => *this.max = Some(new),
47                    Some(old) => match (this.compare)(&new, &old) {
48                        Ordering::Greater => *this.max = Some(new),
49                        _ => *this.max = Some(old),
50                    },
51                }
52                Poll::Pending
53            }
54            None => Poll::Ready(this.max.take()),
55        }
56    }
57}
async_std/stream/stream/max_by.rs

async_std/stream/stream/
max_by.rs
