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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
use std::collections::HashMap;
use std::hash::Hash;
use Data;
use dataflow::channels::pact::Pipeline;
use dataflow::{Stream, Scope};
use dataflow::operators::unary::Unary;
use dataflow::channels::message::Content;
pub trait Accumulate<G: Scope, D: Data> {
fn accumulate<A: Data, F: Fn(&mut A, &mut Content<D>)+'static>(&self, default: A, logic: F) -> Stream<G, A>;
fn count(&self) -> Stream<G, usize> {
self.accumulate(0, |sum, data| *sum += data.len())
}
}
impl<G: Scope, D: Data> Accumulate<G, D> for Stream<G, D>
where G::Timestamp: Hash {
fn accumulate<A: Data, F: Fn(&mut A, &mut Content<D>)+'static>(&self, default: A, logic: F) -> Stream<G, A> {
let mut accums = HashMap::new();
self.unary_notify(Pipeline, "Accumulate", vec![], move |input, output, notificator| {
input.for_each(|time, data| {
logic(&mut accums.entry(time.time().clone()).or_insert(default.clone()), data);
notificator.notify_at(time);
});
notificator.for_each(|time,_,_| {
if let Some(accum) = accums.remove(&time) {
output.session(&time).give(accum);
}
});
})
}
}