1 use {Async, Future, IntoFuture, Poll};
2 use stream::Stream;
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"]
10 pub struct ForEach<S, F, U> where U: IntoFuture {
11 stream: S,
12 f: F,
13 fut: Option<U::Future>,
14 }
15
new<S, F, U>(s: S, f: F) -> ForEach<S, F, U> where S: Stream, F: FnMut(S::Item) -> U, U: IntoFuture<Item = (), Error = S::Error>,16 pub fn new<S, F, U>(s: S, f: F) -> ForEach<S, F, U>
17 where S: Stream,
18 F: FnMut(S::Item) -> U,
19 U: IntoFuture<Item = (), Error = S::Error>,
20 {
21 ForEach {
22 stream: s,
23 f: f,
24 fut: None,
25 }
26 }
27
28 impl<S, F, U> Future for ForEach<S, F, U>
29 where S: Stream,
30 F: FnMut(S::Item) -> U,
31 U: IntoFuture<Item= (), Error = S::Error>,
32 {
33 type Item = ();
34 type Error = S::Error;
35
poll(&mut self) -> Poll<(), S::Error>36 fn poll(&mut self) -> Poll<(), S::Error> {
37 loop {
38 if let Some(mut fut) = self.fut.take() {
39 if fut.poll()?.is_not_ready() {
40 self.fut = Some(fut);
41 return Ok(Async::NotReady);
42 }
43 }
44
45 match try_ready!(self.stream.poll()) {
46 Some(e) => self.fut = Some((self.f)(e).into_future()),
47 None => return Ok(Async::Ready(())),
48 }
49 }
50 }
51 }
52