While-loop in timely dataflow

timely-while.rs

extern crate timely;

use timely::*;
use timely::dataflow::operators::*;
use std::iter::once;

fn main() {
    execute_from_args(std::env::args(), |worker| {
        let index = worker.index();
        worker.dataflow(|scope| {
            let (helper, state) = scope.loop_variable(std::u64::MAX, 1);
            let state = once(0).to_stream(scope)
                .concat(&state)
                .map(|i| i + 1)
                .filter(|i| *i < 5);

            state
                .inspect(move |i| println!("{}: state {}", index, i));
            state
                .connect_loop(helper);
        });
    }).unwrap();
}