Module stream_processor 

High-level stream processor for event processing with windowing and aggregation

This module provides the StreamProcessor which coordinates:

• Event ingestion and time extraction
• Window assignment and management
• Watermark generation and propagation
• Aggregation computation per window
• Result emission

Architecture

┌─────────────┐
│   Events    │
└──────┬──────┘
       │
       ▼
┌─────────────────────┐
│  Time Extraction    │
└──────┬──────────────┘
       │
       ▼
┌─────────────────────┐
│ Window Assignment   │ ◄─── WindowManager
└──────┬──────────────┘
       │
       ▼
┌─────────────────────┐
│   Aggregation       │ ◄─── Aggregators
└──────┬──────────────┘
       │
       ▼
┌─────────────────────┐
│ Watermark Check     │
└──────┬──────────────┘
       │
       ▼
┌─────────────────────┐
│  Trigger & Emit     │
└──────┬──────────────┘
       │
       ▼
┌─────────────────────┐
│    Results          │
└─────────────────────┘

Example

use processor::{
    StreamProcessor, StreamProcessorBuilder,
    TumblingWindowAssigner, OnWatermarkTrigger,
};
use processor::aggregation::CompositeAggregator;
use chrono::Duration;

// Build a stream processor for latency metrics
let processor = StreamProcessorBuilder::new()
    .with_tumbling_window(Duration::minutes(5))
    .with_watermark_trigger()
    .with_aggregation(|_window_id| {
        let mut agg = CompositeAggregator::new();
        agg.add_count("request_count")
           .add_avg("avg_latency")
           .add_p50("p50_latency")
           .add_p95("p95_latency")
           .add_p99("p99_latency")
           .add_max("max_latency");
        agg
    })
    .build();

// Process events
// processor.process_event(event).await?;

Structs

ProcessorStats

Statistics for the stream processor

StreamProcessor

Stream processor for windowed aggregations

StreamProcessorBuilder

Builder for creating stream processors

StreamProcessorConfig

Configuration for the stream processor

WindowResult

Result emitted by the stream processor