reorder operator example

reorder.rs

use std::collections::HashMap;
use timely::dataflow::channels::pact::Pipeline;
use timely::dataflow::operators::generic::Operator;
use timely::dataflow::operators::unordered_input::UnorderedInput;
use timely::dataflow::operators::Inspect;

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {
        let (mut input, cap) = worker.dataflow::<usize, _, _>(|scope| {
            let mut queues = HashMap::new();

            let (input, stream) = scope.new_unordered_input();

            stream
                .inspect_batch(move |epoch, data| {
                    for d in data {
                        println!("{}@{:?}", epoch, d);
                    }
                })
                .unary_notify(
                    Pipeline,
                    "reorder",
                    None,
                    move |input, output, notificator| {
                        input.for_each(|cap, data| {
                            queues
                                .entry(cap.time().clone())
                                .or_insert(Vec::new())
                                .push(data.replace(Vec::new()));

                            println!("{:?}", queues);
                            notificator.notify_at(cap.retain());
                        });

                        // this is never called with I/O block
                        notificator.for_each(|cap, _cnt, _not| {
                            let mut session = output.session(&cap);
                            println!("done with time {:?}", &cap);

                            if let Some(list) = queues.remove(cap.time()) {
                                for mut vector in list.into_iter() {
                                    session.give_vec(&mut vector);
                                }
                            }
                        });
                    },
                )
                .inspect_batch(move |epoch, data| {
                    for d in data {
                        println!("{}@{:?}", epoch, d);
                    }
                });
            input
        });

        input.session(cap.delayed(&0)).give('D');
        input.session(cap.delayed(&2)).give('B');
        input.session(cap.delayed(&1)).give('A');
        input.session(cap.delayed(&2)).give('b');
        input.session(cap.delayed(&3)).give('C');
        input.session(cap.delayed(&3)).give('c');
        input.session(cap.delayed(&1)).give('a');

        worker.step(); // this runs to unary_notificator

        use std::io::stdin;
        stdin().read_line(&mut String::new()).unwrap();
    })
    .unwrap();
}