hydroflow 0.10.0

Hydro's low-level dataflow runtime and IR
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

use hydroflow::util::collect_ready;
use hydroflow::{assert_graphvis_snapshots, hydroflow_syntax};
use multiplatform_test::multiplatform_test;

#[multiplatform_test(test, wasm, env_tracing)]
pub fn test_basic() {
    let (single_tx, single_rx) = hydroflow::util::unbounded_channel::<()>();
    let (egress_tx, mut egress_rx) = hydroflow::util::unbounded_channel();

    let mut df = hydroflow_syntax! {
        join = cross_singleton();
        source_iter([1, 2, 3]) -> persist::<'static>() -> [input]join;
        source_stream(single_rx) -> [single]join;

        join -> for_each(|x| egress_tx.send(x).unwrap());
    };
    assert_graphvis_snapshots!(df);

    df.run_available();
    let out: Vec<_> = collect_ready(&mut egress_rx);
    assert_eq!(out, []);

    single_tx.send(()).unwrap();
    df.run_available();

    let out: Vec<_> = collect_ready(&mut egress_rx);
    assert_eq!(out, vec![(1, ()), (2, ()), (3, ())]);
}

#[multiplatform_test(test, wasm, env_tracing)]
pub fn test_union_defer_tick() {
    let (cross_tx, cross_rx) = hydroflow::util::unbounded_channel::<i32>();
    let (egress_tx, mut egress_rx) = hydroflow::util::unbounded_channel();

    let mut df = hydroflow_syntax! {
        teed_in = source_stream(cross_rx) -> sort() -> tee();
        teed_in -> [input]join;

        deferred_stream -> defer_tick_lazy() -> [0]unioned_stream;

        persisted_stream = source_iter([0]) -> persist::<'static>();
        persisted_stream -> [1]unioned_stream;

        unioned_stream = union();
        unioned_stream -> [single]join;

        join = cross_singleton() -> tee();

        join -> for_each(|x| egress_tx.send(x).unwrap());

        folded_thing = join -> fold(|| 0, |_, _| {});

        teed_in -> [input]joined_folded;
        folded_thing -> [single]joined_folded;
        joined_folded = cross_singleton();
        deferred_stream = joined_folded -> fold(|| 0, |_, _| {}) -> flat_map(|_| []);
    };
    assert_graphvis_snapshots!(df);

    df.run_available();
    let out: Vec<_> = collect_ready(&mut egress_rx);
    assert_eq!(out, vec![]);

    cross_tx.send(1).unwrap();
    cross_tx.send(2).unwrap();
    cross_tx.send(3).unwrap();
    df.run_available();

    let out: Vec<_> = collect_ready(&mut egress_rx);
    assert_eq!(out, vec![(1, 0), (2, 0), (3, 0)]);
}