Struct StreamPipelineBuilder 

pub struct StreamPipelineBuilder { /* private fields */ }

Builder for constructing stream processing pipelines

Provides a fluent API for configuring all aspects of a stream processing pipeline including windows, watermarks, state backends, and operators.

Example

use processor::pipeline::StreamPipelineBuilder;
use processor::config::{WindowConfig, StateBackend};
use std::time::Duration;

let pipeline = StreamPipelineBuilder::new()
    .with_name("my-pipeline")
    .with_description("Processes user events")
    .with_parallelism(8)
    .with_window(WindowConfig::tumbling(60_000))
    .with_watermark_delay(Duration::from_secs(10))
    .with_state_backend(StateBackend::Memory)
    .with_buffer_size(50_000)
    .with_tag("production")
    .with_tag("user-events")
    .build()?;

Implementations

impl StreamPipelineBuilder

pub fn new() -> Self

Create a new pipeline builder with default configuration

pub fn with_name<S: Into<String>>(self, name: S) -> Self

Set the pipeline name

Example

let builder = StreamPipelineBuilder::new()
    .with_name("user-event-processor");

pub fn with_description<S: Into<String>>(self, description: S) -> Self

Set the pipeline description

pub fn with_tag<S: Into<String>>(self, tag: S) -> Self

Add a tag to the pipeline

pub fn with_tags<I, S>(self, tags: I) -> Self

where I: IntoIterator<Item = S>, S: Into<String>,

Add multiple tags to the pipeline

pub fn with_parallelism(self, parallelism: usize) -> Self

Set the number of parallel processing workers

Example

let builder = StreamPipelineBuilder::new()
    .with_parallelism(16);

pub fn with_buffer_size(self, buffer_size: usize) -> Self

Set the event buffer size

pub fn with_metrics(self, enabled: bool) -> Self

Enable or disable metrics collection

pub fn with_window(self, window: WindowConfig) -> Self

Set the window configuration

Example

let builder = StreamPipelineBuilder::new()
    .with_window(WindowConfig::tumbling(60_000));

pub fn with_tumbling_window(self, size_ms: u64) -> Self

Set a tumbling window with the given size

Arguments

• size_ms - Window size in milliseconds

Example

let builder = StreamPipelineBuilder::new()
    .with_tumbling_window(60_000); // 1 minute windows

pub fn with_sliding_window(self, size_ms: u64, slide_ms: u64) -> Self

Set a sliding window with the given size and slide

Arguments

• size_ms - Window size in milliseconds
• slide_ms - Slide interval in milliseconds

Example

let builder = StreamPipelineBuilder::new()
    .with_sliding_window(300_000, 60_000); // 5 min window, 1 min slide

pub fn with_session_window(self, gap_ms: u64) -> Self

Set a session window with the given gap

Arguments

• gap_ms - Session gap in milliseconds

Example

let builder = StreamPipelineBuilder::new()
    .with_session_window(300_000); // 5 minute session gap

pub fn with_allowed_lateness(self, lateness_ms: u64) -> Self

Set allowed lateness for late events

pub fn with_drop_late_events(self, drop: bool) -> Self

Set whether to drop late events

pub fn with_watermark_config(self, watermark: WatermarkConfig) -> Self

Set the watermark configuration

pub fn with_watermark_delay(self, delay: Duration) -> Self

Set the watermark delay (max out-of-orderness)

Example

let builder = StreamPipelineBuilder::new()
    .with_watermark_delay(Duration::from_secs(10));

pub fn with_idle_timeout(self, timeout: Duration) -> Self

Set the idle source timeout

pub fn with_watermark_interval(self, interval: Duration) -> Self

Set the watermark update interval

pub fn with_state_backend(self, backend: StateBackend) -> Self

Set the state backend

Example

let builder = StreamPipelineBuilder::new()
    .with_state_backend(StateBackend::Memory);

pub fn with_state_path<S: Into<PathBuf>>(self, path: S) -> Self

Set the state storage path (for file-based backends)

pub fn with_checkpoint_interval(self, interval: Duration) -> Self

Set the checkpoint interval

pub fn with_incremental_checkpoints(self, enabled: bool) -> Self

Enable or disable incremental checkpoints

pub fn with_checkpoint_retention(self, count: usize) -> Self

Set the number of checkpoints to retain

pub fn with_state_ttl(self, ttl: Duration) -> Self

Set state TTL (time-to-live)

