main.rs

fn main() {
    let (results_chan, results_port) = channel();
    let (gen_chan, gen_port) = channel();
    let merge_chan = merge(results_chan);
    {
        // New scope starts here.
        let mut square_workers: VecDeque<Sender<PipelineMsg>> = vec![square(merge_chan.clone()),
                                                                     square(merge_chan)]
                                                                     .into_iter()
                                                                     .collect();
        generate(gen_chan);
        for msg in gen_port {
            let generated_num = match msg {
                PipelineMsg::Generated(num) => num,
                _ => panic!("unexpected message receiving from gen stage"),
            };
            let worker = square_workers.pop_front().unwrap();
            let _ = worker.send(msg);
            square_workers.push_back(worker);
            if generated_num == 3 {
                // When we get the "last value" we need,
                // we break ouf of the loop, which will result in
                // the gen_port being dropped, stopping the generator.
                // Note: while gen_port is created in the scope "above" this one,
                // by using it in here we've "moved" it to this scope, 
                // hence it will drop when the end of the current scope has been reached.
                break;
            }
        }
        // About to go out of scope,
        // This is reached only when the iteration above is "broken-out of".
    }
    for result in results_port {
        match result {
            PipelineMsg::Merged(_) => continue,
            _ => panic!("unexpected result"),
        }
    }
}