dataflow-rs 3.12.0

A lightweight rules engine for building IFTTT-style automation and data processing pipelines in Rust. Define rules with JSONLogic conditions, execute actions, and chain workflows.
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
111
112
113
114
115
116
117
118
119
120

//! # Sequential map-chain throughput
//!
//! Ten `map` mappings in one task, each reading the value the previous one
//! wrote, across `data` / `temp_data` / `metadata`. Measures per-message cost
//! when every mapping resolves and writes.
//!
//! Run with: `cargo run --example map_performance_test --release`

use dataflow_rs::engine::{Engine, Message, Workflow};
use serde_json::json;
use std::time::Instant;

#[tokio::main]
async fn main() {
    // Create a workflow with multiple sequential mappings
    let workflows_json = json!([
        {
            "id": "perf_test",
            "name": "Performance Test",
            "priority": 1,
            "condition": true,
            "tasks": [
                {
                    // Without this the mappings below read nothing: the eval
                    // context is `{data, metadata, temp_data}` and `payload`
                    // is not part of it, so `{"var": "data.source"}` would
                    // resolve to missing and every mapping would be a no-op.
                    "id": "load",
                    "name": "Load payload into data",
                    "function": {
                        "name": "parse_json",
                        "input": { "source": "payload", "target": "input" }
                    }
                },
                {
                    "id": "multi_map",
                    "name": "Multiple Sequential Mappings",
                    "function": {
                        "name": "map",
                        "input": {
                            "mappings": [
                                {"path": "data.field1", "logic": {"var": "data.input.source"}},
                                {"path": "data.field2", "logic": {"var": "data.field1"}},
                                {"path": "data.field3", "logic": {"var": "data.field2"}},
                                {"path": "data.field4", "logic": {"var": "data.field3"}},
                                {"path": "data.field5", "logic": {"var": "data.field4"}},
                                {"path": "temp_data.temp1", "logic": {"var": "data.field5"}},
                                {"path": "temp_data.temp2", "logic": {"var": "temp_data.temp1"}},
                                {"path": "data.result", "logic": {"var": "temp_data.temp2"}},
                                {"path": "metadata.meta1", "logic": {"var": "data.result"}},
                                {"path": "metadata.meta2", "logic": {"var": "metadata.meta1"}}
                            ]
                        }
                    }
                }
            ]
        }
    ]);

    let workflows: Vec<Workflow> = workflows_json
        .as_array()
        .unwrap()
        .iter()
        .map(|w| serde_json::from_value(w.clone()).unwrap())
        .collect();

    let engine = Engine::builder().with_workflows(workflows).build().unwrap();

    // Warmup
    println!("Warming up...");
    for _ in 0..1000 {
        let mut message = Message::from_value(&json!({
            "source": "test_value"
        }));
        engine.process_message(&mut message).await.unwrap();

        // Fail loudly rather than reporting a throughput number for ten
        // mappings that resolved to nothing. `data.result` is the end of the
        // chain, so it only holds the seed value if all ten fired.
        assert_eq!(
            message.data()["result"],
            json!("test_value").into(),
            "the mapping chain did not resolve — this benchmark would be measuring no-ops"
        );
    }

    // Benchmark
    let iterations = 100_000;
    println!(
        "Running {} iterations with 10 sequential mappings each...",
        iterations
    );

    let start = Instant::now();
    for _ in 0..iterations {
        let mut message = Message::from_value(&json!({
            "source": "test_value"
        }));
        engine.process_message(&mut message).await.unwrap();
    }
    let duration = start.elapsed();

    let throughput = iterations as f64 / duration.as_secs_f64();
    let avg_time = duration.as_micros() as f64 / iterations as f64;

    println!("\nPerformance Results:");
    println!("====================");
    println!("Total iterations: {}", iterations);
    println!("Total time: {:?}", duration);
    println!("Throughput: {:.0} messages/second", throughput);
    println!("Average time per message: {:.2} μs", avg_time);
    println!(
        "Total mappings processed: {} (10 per message)",
        iterations * 10
    );
    println!(
        "Mapping throughput: {:.0} mappings/second",
        throughput * 10.0
    );
}