timely 0.2.0

A low-latency data-parallel dataflow system in Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

use std::collections::HashMap;
use std::hash::Hash;

use Data;
use dataflow::channels::pact::Pipeline;
use dataflow::{Stream, Scope};
use dataflow::operators::unary::Unary;

pub trait Queue {
    fn queue(&self) -> Self;
}

impl<G: Scope, D: Data> Queue for Stream<G, D>
where G::Timestamp: Hash {
    fn queue(&self) -> Stream<G, D> {
        let mut elements = HashMap::new();
        self.unary_notify(Pipeline, "Queue", vec![], move |input, output, notificator| {
            while let Some((time, data)) = input.next() {
                let set = elements.entry(*time).or_insert_with(Vec::new);
                for datum in data.drain(..) { set.push(datum); }
                notificator.notify_at(time);
            }

            for (time, _count) in notificator {
                if let Some(mut data) = elements.remove(&time) {
                    output.session(&time).give_iterator(data.drain(..));
                }
            }
        })
    }
}