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

//! Transform pipeline for multi-step processing
//!
//! A TransformPipeline chains multiple steps together, executing them sequentially.
//! Messages flow through each step; if any step filters out the message, processing stops.

use async_trait::async_trait;

use crate::common::message::Message;
use crate::error::Result;
use crate::transform::Transform;
use crate::transform::step::Step;

/// Transform pipeline that chains multiple steps together.
pub struct TransformPipeline {
    id: String,
    /// Steps are executed sequentially. Stateful steps should use interior mutability
    /// (e.g. `Mutex`) if they need to maintain shared state.
    steps: Vec<Box<dyn Step>>,
}

impl TransformPipeline {
    /// Create a new transform pipeline with the given steps.
    pub fn new(id: impl Into<String>, steps: Vec<Box<dyn Step>>) -> Self {
        Self {
            id: id.into(),
            steps,
        }
    }
}

#[async_trait]
impl Transform for TransformPipeline {
    fn id(&self) -> &str {
        &self.id
    }

    async fn process(&self, msg: Message) -> Result<Vec<Message>> {
        let mut current = msg;

        // processing start log
        tracing::debug!(
            transform_id = %self.id,
            "Pipeline processing message"
        );

        for step in &self.steps {
            let step_type = step.step_type();

            // step start log
            tracing::trace!(
                transform_id = %self.id,
                step = %step_type,
                "Executing step"
            );

            match step.process(current)? {
                Some(output) => {
                    current = output;
                }
                None => {
                    tracing::debug!(
                        transform_id = %self.id,
                        step = %step_type,
                        "Message filtered out by step"
                    );
                    return Ok(vec![]);
                }
            }
        }

        tracing::debug!(
            transform_id = %self.id,
            "Pipeline finished processing"
        );

        Ok(vec![current])
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    /// A simple pass-through step for testing
    struct PassThroughStep;

    impl Step for PassThroughStep {
        fn step_type(&self) -> &'static str {
            "passthrough"
        }

        fn process(&self, msg: Message) -> Result<Option<Message>> {
            Ok(Some(msg))
        }
    }

    /// A step that filters out messages with payload containing {"drop": true}
    struct DropIfTrueStep;

    impl Step for DropIfTrueStep {
        fn step_type(&self) -> &'static str {
            "drop_if_true"
        }

        fn process(&self, msg: Message) -> Result<Option<Message>> {
            if msg.payload.get("drop") == Some(&serde_json::json!(true)) {
                return Ok(None);
            }
            Ok(Some(msg))
        }
    }

    #[tokio::test]
    async fn test_pipeline_passes_through_all_steps() {
        let pipeline = TransformPipeline::new(
            "test",
            vec![Box::new(PassThroughStep), Box::new(PassThroughStep)],
        );

        let msg = Message::new("test", json!({"value": 42}));
        let result = pipeline.process(msg).await.unwrap();

        assert_eq!(result.len(), 1);
        assert_eq!(result[0].payload["value"], 42);
    }

    #[tokio::test]
    async fn test_pipeline_stops_on_filter() {
        let pipeline = TransformPipeline::new(
            "test",
            vec![Box::new(DropIfTrueStep), Box::new(PassThroughStep)],
        );

        // Message that should be dropped
        let msg = Message::new("test", json!({"drop": true}));
        let result = pipeline.process(msg).await.unwrap();
        assert!(result.is_empty());

        // Message that should pass
        let msg = Message::new("test", json!({"drop": false, "data": "hello"}));
        let result = pipeline.process(msg).await.unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].payload["data"], "hello");
    }
}