futures_util/stream/
for_each.rs

1use futures_core::{Async, Future, IntoFuture, Poll, Stream};
2use futures_core::task;
3
4/// A stream combinator which executes a unit closure over each item on a
5/// stream.
6///
7/// This structure is returned by the `Stream::for_each` method.
8#[derive(Debug)]
9#[must_use = "streams do nothing unless polled"]
10pub struct ForEach<S, U, F> where U: IntoFuture {
11    stream: Option<S>,
12    f: F,
13    fut: Option<U::Future>,
14}
15
16pub fn new<S, U, F>(s: S, f: F) -> ForEach<S, U, F>
17    where S: Stream,
18          F: FnMut(S::Item) -> U,
19          U: IntoFuture<Item = (), Error = S::Error>,
20{
21    ForEach {
22        stream: Some(s),
23        f: f,
24        fut: None,
25    }
26}
27
28impl<S, U, F> Future for ForEach<S, U, F>
29    where S: Stream,
30          F: FnMut(S::Item) -> U,
31          U: IntoFuture<Item= (), Error = S::Error>,
32{
33    type Item = S;
34    type Error = S::Error;
35
36    fn poll(&mut self, cx: &mut task::Context) -> Poll<S, S::Error> {
37        loop {
38            if let Some(mut fut) = self.fut.take() {
39                if fut.poll(cx)?.is_pending() {
40                    self.fut = Some(fut);
41                    return Ok(Async::Pending);
42                }
43            }
44
45            match self.stream {
46                Some(ref mut stream) => {
47                    match try_ready!(stream.poll_next(cx)) {
48                        Some(e) => self.fut = Some((self.f)(e).into_future()),
49                        None => break,
50                    }
51                }
52                None => panic!("poll after a ForEach was done"),
53            }
54        }
55        Ok(Async::Ready(self.stream.take().unwrap()))
56    }
57}
futures_util/stream/for_each.rs

futures_util/stream/
for_each.rs
