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
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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

#![cfg(feature = "http-client")]

use pipeflow::source::Source;

use pipeflow::source::http_client::{HttpClientConfig, HttpClientSource};
use std::time::{Duration, Instant};
use tokio::sync::broadcast;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};

#[tokio::test]
async fn test_interval_scheduling() {
    let mock_server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/data"))
        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
            "status": "ok"
        })))
        .mount(&mock_server)
        .await;

    let config = HttpClientConfig {
        url: format!("{}/data", mock_server.uri()),
        interval: Duration::from_millis(200),
        ..Default::default()
    };

    let source = HttpClientSource::new("test_interval", config).unwrap();
    let (tx, mut rx) = broadcast::channel(100);
    let (shutdown_tx, _) = broadcast::channel(1);
    let shutdown_rx = shutdown_tx.subscribe();

    let source_handle = tokio::spawn(async move {
        let sender = pipeflow::source::MessageSender::new(tx, None);
        source.run(sender, shutdown_rx).await
    });

    let mut timestamps = Vec::new();
    let _start_time = Instant::now();

    // Collect 4 messages (approx 0s, 0.2s, 0.4s, 0.6s)
    // First message is immediate
    for _ in 0..4 {
        if tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .is_ok()
        {
            timestamps.push(Instant::now());
        } else {
            break;
        }
    }

    let _ = shutdown_tx.send(());
    let _ = source_handle.await;

    assert_eq!(timestamps.len(), 4, "Should receive 4 messages");

    // Check intervals
    // 0 -> 1: ~200ms
    // 1 -> 2: ~200ms
    // 2 -> 3: ~200ms
    // We allow some jitter (e.g. ±50ms)
    for i in 0..timestamps.len() - 1 {
        let diff = timestamps[i + 1].duration_since(timestamps[i]);
        println!("Interval {}: {:?}", i, diff);
        assert!(
            diff >= Duration::from_millis(150),
            "Interval too short: {:?}",
            diff
        );
        // On slow CI/local matching, it might be slightly longer, but shouldn't be too long
        assert!(
            diff <= Duration::from_millis(400),
            "Interval too long: {:?}",
            diff
        );
    }
}

#[tokio::test]
async fn test_cron_scheduling() {
    let mock_server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/cron"))
        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
            "status": "ok"
        })))
        .mount(&mock_server)
        .await;

    // Schedule: every second
    let config = HttpClientConfig {
        url: format!("{}/cron", mock_server.uri()),
        schedule: Some("* * * * * *".to_string()),
        ..Default::default()
    };

    let source = HttpClientSource::new("test_cron", config).unwrap();
    let (tx, mut rx) = broadcast::channel(100);
    let (shutdown_tx, _) = broadcast::channel(1);
    let shutdown_rx = shutdown_tx.subscribe();

    let source_handle = tokio::spawn(async move {
        let sender = pipeflow::source::MessageSender::new(tx, None);
        source.run(sender, shutdown_rx).await
    });

    let mut timestamps = Vec::new();

    // Collect 3 messages
    // Cron runs at sec boundary.
    // E.g. 10:00:01, 10:00:02, 10:00:03...
    // The first run depends on when we start.
    // Wait for up to 4 seconds to catch at least 3 ticks.
    for _ in 0..3 {
        if tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .is_ok()
        {
            timestamps.push(Instant::now());
        }
    }

    let _ = shutdown_tx.send(());
    let _ = source_handle.await;

    assert_eq!(timestamps.len(), 3, "Should receive 3 messages from cron");

    for i in 0..timestamps.len() - 1 {
        let diff = timestamps[i + 1].duration_since(timestamps[i]);
        println!("Cron Interval {}: {:?}", i, diff);
        // Cron guarantees approx 1 second
        assert!(
            diff >= Duration::from_millis(800),
            "Cron interval too short: {:?}",
            diff
        );
        assert!(
            diff <= Duration::from_millis(1500),
            "Cron interval too long: {:?}",
            diff
        );
    }
}