1use core::cmp::Ordering;
2use core::future::Future;
3use core::pin::Pin;
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 MaxByKeyFuture<S, T, K> {
14 #[pin]
15 stream: S,
16 max: Option<(T, T)>,
17 key_by: K,
18 }
19}
20
21impl<S, T, K> MaxByKeyFuture<S, T, K> {
22 pub(super) fn new(stream: S, key_by: K) -> Self {
23 Self {
24 stream,
25 max: None,
26 key_by,
27 }
28 }
29}
30
31impl<S, K> Future for MaxByKeyFuture<S, S::Item, K>
32where
33 S: Stream,
34 K: FnMut(&S::Item) -> S::Item,
35 S::Item: Ord,
36{
37 type Output = Option<S::Item>;
38
39 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
40 fn key<B, T>(mut f: impl FnMut(&T) -> B) -> impl FnMut(T) -> (B, T) {
41 move |x| (f(&x), x)
42 }
43
44 let this = self.project();
45 let next = futures_core::ready!(this.stream.poll_next(cx));
46
47 match next {
48 Some(new) => {
49 let (key, value) = key(this.key_by)(new);
50 cx.waker().wake_by_ref();
51
52 match this.max.take() {
53 None => *this.max = Some((key, value)),
54
55 Some(old) => match key.cmp(&old.0) {
56 Ordering::Greater => *this.max = Some((key, value)),
57 _ => *this.max = Some(old),
58 },
59 }
60 Poll::Pending
61 }
62 None => Poll::Ready(this.max.take().map(|max| max.1)),
63 }
64 }
65}
66