arcon 0.2.1

A runtime for writing streaming applications
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
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

pub mod appender;
pub mod arrow;
pub mod incremental;

#[cfg(test)]
mod tests {
    use super::appender::*;
    use super::incremental::*;
    use crate::index::WindowIndex;
    use crate::stream::operator::window::WindowContext;
    use crate::test_utils::temp_backend;
    use arcon_state::Sled;
    use std::sync::Arc;

    #[test]
    fn sum_appender_window_test() {
        let backend = Arc::new(temp_backend::<Sled>());

        fn materializer(buffer: &[i32]) -> i32 {
            buffer.iter().sum()
        }

        let mut window = AppenderWindow::new(backend, &materializer);

        for i in 0..10 {
            let _ = window.on_element(i, WindowContext::new(0, 0));
        }

        let sum = window.result(WindowContext::new(0, 0)).unwrap();

        let expected: i32 = 45;
        assert_eq!(sum, expected);
    }

    #[test]
    fn sum_incremental_window_test() {
        let backend = Arc::new(temp_backend::<Sled>());

        fn init(i: i32) -> u64 {
            i as u64
        }
        fn aggregation(i: i32, agg: &u64) -> u64 {
            agg + i as u64
        }

        let mut window = IncrementalWindow::new(backend, &init, &aggregation);

        for i in 0..10 {
            let _ = window.on_element(i, WindowContext::new(0, 0));
        }

        for i in 0..20 {
            let _ = window.on_element(i, WindowContext::new(1, 1));
        }

        let sum_one = window.result(WindowContext::new(0, 0)).unwrap();
        assert_eq!(sum_one, 45);
        let sum_two = window.result(WindowContext::new(1, 1)).unwrap();
        assert_eq!(sum_two, 190);
    }
}