pub fn with_max_state_size(self, size_bytes: u64) -> Self

Set maximum state size for memory backends

pub fn with_state_compression(self, enabled: bool) -> Self

Enable state compression

pub fn with_percentiles(self, percentiles: Vec<u8>) -> Self

Set aggregation percentiles to compute

pub fn with_histogram_buckets(self, buckets: Vec<f64>) -> Self

Set histogram buckets

pub fn with_min_samples(self, min: usize) -> Self

Set minimum samples required for aggregation

pub fn with_max_samples(self, max: usize) -> Self

Set maximum samples to keep in memory

pub fn with_outlier_removal(self, threshold: f64) -> Self

Enable outlier detection and removal

pub fn with_deduplication(self, enabled: bool, ttl: Duration) -> Self

Enable event deduplication

Example

let builder = StreamPipelineBuilder::new()
    .with_deduplication(true, Duration::from_secs(3600));

pub fn with_deduplication_config(self, config: DeduplicationConfig) -> Self

Set the deduplication configuration

Example

let dedup_config = DeduplicationConfig::new()
    .enabled()
    .with_ttl(Duration::from_secs(7200))
    .with_strategy(DeduplicationStrategy::EventId);

let builder = StreamPipelineBuilder::new()
    .with_deduplication_config(dedup_config);

pub fn with_deduplication_ttl(self, ttl: Duration) -> Self

Set the deduplication TTL

Example

let builder = StreamPipelineBuilder::new()
    .with_deduplication_ttl(Duration::from_secs(3600));

pub fn with_deduplication_strategy( self, strategy: DeduplicationStrategy, ) -> Self

Set the deduplication strategy

Example

let builder = StreamPipelineBuilder::new()
    .with_deduplication_strategy(DeduplicationStrategy::ContentHash);

pub fn with_deduplication_redis_prefix<S: Into<String>>(self, prefix: S) -> Self

Set the Redis key prefix for deduplication

Example

let builder = StreamPipelineBuilder::new()
    .with_deduplication_redis_prefix("myapp:dedup");

pub fn with_normalization(self, enabled: bool, interval: Duration) -> Self

Enable time-series normalization

Example

let builder = StreamPipelineBuilder::new()
    .with_normalization(true, Duration::from_secs(5));

pub fn with_normalization_config(self, config: NormalizationConfig) -> Self

Set the normalization configuration

Example

let norm_config = NormalizationConfig::new()
    .enabled()
    .with_interval(Duration::from_secs(5))
    .with_fill_strategy(FillStrategy::LinearInterpolation)
    .with_alignment(AlignmentStrategy::Start);

let builder = StreamPipelineBuilder::new()
    .with_normalization_config(norm_config);

pub fn with_normalization_interval(self, interval: Duration) -> Self

Set the normalization interval

Example

let builder = StreamPipelineBuilder::new()
    .with_normalization_interval(Duration::from_secs(5));

pub fn with_fill_strategy(self, strategy: FillStrategy) -> Self

Set the fill strategy for normalization

Example

let builder = StreamPipelineBuilder::new()
    .with_fill_strategy(FillStrategy::LinearInterpolation);

pub fn with_alignment_strategy(self, strategy: AlignmentStrategy) -> Self

Set the alignment strategy for normalization

Example

let builder = StreamPipelineBuilder::new()
    .with_alignment_strategy(AlignmentStrategy::Start);

pub fn with_normalization_buffer_size(self, size: usize) -> Self

Set the normalization buffer size

Example

let builder = StreamPipelineBuilder::new()
    .with_normalization_buffer_size(2000);

pub fn with_interpolation_max_gap(self, gap: Duration) -> Self

Set the interpolation max gap for normalization

Example

let builder = StreamPipelineBuilder::new()
    .with_interpolation_max_gap(Duration::from_secs(10));

pub fn with_normalization_outlier_detection(self, threshold: f64) -> Self

Enable outlier detection in normalization

Example

let builder = StreamPipelineBuilder::new()
    .with_normalization_outlier_detection(3.0);

pub fn build(self) -> Result<StreamPipeline>

Build the pipeline with the configured settings

Errors

Returns an error if the configuration is invalid.

Example

let pipeline = StreamPipelineBuilder::new()
    .with_name("my-pipeline")
    .with_parallelism(4)
    .build()?;

Trait Implementations

impl Default for StreamPipelineBuilder

fn default() -> Self

Returns the “default value” for a type.