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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

pub mod filter;
pub mod flatmap;
pub mod map;
pub mod map_in_place;

pub use filter::Filter;
pub use flatmap::FlatMap;
pub use map::Map;
pub use map_in_place::MapInPlace;

#[cfg(test)]
mod tests {
    use crate::prelude::*;

    fn wait(millis: u64) {
        std::thread::sleep(std::time::Duration::from_millis(millis));
    }

    #[test]
    fn map_test() {
        let app = (0..10)
            .to_stream(|conf| {
                conf.set_arcon_time(ArconTime::Process);
            })
            .map(|x| x + 10)
            .debug()
            .builder()
            .build();

        check_map_result(app);
    }

    #[test]
    fn map_in_place_test() {
        let app = (0..10)
            .to_stream(|conf| {
                conf.set_arcon_time(ArconTime::Process);
            })
            .map_in_place(|x| *x += 10)
            .debug()
            .builder()
            .build();

        check_map_result(app);
    }

    // helper to check common result between Map/MapInPlace
    fn check_map_result(mut app: Application) {
        app.run();
        wait(1000);

        let debug_node = app.get_debug_node::<i32>().unwrap();

        debug_node.on_definition(|cd| {
            let sum: i32 = cd.data.iter().map(|elem| elem.data).sum();
            assert_eq!(sum, 145);
        });
    }

    #[test]
    fn filter_test() {
        let mut app = (0..10i32)
            .to_stream(|conf| {
                conf.set_arcon_time(ArconTime::Process);
            })
            .filter(|x| *x < 5)
            .debug()
            .builder()
            .build();

        app.run();
        wait(1000);

        let debug_node = app.get_debug_node::<i32>().unwrap();

        debug_node.on_definition(|cd| {
            assert_eq!(cd.data.len(), 5);
        });
    }

    #[test]
    fn flatmap_test() {
        let mut builder = (0..5i32)
            .to_stream(|conf| {
                conf.set_arcon_time(ArconTime::Process);
            })
            .flat_map(|x| (0..x))
            .debug()
            .builder();

        let mut app = builder.build();

        app.run();
        wait(1000);

        let debug_node = app.get_debug_node::<i32>().unwrap();

        debug_node.on_definition(|cd| {
            assert_eq!(cd.data.len(), 10);
            let sum: i32 = cd.data.iter().map(|elem| elem.data).sum();
            assert_eq!(sum, 10);
        });
    }
}