flowstdlib 1.0.0

The standard library of functions and flows for 'flow' programs
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
105
106
107
108
109
110

use serde_json::{json, Value};

use flowcore::errors::Result;
use flowcore::flow_output;
use flowcore::RunAgain;
use flowmacro::flow_function;

#[flow_function]
fn inner_accumulate(
    values: &Value,
    partial: &Value,
    chunk_size: &Value,
) -> Result<(Option<Value>, RunAgain)> {
    if values.is_null() {
        let partial_arr = partial.as_array().ok_or("Could not get partial")?;
        return if partial_arr.is_empty() {
            flow_output!("chunk" => Value::Null)
        } else {
            flow_output!("chunk" => Value::Array(partial_arr.clone()))
        };
    }

    let mut partial_input = partial.clone();
    let chunk_size = chunk_size.as_u64().filter(|&s| s > 0);

    let partial = partial_input
        .as_array_mut()
        .ok_or("Could not get partial")?;
    partial.push(values.clone());

    if let Some(size) = chunk_size {
        if partial.len() >= usize::try_from(size)? {
            flow_output!(
                "chunk" => Value::Array(partial.clone()),
                "partial" => Value::Array(vec![]),
                "chunk_size" => json!(size),
            )
        } else {
            flow_output!(
                "partial" => Value::Array(partial.clone()),
                "chunk_size" => json!(size),
            )
        }
    } else {
        flow_output!(
            "partial" => Value::Array(partial.clone()),
            "chunk_size" => json!(0),
        )
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod test {
    use serde_json::json;

    use super::inner_accumulate;

    #[test]
    fn accumulate_start_and_finish() {
        let value = json!(1);
        let partial = json!([]);
        let chunk_size = json!(1);

        let (result, _) =
            inner_accumulate(&value, &partial, &chunk_size).expect("_accumulate() failed");
        let output = result.expect("Could not get the Value from the output");
        assert_eq!(
            output
                .pointer("/chunk")
                .expect("Could not get the /chunk from the output"),
            &json!([1])
        );
    }

    #[test]
    fn accumulate_start_not_finish() {
        let value = json!(1);
        let partial = json!([]);
        let chunk_size = json!(2);

        let (result, _) =
            inner_accumulate(&value, &partial, &chunk_size).expect("_accumulate() failed");
        let output = result.expect("Could not get the Value from the output");
        assert_eq!(output.pointer("/chunk"), None);
        assert_eq!(
            output
                .pointer("/partial")
                .expect("Could not get the /partial from the output"),
            &json!([1])
        );
    }

    #[test]
    fn accumulate_started_then_finish() {
        let value = json!(2);
        let partial = json!([1]);
        let chunk_size = json!(2);

        let (result, _) =
            inner_accumulate(&value, &partial, &chunk_size).expect("_accumulate() failed");
        let output = result.expect("Could not get the Value from the output");
        assert_eq!(
            output
                .pointer("/chunk")
                .expect("Could not get the /chunk from the output"),
            &json!([1, 2])
        );
    }
}