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
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

//! E2E tests for Redis GET -> SET pipeline
//!
//! Requires a Redis server. Set REDIS_URL to enable the test.

#[cfg(feature = "redis")]
mod redis_e2e {
    use std::time::Duration;

    use anyhow::Result;
    use pipeflow::config::Config;
    use pipeflow::engine::Engine;
    use redis::AsyncCommands;
    use redis::aio::ConnectionManager;

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn test_redis_get_set_with_ttl() -> Result<()> {
        let redis_url = match std::env::var("REDIS_URL") {
            Ok(url) => url,
            Err(_) => {
                eprintln!("REDIS_URL not set; skipping redis e2e test");
                return Ok(());
            }
        };

        let input_key = format!("pipeflow:e2e:in:{}", uuid::Uuid::now_v7());
        let output_key = format!("pipeflow:e2e:out:{}", uuid::Uuid::now_v7());

        let client = redis::Client::open(redis_url.clone())?;
        let mut conn = ConnectionManager::new(client).await?;
        let _: () = conn.set(&input_key, "hello").await?;

        let yaml = format!(
            r#"
pipeline:
  sources:
    - id: redis_in
      type: redis
      config:
        url: "{redis_url}"
        key: "{input_key}"
        mode: oneshot
        parse_json: false
  transforms:
    - id: pass
      inputs: [redis_in]
      outputs: [redis_out]
  sinks:
    - id: redis_out
      type: redis
      config:
        url: "{redis_url}"
        key:
          value: "{output_key}"
        value:
          from: "$.value"
        ttl: "30s"
"#
        );

        let config = Config::from_yaml(&yaml)?;
        let mut engine = Engine::from_config(config)?;
        engine.build().await?;

        let run = engine.run_with_signal(async {
            tokio::time::sleep(Duration::from_secs(5)).await;
        });
        tokio::time::timeout(Duration::from_secs(5), run)
            .await
            .expect("engine timeout")?;

        let value: String = conn.get(&output_key).await?;
        assert_eq!(value, "hello");

        let ttl: i64 = conn.ttl(&output_key).await?;
        assert!(ttl > 0, "expected TTL to be set for key");

        let _: () = conn.del(&input_key).await?;
        let _: () = conn.del(&output_key).await?;

        Ok(())
    }
}