pipeflow 0.0.4

A lightweight, configuration-driven data pipeline framework
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

use pipeflow::config::Config;
use pipeflow::engine::Engine;

#[tokio::test]
async fn test_cycle_detection() {
    let yaml = r#"
pipeline:
  sources:
    - id: source1
      type: http_client
      config:
        url: "http://localhost:8080"
        interval: "1m"

  transforms:
    - id: transform1
      inputs: ["source1", "transform2"] # Cycle!
      outputs: ["transform2"]

    - id: transform2
      inputs: ["transform1"]
      outputs: ["sink1"]

  sinks:
    - id: sink1
      type: console
"#;

    let config: Config = serde_yaml::from_str(yaml).expect("Failed to parse YAML");
    let result = Engine::from_config(config);

    let err = match result {
        Ok(_) => panic!("Expected cycle detection error"),
        Err(err) => err,
    };
    assert!(err.to_string().to_lowercase().contains("cycle detected"));
}