Naive streams

naive-streams.rs

#[derive(Clone)]
struct Repeat {
    current: u32,
    end: u32,
}

impl Repeat {
    fn new(current: u32, end: u32) -> Self {
        Repeat {current, end}
    }
}

impl Future for Repeat {
    type Error = ();
    type Item = Option<(u32, Self)>;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        if self.current < self.end {
            Ok(Async::Ready(Some((self.current, Repeat::new(self.current + 1, self.end)))))
        } else {
            Ok(Async::Ready(None))
        }
    }
}

fn main() {
    let pool = CpuPool::new(1);

    let f = loop_fn(Repeat::new(10, 20), |mut stream| match stream.poll() {
        Ok(Async::Ready(Some((item, next)))) => {
            println!("Item: {}", item);
            Ok(Loop::Continue(next))
        }
        Ok(Async::Ready(None)) => Ok(Loop::Break(stream)),
        Ok(Async::NotReady) => Ok(Loop::Continue(stream)),
        Err(err) => Err(err),
    });

    pool.spawn(f).wait().unwrap();
}