llm_optimizer_processor/pipeline/
builder.rs

1//! Stream pipeline builder for fluent API configuration
2//!
3//! This module provides a builder pattern for constructing stream processing pipelines
4//! with a clean, composable API.
5
6use crate::config::{ProcessorConfig, StateBackend, WindowConfig, WatermarkConfig};
7use crate::error::{ProcessorError, Result};
8use crate::pipeline::executor::StreamExecutor;
9use crate::pipeline::operator::StreamOperator;
10use serde::{Deserialize, Serialize};
11use std::sync::Arc;
12use std::time::Duration;
13
14/// Configuration for a stream processing pipeline
15#[derive(Debug, Clone, Serialize, Deserialize)]
16pub struct PipelineConfig {
17    /// Pipeline name for identification
18    pub name: String,
19
20    /// Processor configuration
21    pub processor: ProcessorConfig,
22
23    /// Description of the pipeline
24    pub description: Option<String>,
25
26    /// Tags for categorization
27    pub tags: Vec<String>,
28}
29
30impl Default for PipelineConfig {
31    fn default() -> Self {
32        Self {
33            name: "unnamed-pipeline".to_string(),
34            processor: ProcessorConfig::default(),
35            description: None,
36            tags: Vec::new(),
37        }
38    }
39}
40
41impl PipelineConfig {
42    /// Validate the pipeline configuration
43    pub fn validate(&self) -> Result<()> {
44        if self.name.is_empty() {
45            return Err(ProcessorError::Configuration {
46                source: "pipeline name cannot be empty".into(),
47            });
48        }
49
50        self.processor.validate()?;
51        Ok(())
52    }
53}
54
55/// Builder for constructing stream processing pipelines
56///
57/// Provides a fluent API for configuring all aspects of a stream processing pipeline
58/// including windows, watermarks, state backends, and operators.
59///
60/// # Example
61///
62/// ```rust
63/// use processor::pipeline::StreamPipelineBuilder;
64/// use processor::config::{WindowConfig, StateBackend};
65/// use std::time::Duration;
66///
67/// # fn example() -> anyhow::Result<()> {
68/// let pipeline = StreamPipelineBuilder::new()
69///     .with_name("my-pipeline")
70///     .with_description("Processes user events")
71///     .with_parallelism(8)
72///     .with_window(WindowConfig::tumbling(60_000))
73///     .with_watermark_delay(Duration::from_secs(10))
74///     .with_state_backend(StateBackend::Memory)
75///     .with_buffer_size(50_000)
76///     .with_tag("production")
77///     .with_tag("user-events")
78///     .build()?;
79/// # Ok(())
80/// # }
81/// ```
82pub struct StreamPipelineBuilder {
83    config: PipelineConfig,
84}
85
86impl Default for StreamPipelineBuilder {
87    fn default() -> Self {
88        Self::new()
89    }
90}
91
92impl StreamPipelineBuilder {
93    /// Create a new pipeline builder with default configuration
94    pub fn new() -> Self {
95        Self {
96            config: PipelineConfig::default(),
97        }
98    }
99
100    /// Set the pipeline name
101    ///
102    /// # Example
103    ///
104    /// ```rust
105    /// # use processor::pipeline::StreamPipelineBuilder;
106    /// let builder = StreamPipelineBuilder::new()
107    ///     .with_name("user-event-processor");
108    /// ```
109    pub fn with_name<S: Into<String>>(mut self, name: S) -> Self {
110        self.config.name = name.into();
111        self
112    }
113
114    /// Set the pipeline description
115    pub fn with_description<S: Into<String>>(mut self, description: S) -> Self {
116        self.config.description = Some(description.into());
117        self
118    }
119
120    /// Add a tag to the pipeline
121    pub fn with_tag<S: Into<String>>(mut self, tag: S) -> Self {
122        self.config.tags.push(tag.into());
123        self
124    }
125
126    /// Add multiple tags to the pipeline
127    pub fn with_tags<I, S>(mut self, tags: I) -> Self
128    where
129        I: IntoIterator<Item = S>,
130        S: Into<String>,
131    {
132        self.config.tags.extend(tags.into_iter().map(|s| s.into()));
133        self
134    }
135
136    /// Set the number of parallel processing workers
137    ///
138    /// # Example
139    ///
140    /// ```rust
141    /// # use processor::pipeline::StreamPipelineBuilder;
142    /// let builder = StreamPipelineBuilder::new()
143    ///     .with_parallelism(16);
144    /// ```
145    pub fn with_parallelism(mut self, parallelism: usize) -> Self {
146        self.config.processor.parallelism = parallelism;
147        self
148    }
149
150    /// Set the event buffer size
151    pub fn with_buffer_size(mut self, buffer_size: usize) -> Self {
152        self.config.processor.buffer_size = buffer_size;
153        self
154    }
155
156    /// Enable or disable metrics collection
157    pub fn with_metrics(mut self, enabled: bool) -> Self {
158        self.config.processor.metrics_enabled = enabled;
159        self
160    }
161
162    /// Set the window configuration
163    ///
164    /// # Example
165    ///
166    /// ```rust
167    /// # use processor::pipeline::StreamPipelineBuilder;
168    /// # use processor::config::WindowConfig;
169    /// let builder = StreamPipelineBuilder::new()
170    ///     .with_window(WindowConfig::tumbling(60_000));
171    /// ```
172    pub fn with_window(mut self, window: WindowConfig) -> Self {
173        self.config.processor.window = window;
174        self
175    }
176
177    /// Set a tumbling window with the given size
178    ///
179    /// # Arguments
180    /// * `size_ms` - Window size in milliseconds
181    ///
182    /// # Example
183    ///
184    /// ```rust
185    /// # use processor::pipeline::StreamPipelineBuilder;
186    /// let builder = StreamPipelineBuilder::new()
187    ///     .with_tumbling_window(60_000); // 1 minute windows
188    /// ```
189    pub fn with_tumbling_window(mut self, size_ms: u64) -> Self {
190        self.config.processor.window = WindowConfig::tumbling(size_ms);
191        self
192    }
193
194    /// Set a sliding window with the given size and slide
195    ///
196    /// # Arguments
197    /// * `size_ms` - Window size in milliseconds
198    /// * `slide_ms` - Slide interval in milliseconds
199    ///
200    /// # Example
201    ///
202    /// ```rust
203    /// # use processor::pipeline::StreamPipelineBuilder;
204    /// let builder = StreamPipelineBuilder::new()
205    ///     .with_sliding_window(300_000, 60_000); // 5 min window, 1 min slide
206    /// ```
207    pub fn with_sliding_window(mut self, size_ms: u64, slide_ms: u64) -> Self {
208        self.config.processor.window = WindowConfig::sliding(size_ms, slide_ms);
209        self
210    }
211
212    /// Set a session window with the given gap
213    ///
214    /// # Arguments
215    /// * `gap_ms` - Session gap in milliseconds
216    ///
217    /// # Example
218    ///
219    /// ```rust
220    /// # use processor::pipeline::StreamPipelineBuilder;
221    /// let builder = StreamPipelineBuilder::new()
222    ///     .with_session_window(300_000); // 5 minute session gap
223    /// ```
224    pub fn with_session_window(mut self, gap_ms: u64) -> Self {
225        self.config.processor.window = WindowConfig::session(gap_ms);
226        self
227    }
228
229    /// Set allowed lateness for late events
230    pub fn with_allowed_lateness(mut self, lateness_ms: u64) -> Self {
231        self.config.processor.window.allowed_lateness_ms = lateness_ms;
232        self
233    }
234
235    /// Set whether to drop late events
236    pub fn with_drop_late_events(mut self, drop: bool) -> Self {
237        self.config.processor.window.drop_late_events = drop;
238        self
239    }
240
241    /// Set the watermark configuration
242    pub fn with_watermark_config(mut self, watermark: WatermarkConfig) -> Self {
243        self.config.processor.watermark = watermark;
244        self
245    }
246
247    /// Set the watermark delay (max out-of-orderness)
248    ///
249    /// # Example
250    ///
251    /// ```rust
252    /// # use processor::pipeline::StreamPipelineBuilder;
253    /// # use std::time::Duration;
254    /// let builder = StreamPipelineBuilder::new()
255    ///     .with_watermark_delay(Duration::from_secs(10));
256    /// ```
257    pub fn with_watermark_delay(mut self, delay: Duration) -> Self {
258        self.config.processor.watermark.max_delay_ms = delay.as_millis() as u64;
259        self
260    }
261
262    /// Set the idle source timeout
263    pub fn with_idle_timeout(mut self, timeout: Duration) -> Self {
264        self.config.processor.watermark.idle_timeout_ms = timeout.as_millis() as u64;
265        self
266    }
267
268    /// Set the watermark update interval
269    pub fn with_watermark_interval(mut self, interval: Duration) -> Self {
270        self.config.processor.watermark.update_interval_ms = interval.as_millis() as u64;
271        self
272    }
273
274    /// Set the state backend
275    ///
276    /// # Example
277    ///
278    /// ```rust
279    /// # use processor::pipeline::StreamPipelineBuilder;
280    /// # use processor::config::StateBackend;
281    /// let builder = StreamPipelineBuilder::new()
282    ///     .with_state_backend(StateBackend::Memory);
283    /// ```
284    pub fn with_state_backend(mut self, backend: StateBackend) -> Self {
285        self.config.processor.state.backend = backend;
286        self
287    }
288
289    /// Set the state storage path (for file-based backends)
290    pub fn with_state_path<S: Into<std::path::PathBuf>>(mut self, path: S) -> Self {
291        self.config.processor.state.storage_path = Some(path.into());
292        self
293    }
294
295    /// Set the checkpoint interval
296    pub fn with_checkpoint_interval(mut self, interval: Duration) -> Self {
297        self.config.processor.state.checkpoint_interval_ms = interval.as_millis() as u64;
298        self
299    }
300
301    /// Enable or disable incremental checkpoints
302    pub fn with_incremental_checkpoints(mut self, enabled: bool) -> Self {
303        self.config.processor.state.incremental_checkpoints = enabled;
304        self
305    }
306
307    /// Set the number of checkpoints to retain
308    pub fn with_checkpoint_retention(mut self, count: usize) -> Self {
309        self.config.processor.state.checkpoint_retention = count;
310        self
311    }
312
313    /// Set state TTL (time-to-live)
314    pub fn with_state_ttl(mut self, ttl: Duration) -> Self {
315        self.config.processor.state.ttl_ms = Some(ttl.as_millis() as u64);
316        self
317    }
318
319    /// Set maximum state size for memory backends
320    pub fn with_max_state_size(mut self, size_bytes: u64) -> Self {
321        self.config.processor.state.max_size_bytes = Some(size_bytes);
322        self
323    }
324
325    /// Enable state compression
326    pub fn with_state_compression(mut self, enabled: bool) -> Self {
327        self.config.processor.state.compression_enabled = enabled;
328        self
329    }
330
331    /// Set aggregation percentiles to compute
332    pub fn with_percentiles(mut self, percentiles: Vec<u8>) -> Self {
333        self.config.processor.aggregation.percentiles = percentiles;
334        self
335    }
336
337    /// Set histogram buckets
338    pub fn with_histogram_buckets(mut self, buckets: Vec<f64>) -> Self {
339        self.config.processor.aggregation.histogram_buckets = Some(buckets);
340        self
341    }
342
343    /// Set minimum samples required for aggregation
344    pub fn with_min_samples(mut self, min: usize) -> Self {
345        self.config.processor.aggregation.min_samples = min;
346        self
347    }
348
349    /// Set maximum samples to keep in memory
350    pub fn with_max_samples(mut self, max: usize) -> Self {
351        self.config.processor.aggregation.max_samples = max;
352        self
353    }
354
355    /// Enable outlier detection and removal
356    pub fn with_outlier_removal(mut self, threshold: f64) -> Self {
357        self.config.processor.aggregation.remove_outliers = true;
358        self.config.processor.aggregation.outlier_threshold = threshold;
359        self
360    }
361
362    /// Enable event deduplication
363    ///
364    /// # Example
365    ///
366    /// ```rust
367    /// # use processor::pipeline::StreamPipelineBuilder;
368    /// # use std::time::Duration;
369    /// let builder = StreamPipelineBuilder::new()
370    ///     .with_deduplication(true, Duration::from_secs(3600));
371    /// ```
372    pub fn with_deduplication(mut self, enabled: bool, ttl: Duration) -> Self {
373        self.config.processor.deduplication.enabled = enabled;
374        self.config.processor.deduplication.ttl = ttl;
375        self
376    }
377
378    /// Set the deduplication configuration
379    ///
380    /// # Example
381    ///
382    /// ```rust
383    /// # use processor::pipeline::StreamPipelineBuilder;
384    /// # use processor::config::{DeduplicationConfig, DeduplicationStrategy};
385    /// # use std::time::Duration;
386    /// let dedup_config = DeduplicationConfig::new()
387    ///     .enabled()
388    ///     .with_ttl(Duration::from_secs(7200))
389    ///     .with_strategy(DeduplicationStrategy::EventId);
390    ///
391    /// let builder = StreamPipelineBuilder::new()
392    ///     .with_deduplication_config(dedup_config);
393    /// ```
394    pub fn with_deduplication_config(mut self, config: crate::config::DeduplicationConfig) -> Self {
395        self.config.processor.deduplication = config;
396        self
397    }
398
399    /// Set the deduplication TTL
400    ///
401    /// # Example
402    ///
403    /// ```rust
404    /// # use processor::pipeline::StreamPipelineBuilder;
405    /// # use std::time::Duration;
406    /// let builder = StreamPipelineBuilder::new()
407    ///     .with_deduplication_ttl(Duration::from_secs(3600));
408    /// ```
409    pub fn with_deduplication_ttl(mut self, ttl: Duration) -> Self {
410        self.config.processor.deduplication.ttl = ttl;
411        self
412    }
413
414    /// Set the deduplication strategy
415    ///
416    /// # Example
417    ///
418    /// ```rust
419    /// # use processor::pipeline::StreamPipelineBuilder;
420    /// # use processor::config::DeduplicationStrategy;
421    /// let builder = StreamPipelineBuilder::new()
422    ///     .with_deduplication_strategy(DeduplicationStrategy::ContentHash);
423    /// ```
424    pub fn with_deduplication_strategy(mut self, strategy: crate::config::DeduplicationStrategy) -> Self {
425        self.config.processor.deduplication.strategy = strategy;
426        self
427    }
428
429    /// Set the Redis key prefix for deduplication
430    ///
431    /// # Example
432    ///
433    /// ```rust
434    /// # use processor::pipeline::StreamPipelineBuilder;
435    /// let builder = StreamPipelineBuilder::new()
436    ///     .with_deduplication_redis_prefix("myapp:dedup");
437    /// ```
438    pub fn with_deduplication_redis_prefix<S: Into<String>>(mut self, prefix: S) -> Self {
439        self.config.processor.deduplication.redis_key_prefix = prefix.into();
440        self
441    }
442
443    /// Enable time-series normalization
444    ///
445    /// # Example
446    ///
447    /// ```rust
448    /// # use processor::pipeline::StreamPipelineBuilder;
449    /// # use std::time::Duration;
450    /// let builder = StreamPipelineBuilder::new()
451    ///     .with_normalization(true, Duration::from_secs(5));
452    /// ```
453    pub fn with_normalization(mut self, enabled: bool, interval: Duration) -> Self {
454        self.config.processor.normalization.enabled = enabled;
455        self.config.processor.normalization.interval = interval;
456        self
457    }
458
459    /// Set the normalization configuration
460    ///
461    /// # Example
462    ///
463    /// ```rust
464    /// # use processor::pipeline::StreamPipelineBuilder;
465    /// # use processor::config::{NormalizationConfig, FillStrategy, AlignmentStrategy};
466    /// # use std::time::Duration;
467    /// let norm_config = NormalizationConfig::new()
468    ///     .enabled()
469    ///     .with_interval(Duration::from_secs(5))
470    ///     .with_fill_strategy(FillStrategy::LinearInterpolation)
471    ///     .with_alignment(AlignmentStrategy::Start);
472    ///
473    /// let builder = StreamPipelineBuilder::new()
474    ///     .with_normalization_config(norm_config);
475    /// ```
476    pub fn with_normalization_config(mut self, config: crate::config::NormalizationConfig) -> Self {
477        self.config.processor.normalization = config;
478        self
479    }
480
481    /// Set the normalization interval
482    ///
483    /// # Example
484    ///
485    /// ```rust
486    /// # use processor::pipeline::StreamPipelineBuilder;
487    /// # use std::time::Duration;
488    /// let builder = StreamPipelineBuilder::new()
489    ///     .with_normalization_interval(Duration::from_secs(5));
490    /// ```
491    pub fn with_normalization_interval(mut self, interval: Duration) -> Self {
492        self.config.processor.normalization.interval = interval;
493        self
494    }
495
496    /// Set the fill strategy for normalization
497    ///
498    /// # Example
499    ///
500    /// ```rust
501    /// # use processor::pipeline::StreamPipelineBuilder;
502    /// # use processor::config::FillStrategy;
503    /// let builder = StreamPipelineBuilder::new()
504    ///     .with_fill_strategy(FillStrategy::LinearInterpolation);
505    /// ```
506    pub fn with_fill_strategy(mut self, strategy: crate::config::FillStrategy) -> Self {
507        self.config.processor.normalization.fill_strategy = strategy;
508        self
509    }
510
511    /// Set the alignment strategy for normalization
512    ///
513    /// # Example
514    ///
515    /// ```rust
516    /// # use processor::pipeline::StreamPipelineBuilder;
517    /// # use processor::config::AlignmentStrategy;
518    /// let builder = StreamPipelineBuilder::new()
519    ///     .with_alignment_strategy(AlignmentStrategy::Start);
520    /// ```
521    pub fn with_alignment_strategy(mut self, strategy: crate::config::AlignmentStrategy) -> Self {
522        self.config.processor.normalization.alignment = strategy;
523        self
524    }
525
526    /// Set the normalization buffer size
527    ///
528    /// # Example
529    ///
530    /// ```rust
531    /// # use processor::pipeline::StreamPipelineBuilder;
532    /// let builder = StreamPipelineBuilder::new()
533    ///     .with_normalization_buffer_size(2000);
534    /// ```
535    pub fn with_normalization_buffer_size(mut self, size: usize) -> Self {
536        self.config.processor.normalization.buffer_size = size;
537        self
538    }
539
540    /// Set the interpolation max gap for normalization
541    ///
542    /// # Example
543    ///
544    /// ```rust
545    /// # use processor::pipeline::StreamPipelineBuilder;
546    /// # use std::time::Duration;
547    /// let builder = StreamPipelineBuilder::new()
548    ///     .with_interpolation_max_gap(Duration::from_secs(10));
549    /// ```
550    pub fn with_interpolation_max_gap(mut self, gap: Duration) -> Self {
551        self.config.processor.normalization.interpolation_max_gap = gap;
552        self
553    }
554
555    /// Enable outlier detection in normalization
556    ///
557    /// # Example
558    ///
559    /// ```rust
560    /// # use processor::pipeline::StreamPipelineBuilder;
561    /// let builder = StreamPipelineBuilder::new()
562    ///     .with_normalization_outlier_detection(3.0);
563    /// ```
564    pub fn with_normalization_outlier_detection(mut self, threshold: f64) -> Self {
565        self.config.processor.normalization.drop_outliers = true;
566        self.config.processor.normalization.outlier_threshold = threshold;
567        self
568    }
569
570    /// Build the pipeline with the configured settings
571    ///
572    /// # Errors
573    ///
574    /// Returns an error if the configuration is invalid.
575    ///
576    /// # Example
577    ///
578    /// ```rust
579    /// # use processor::pipeline::StreamPipelineBuilder;
580    /// # fn example() -> anyhow::Result<()> {
581    /// let pipeline = StreamPipelineBuilder::new()
582    ///     .with_name("my-pipeline")
583    ///     .with_parallelism(4)
584    ///     .build()?;
585    /// # Ok(())
586    /// # }
587    /// ```
588    pub fn build(self) -> Result<StreamPipeline> {
589        self.config.validate()?;
590
591        Ok(StreamPipeline {
592            config: Arc::new(self.config),
593            operators: Vec::new(),
594        })
595    }
596}
597
598/// A configured stream processing pipeline
599///
600/// This represents a fully configured pipeline that can be used to create executors
601/// and add operators.
602#[derive(Clone)]
603pub struct StreamPipeline {
604    config: Arc<PipelineConfig>,
605    operators: Vec<Arc<dyn StreamOperator>>,
606}
607
608impl StreamPipeline {
609    /// Get the pipeline configuration
610    pub fn config(&self) -> &PipelineConfig {
611        &self.config
612    }
613
614    /// Get the pipeline name
615    pub fn name(&self) -> &str {
616        &self.config.name
617    }
618
619    /// Get the pipeline description
620    pub fn description(&self) -> Option<&str> {
621        self.config.description.as_deref()
622    }
623
624    /// Get the pipeline tags
625    pub fn tags(&self) -> &[String] {
626        &self.config.tags
627    }
628
629    /// Add an operator to the pipeline
630    ///
631    /// # Example
632    ///
633    /// ```rust
634    /// # use processor::pipeline::{StreamPipelineBuilder, StreamOperator};
635    /// # use processor::pipeline::operator::MapOperator;
636    /// # fn example() -> anyhow::Result<()> {
637    /// let mut pipeline = StreamPipelineBuilder::new().build()?;
638    ///
639    /// // Add a map operator
640    /// // let operator = MapOperator::new(|x| x * 2);
641    /// // pipeline.add_operator(operator);
642    /// # Ok(())
643    /// # }
644    /// ```
645    pub fn add_operator<O: StreamOperator + 'static>(&mut self, operator: O) {
646        self.operators.push(Arc::new(operator));
647    }
648
649    /// Get all operators in the pipeline
650    pub fn operators(&self) -> &[Arc<dyn StreamOperator>] {
651        &self.operators
652    }
653
654    /// Create an executor for this pipeline
655    ///
656    /// # Example
657    ///
658    /// ```rust
659    /// # use processor::pipeline::StreamPipelineBuilder;
660    /// # async fn example() -> anyhow::Result<()> {
661    /// let pipeline = StreamPipelineBuilder::new()
662    ///     .with_name("test-pipeline")
663    ///     .build()?;
664    ///
665    /// let executor = pipeline.create_executor();
666    /// // executor.run().await?;
667    /// # Ok(())
668    /// # }
669    /// ```
670    pub fn create_executor<T>(
671        &self,
672    ) -> StreamExecutor<T>
673    where
674        T: Clone + Send + Sync + 'static + Serialize + for<'de> Deserialize<'de>,
675    {
676        StreamExecutor::new(self.clone())
677    }
678}
679
680impl std::fmt::Debug for StreamPipeline {
681    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
682        f.debug_struct("StreamPipeline")
683            .field("config", &self.config)
684            .field("operator_count", &self.operators.len())
685            .finish()
686    }
687}
688
689#[cfg(test)]
690mod tests {
691    use super::*;
692    use crate::config::WindowType;
693
694    #[test]
695    fn test_builder_default() {
696        let pipeline = StreamPipelineBuilder::new().build().unwrap();
697        assert_eq!(pipeline.name(), "unnamed-pipeline");
698    }
699
700    #[test]
701    fn test_builder_with_name() {
702        let pipeline = StreamPipelineBuilder::new()
703            .with_name("test-pipeline")
704            .build()
705            .unwrap();
706
707        assert_eq!(pipeline.name(), "test-pipeline");
708    }
709
710    #[test]
711    fn test_builder_with_description() {
712        let pipeline = StreamPipelineBuilder::new()
713            .with_name("test")
714            .with_description("A test pipeline")
715            .build()
716            .unwrap();
717
718        assert_eq!(pipeline.description(), Some("A test pipeline"));
719    }
720
721    #[test]
722    fn test_builder_with_tags() {
723        let pipeline = StreamPipelineBuilder::new()
724            .with_name("test")
725            .with_tag("production")
726            .with_tag("metrics")
727            .build()
728            .unwrap();
729
730        assert_eq!(pipeline.tags(), &["production", "metrics"]);
731    }
732
733    #[test]
734    fn test_builder_with_multiple_tags() {
735        let pipeline = StreamPipelineBuilder::new()
736            .with_name("test")
737            .with_tags(vec!["tag1", "tag2", "tag3"])
738            .build()
739            .unwrap();
740
741        assert_eq!(pipeline.tags().len(), 3);
742    }
743
744    #[test]
745    fn test_builder_with_parallelism() {
746        let pipeline = StreamPipelineBuilder::new()
747            .with_name("test")
748            .with_parallelism(16)
749            .build()
750            .unwrap();
751
752        assert_eq!(pipeline.config().processor.parallelism, 16);
753    }
754
755    #[test]
756    fn test_builder_with_buffer_size() {
757        let pipeline = StreamPipelineBuilder::new()
758            .with_name("test")
759            .with_buffer_size(50_000)
760            .build()
761            .unwrap();
762
763        assert_eq!(pipeline.config().processor.buffer_size, 50_000);
764    }
765
766    #[test]
767    fn test_builder_with_tumbling_window() {
768        let pipeline = StreamPipelineBuilder::new()
769            .with_name("test")
770            .with_tumbling_window(60_000)
771            .build()
772            .unwrap();
773
774        assert_eq!(
775            pipeline.config().processor.window.window_type,
776            WindowType::Tumbling
777        );
778        assert_eq!(pipeline.config().processor.window.size_ms, Some(60_000));
779    }
780
781    #[test]
782    fn test_builder_with_sliding_window() {
783        let pipeline = StreamPipelineBuilder::new()
784            .with_name("test")
785            .with_sliding_window(300_000, 60_000)
786            .build()
787            .unwrap();
788
789        assert_eq!(
790            pipeline.config().processor.window.window_type,
791            WindowType::Sliding
792        );
793        assert_eq!(pipeline.config().processor.window.size_ms, Some(300_000));
794        assert_eq!(pipeline.config().processor.window.slide_ms, Some(60_000));
795    }
796
797    #[test]
798    fn test_builder_with_session_window() {
799        let pipeline = StreamPipelineBuilder::new()
800            .with_name("test")
801            .with_session_window(300_000)
802            .build()
803            .unwrap();
804
805        assert_eq!(
806            pipeline.config().processor.window.window_type,
807            WindowType::Session
808        );
809        assert_eq!(pipeline.config().processor.window.gap_ms, Some(300_000));
810    }
811
812    #[test]
813    fn test_builder_with_watermark_delay() {
814        let pipeline = StreamPipelineBuilder::new()
815            .with_name("test")
816            .with_watermark_delay(Duration::from_secs(10))
817            .build()
818            .unwrap();
819
820        assert_eq!(pipeline.config().processor.watermark.max_delay_ms, 10_000);
821    }
822
823    #[test]
824    fn test_builder_with_state_backend() {
825        let pipeline = StreamPipelineBuilder::new()
826            .with_name("test")
827            .with_state_backend(StateBackend::FileSystem)
828            .with_state_path("/tmp/state")
829            .build()
830            .unwrap();
831
832        assert_eq!(
833            pipeline.config().processor.state.backend,
834            StateBackend::FileSystem
835        );
836        assert!(pipeline.config().processor.state.storage_path.is_some());
837    }
838
839    #[test]
840    fn test_builder_with_checkpoint_config() {
841        let pipeline = StreamPipelineBuilder::new()
842            .with_name("test")
843            .with_checkpoint_interval(Duration::from_secs(30))
844            .with_checkpoint_retention(5)
845            .with_incremental_checkpoints(false)
846            .build()
847            .unwrap();
848
849        assert_eq!(
850            pipeline.config().processor.state.checkpoint_interval_ms,
851            30_000
852        );
853        assert_eq!(pipeline.config().processor.state.checkpoint_retention, 5);
854        assert!(!pipeline.config().processor.state.incremental_checkpoints);
855    }
856
857    #[test]
858    fn test_builder_with_state_ttl() {
859        let pipeline = StreamPipelineBuilder::new()
860            .with_name("test")
861            .with_state_ttl(Duration::from_secs(3600))
862            .build()
863            .unwrap();
864
865        assert_eq!(pipeline.config().processor.state.ttl_ms, Some(3_600_000));
866    }
867
868    #[test]
869    fn test_builder_with_aggregation_config() {
870        let pipeline = StreamPipelineBuilder::new()
871            .with_name("test")
872            .with_percentiles(vec![50, 90, 95, 99])
873            .with_min_samples(10)
874            .with_max_samples(200_000)
875            .build()
876            .unwrap();
877
878        assert_eq!(
879            pipeline.config().processor.aggregation.percentiles,
880            vec![50, 90, 95, 99]
881        );
882        assert_eq!(pipeline.config().processor.aggregation.min_samples, 10);
883        assert_eq!(pipeline.config().processor.aggregation.max_samples, 200_000);
884    }
885
886    #[test]
887    fn test_builder_with_outlier_removal() {
888        let pipeline = StreamPipelineBuilder::new()
889            .with_name("test")
890            .with_outlier_removal(3.0)
891            .build()
892            .unwrap();
893
894        assert!(pipeline.config().processor.aggregation.remove_outliers);
895        assert_eq!(
896            pipeline.config().processor.aggregation.outlier_threshold,
897            3.0
898        );
899    }
900
901    #[test]
902    fn test_builder_fluent_api() {
903        let pipeline = StreamPipelineBuilder::new()
904            .with_name("complex-pipeline")
905            .with_description("A complex stream processing pipeline")
906            .with_tags(vec!["production", "metrics", "aggregation"])
907            .with_parallelism(8)
908            .with_buffer_size(100_000)
909            .with_tumbling_window(60_000)
910            .with_watermark_delay(Duration::from_secs(5))
911            .with_state_backend(StateBackend::Memory)
912            .with_percentiles(vec![50, 95, 99])
913            .build()
914            .unwrap();
915
916        assert_eq!(pipeline.name(), "complex-pipeline");
917        assert_eq!(pipeline.config().processor.parallelism, 8);
918        assert_eq!(pipeline.config().processor.buffer_size, 100_000);
919    }
920
921    #[test]
922    fn test_pipeline_validation_fails_on_empty_name() {
923        let result = StreamPipelineBuilder::new()
924            .with_name("")
925            .build();
926
927        assert!(result.is_err());
928    }
929
930    #[test]
931    fn test_pipeline_create_executor() {
932        let pipeline = StreamPipelineBuilder::new()
933            .with_name("test")
934            .build()
935            .unwrap();
936
937        let _executor: StreamExecutor<i32> = pipeline.create_executor();
938    }
939
940    #[test]
941    fn test_builder_with_deduplication() {
942        let pipeline = StreamPipelineBuilder::new()
943            .with_name("test")
944            .with_deduplication(true, Duration::from_secs(3600))
945            .build()
946            .unwrap();
947
948        assert!(pipeline.config().processor.deduplication.enabled);
949        assert_eq!(
950            pipeline.config().processor.deduplication.ttl,
951            Duration::from_secs(3600)
952        );
953    }
954
955    #[test]
956    fn test_builder_with_deduplication_config() {
957        use crate::config::{DeduplicationConfig, DeduplicationStrategy};
958
959        let dedup_config = DeduplicationConfig::new()
960            .enabled()
961            .with_ttl(Duration::from_secs(7200))
962            .with_strategy(DeduplicationStrategy::EventId)
963            .with_redis_key_prefix("custom_dedup");
964
965        let pipeline = StreamPipelineBuilder::new()
966            .with_name("test")
967            .with_deduplication_config(dedup_config)
968            .build()
969            .unwrap();
970
971        let config = &pipeline.config().processor.deduplication;
972        assert!(config.enabled);
973        assert_eq!(config.ttl, Duration::from_secs(7200));
974        assert_eq!(config.strategy, DeduplicationStrategy::EventId);
975        assert_eq!(config.redis_key_prefix, "custom_dedup");
976    }
977
978    #[test]
979    fn test_builder_with_deduplication_ttl() {
980        let pipeline = StreamPipelineBuilder::new()
981            .with_name("test")
982            .with_deduplication_ttl(Duration::from_secs(1800))
983            .build()
984            .unwrap();
985
986        assert_eq!(
987            pipeline.config().processor.deduplication.ttl,
988            Duration::from_secs(1800)
989        );
990    }
991
992    #[test]
993    fn test_builder_with_deduplication_strategy() {
994        use crate::config::DeduplicationStrategy;
995
996        let pipeline = StreamPipelineBuilder::new()
997            .with_name("test")
998            .with_deduplication_strategy(DeduplicationStrategy::CompositeKey)
999            .build()
1000            .unwrap();
1001
1002        assert_eq!(
1003            pipeline.config().processor.deduplication.strategy,
1004            DeduplicationStrategy::CompositeKey
1005        );
1006    }
1007
1008    #[test]
1009    fn test_builder_with_deduplication_redis_prefix() {
1010        let pipeline = StreamPipelineBuilder::new()
1011            .with_name("test")
1012            .with_deduplication_redis_prefix("myapp:dedup")
1013            .build()
1014            .unwrap();
1015
1016        assert_eq!(
1017            pipeline.config().processor.deduplication.redis_key_prefix,
1018            "myapp:dedup"
1019        );
1020    }
1021
1022    #[test]
1023    fn test_builder_deduplication_fluent_api() {
1024        use crate::config::DeduplicationStrategy;
1025
1026        let pipeline = StreamPipelineBuilder::new()
1027            .with_name("dedup-pipeline")
1028            .with_deduplication(true, Duration::from_secs(3600))
1029            .with_deduplication_strategy(DeduplicationStrategy::ContentHash)
1030            .with_deduplication_redis_prefix("prod:dedup")
1031            .build()
1032            .unwrap();
1033
1034        let config = &pipeline.config().processor.deduplication;
1035        assert!(config.enabled);
1036        assert_eq!(config.ttl, Duration::from_secs(3600));
1037        assert_eq!(config.strategy, DeduplicationStrategy::ContentHash);
1038        assert_eq!(config.redis_key_prefix, "prod:dedup");
1039    }
1040
1041    #[test]
1042    fn test_builder_with_normalization() {
1043        let pipeline = StreamPipelineBuilder::new()
1044            .with_name("test")
1045            .with_normalization(true, Duration::from_secs(5))
1046            .build()
1047            .unwrap();
1048
1049        assert!(pipeline.config().processor.normalization.enabled);
1050        assert_eq!(
1051            pipeline.config().processor.normalization.interval,
1052            Duration::from_secs(5)
1053        );
1054    }
1055
1056    #[test]
1057    fn test_builder_with_normalization_config() {
1058        use crate::config::{NormalizationConfig, FillStrategy, AlignmentStrategy};
1059
1060        let norm_config = NormalizationConfig::new()
1061            .enabled()
1062            .with_interval(Duration::from_secs(10))
1063            .with_fill_strategy(FillStrategy::LinearInterpolation)
1064            .with_alignment(AlignmentStrategy::End);
1065
1066        let pipeline = StreamPipelineBuilder::new()
1067            .with_name("test")
1068            .with_normalization_config(norm_config)
1069            .build()
1070            .unwrap();
1071
1072        let config = &pipeline.config().processor.normalization;
1073        assert!(config.enabled);
1074        assert_eq!(config.interval, Duration::from_secs(10));
1075        assert_eq!(config.fill_strategy, FillStrategy::LinearInterpolation);
1076        assert_eq!(config.alignment, AlignmentStrategy::End);
1077    }
1078
1079    #[test]
1080    fn test_builder_with_normalization_interval() {
1081        let pipeline = StreamPipelineBuilder::new()
1082            .with_name("test")
1083            .with_normalization_interval(Duration::from_secs(15))
1084            .build()
1085            .unwrap();
1086
1087        assert_eq!(
1088            pipeline.config().processor.normalization.interval,
1089            Duration::from_secs(15)
1090        );
1091    }
1092
1093    #[test]
1094    fn test_builder_with_fill_strategy() {
1095        use crate::config::FillStrategy;
1096
1097        let pipeline = StreamPipelineBuilder::new()
1098            .with_name("test")
1099            .with_fill_strategy(FillStrategy::BackwardFill)
1100            .build()
1101            .unwrap();
1102
1103        assert_eq!(
1104            pipeline.config().processor.normalization.fill_strategy,
1105            FillStrategy::BackwardFill
1106        );
1107    }
1108
1109    #[test]
1110    fn test_builder_with_alignment_strategy() {
1111        use crate::config::AlignmentStrategy;
1112
1113        let pipeline = StreamPipelineBuilder::new()
1114            .with_name("test")
1115            .with_alignment_strategy(AlignmentStrategy::Center)
1116            .build()
1117            .unwrap();
1118
1119        assert_eq!(
1120            pipeline.config().processor.normalization.alignment,
1121            AlignmentStrategy::Center
1122        );
1123    }
1124
1125    #[test]
1126    fn test_builder_with_normalization_buffer_size() {
1127        let pipeline = StreamPipelineBuilder::new()
1128            .with_name("test")
1129            .with_normalization_buffer_size(5000)
1130            .build()
1131            .unwrap();
1132
1133        assert_eq!(
1134            pipeline.config().processor.normalization.buffer_size,
1135            5000
1136        );
1137    }
1138
1139    #[test]
1140    fn test_builder_with_interpolation_max_gap() {
1141        let pipeline = StreamPipelineBuilder::new()
1142            .with_name("test")
1143            .with_interpolation_max_gap(Duration::from_secs(20))
1144            .build()
1145            .unwrap();
1146
1147        assert_eq!(
1148            pipeline.config().processor.normalization.interpolation_max_gap,
1149            Duration::from_secs(20)
1150        );
1151    }
1152
1153    #[test]
1154    fn test_builder_with_normalization_outlier_detection() {
1155        let pipeline = StreamPipelineBuilder::new()
1156            .with_name("test")
1157            .with_normalization_outlier_detection(2.5)
1158            .build()
1159            .unwrap();
1160
1161        assert!(pipeline.config().processor.normalization.drop_outliers);
1162        assert_eq!(
1163            pipeline.config().processor.normalization.outlier_threshold,
1164            2.5
1165        );
1166    }
1167
1168    #[test]
1169    fn test_builder_normalization_fluent_api() {
1170        use crate::config::{FillStrategy, AlignmentStrategy};
1171
1172        let pipeline = StreamPipelineBuilder::new()
1173            .with_name("norm-pipeline")
1174            .with_normalization(true, Duration::from_secs(5))
1175            .with_fill_strategy(FillStrategy::LinearInterpolation)
1176            .with_alignment_strategy(AlignmentStrategy::Start)
1177            .with_normalization_buffer_size(2000)
1178            .with_interpolation_max_gap(Duration::from_secs(15))
1179            .with_normalization_outlier_detection(3.0)
1180            .build()
1181            .unwrap();
1182
1183        let config = &pipeline.config().processor.normalization;
1184        assert!(config.enabled);
1185        assert_eq!(config.interval, Duration::from_secs(5));
1186        assert_eq!(config.fill_strategy, FillStrategy::LinearInterpolation);
1187        assert_eq!(config.alignment, AlignmentStrategy::Start);
1188        assert_eq!(config.buffer_size, 2000);
1189        assert_eq!(config.interpolation_max_gap, Duration::from_secs(15));
1190        assert!(config.drop_outliers);
1191        assert_eq!(config.outlier_threshold, 3.0);
1192    }
1193}
llm_optimizer_processor/pipeline/builder.rs

llm_optimizer_processor/pipeline/
builder.rs
