futures_util/stream/stream/
for_each_concurrent.rs

Help
1use crate::stream::{FuturesUnordered, StreamExt};
2use core::fmt;
3use core::num::NonZeroUsize;
4use core::pin::Pin;
5use futures_core::future::{FusedFuture, Future};
6use futures_core::stream::Stream;
7use futures_core::task::{Context, Poll};
8use pin_project_lite::pin_project;
9
10pin_project! {
11    /// Future for the [`for_each_concurrent`](super::StreamExt::for_each_concurrent)
12    /// method.
13    #[must_use = "futures do nothing unless you `.await` or poll them"]
14    pub struct ForEachConcurrent<St, Fut, F> {
15        #[pin]
16        stream: Option<St>,
17        f: F,
18        futures: FuturesUnordered<Fut>,
19        limit: Option<NonZeroUsize>,
20    }
21}
22
23impl<St, Fut, F> fmt::Debug for ForEachConcurrent<St, Fut, F>
24where
25    St: fmt::Debug,
26    Fut: fmt::Debug,
27{
28    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
29        f.debug_struct("ForEachConcurrent")
30            .field("stream", &self.stream)
31            .field("futures", &self.futures)
32            .field("limit", &self.limit)
33            .finish()
34    }
35}
36
37impl<St, Fut, F> ForEachConcurrent<St, Fut, F>
38where
39    St: Stream,
40    F: FnMut(St::Item) -> Fut,
41    Fut: Future<Output = ()>,
42{
43    pub(super) fn new(stream: St, limit: Option<usize>, f: F) -> Self {
44        Self {
45            stream: Some(stream),
46            // Note: `limit` = 0 gets ignored.
47            limit: limit.and_then(NonZeroUsize::new),
48            f,
49            futures: FuturesUnordered::new(),
50        }
51    }
52}
53
54impl<St, Fut, F> FusedFuture for ForEachConcurrent<St, Fut, F>
55where
56    St: Stream,
57    F: FnMut(St::Item) -> Fut,
58    Fut: Future<Output = ()>,
59{
60    fn is_terminated(&self) -> bool {
61        self.stream.is_none() && self.futures.is_empty()
62    }
63}
64
65impl<St, Fut, F> Future for ForEachConcurrent<St, Fut, F>
66where
67    St: Stream,
68    F: FnMut(St::Item) -> Fut,
69    Fut: Future<Output = ()>,
70{
71    type Output = ();
72
73    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
74        let mut this = self.project();
75        loop {
76            let mut made_progress_this_iter = false;
77
78            // Check if we've already created a number of futures greater than `limit`
79            if this.limit.map(|limit| limit.get() > this.futures.len()).unwrap_or(true) {
80                let mut stream_completed = false;
81                let elem = if let Some(stream) = this.stream.as_mut().as_pin_mut() {
82                    match stream.poll_next(cx) {
83                        Poll::Ready(Some(elem)) => {
84                            made_progress_this_iter = true;
85                            Some(elem)
86                        }
87                        Poll::Ready(None) => {
88                            stream_completed = true;
89                            None
90                        }
91                        Poll::Pending => None,
92                    }
93                } else {
94                    None
95                };
96                if stream_completed {
97                    this.stream.set(None);
98                }
99                if let Some(elem) = elem {
100                    this.futures.push((this.f)(elem));
101                }
102            }
103
104            match this.futures.poll_next_unpin(cx) {
105                Poll::Ready(Some(())) => made_progress_this_iter = true,
106                Poll::Ready(None) => {
107                    if this.stream.is_none() {
108                        return Poll::Ready(());
109                    }
110                }
111                Poll::Pending => {}
112            }
113
114            if !made_progress_this_iter {
115                return Poll::Pending;
116            }
117        }
118    }
119}
futures_util/stream/stream/for_each_concurrent.rs

futures_util/stream/stream/
for_each_concurrent.rs
