oxigdal-streaming 0.1.4

Real-time data processing and streaming pipelines for OxiGDAL
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

//! Pipeline builder for constructing streaming pipelines.

use super::executor::PipelineExecutor;
use super::stage::PipelineStage;
use crate::error::{Result, StreamingError};
use std::sync::Arc;

/// Configuration for a pipeline.
#[derive(Debug, Clone)]
pub struct PipelineConfig {
    /// Maximum number of concurrent stages
    pub max_concurrent_stages: usize,

    /// Buffer size between stages
    pub buffer_size: usize,

    /// Enable metrics collection
    pub enable_metrics: bool,

    /// Enable stage profiling
    pub enable_profiling: bool,
}

impl Default for PipelineConfig {
    fn default() -> Self {
        Self {
            max_concurrent_stages: num_cpus::get(),
            buffer_size: 100,
            enable_metrics: true,
            enable_profiling: false,
        }
    }
}

/// Builder for constructing streaming pipelines.
pub struct PipelineBuilder {
    config: PipelineConfig,
    stages: Vec<Arc<dyn PipelineStage>>,
}

impl PipelineBuilder {
    /// Create a new pipeline builder.
    pub fn new() -> Self {
        Self {
            config: PipelineConfig::default(),
            stages: Vec::new(),
        }
    }

    /// Create a new pipeline builder with config.
    pub fn with_config(config: PipelineConfig) -> Self {
        Self {
            config,
            stages: Vec::new(),
        }
    }

    /// Add a stage to the pipeline.
    pub fn add_stage(mut self, stage: Arc<dyn PipelineStage>) -> Self {
        self.stages.push(stage);
        self
    }

    /// Set the maximum number of concurrent stages.
    pub fn max_concurrent_stages(mut self, max: usize) -> Self {
        self.config.max_concurrent_stages = max;
        self
    }

    /// Set the buffer size between stages.
    pub fn buffer_size(mut self, size: usize) -> Self {
        self.config.buffer_size = size;
        self
    }

    /// Enable metrics collection.
    pub fn enable_metrics(mut self, enable: bool) -> Self {
        self.config.enable_metrics = enable;
        self
    }

    /// Enable stage profiling.
    pub fn enable_profiling(mut self, enable: bool) -> Self {
        self.config.enable_profiling = enable;
        self
    }

    /// Build the pipeline executor.
    pub async fn build(self) -> Result<PipelineExecutor> {
        if self.stages.is_empty() {
            return Err(StreamingError::ConfigError(
                "Pipeline must have at least one stage".to_string()
            ));
        }

        PipelineExecutor::new(self.config, self.stages).await
    }
}

impl Default for PipelineBuilder {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::pipeline::stage::TransformStage;

    #[test]
    fn test_pipeline_builder() {
        let builder = PipelineBuilder::new()
            .max_concurrent_stages(4)
            .buffer_size(50)
            .enable_metrics(true);

        assert_eq!(builder.config.max_concurrent_stages, 4);
        assert_eq!(builder.config.buffer_size, 50);
        assert_eq!(builder.config.enable_metrics, true);
    }

    #[tokio::test]
    async fn test_empty_pipeline() {
        let builder = PipelineBuilder::new();
        let result = builder.build().await;
        assert!(result.is_err());
    }
}