Real-Time Event Stream Processor

Problem Statement

Build a real-time event stream processor that consumes events from multiple sources (simulated sensors, logs, metrics), applies transformations, filters, aggregations, and routes results to different outputs. The system must handle backpressure, buffer events efficiently, perform windowed aggregations (count events per time window), and maintain high throughput without dropping data.

Use Cases

• IoT sensor data processing - Process temperature, pressure, humidity readings from thousands of sensors
• Log aggregation systems - Collect logs from distributed services, filter, aggregate error counts
• Financial market data - Process tick data, calculate moving averages, detect anomalies
• Real-time analytics - Track user events (clicks, page views), compute metrics per second
• Network monitoring - Process packet data, detect patterns, alert on anomalies
• Stream ETL pipelines - Extract events from Kafka/RabbitMQ, transform, load into databases
• Gaming telemetry - Process player actions, compute statistics, detect cheating

Why It Matters

Backpressure is Critical: Without backpressure, fast producers overwhelm slow consumers. Example: sensor sending 10,000 events/sec, processor handling 1,000/sec → 9,000 events/sec accumulate in memory → OOM crash in 10 seconds. Proper backpressure slows the producer or buffers intelligently.

Windowed Aggregations: Real-time analytics require time-based calculations. “Events per second” needs 1-second tumbling windows. “Moving average over 5 minutes” needs sliding windows. Without async streams, you’d manually manage timers and state—error-prone and complex.

Memory Efficiency: Loading all data into memory before processing fails for infinite streams. Streams process events one-at-a-time (or in small chunks), keeping memory constant regardless of stream length. 1GB/sec stream processed in 10MB memory.

Throughput: Sequential event processing at 10ms/event = 100 events/sec. Batching 100 events + parallel processing = 10,000 events/sec (100x improvement). Async streams make batching and parallelism composable.

Example performance:

Sequential processing:     100 events/sec (10ms each)
Batched (100 per batch):   10,000 events/sec
Parallel batches (10x):    100,000 events/sec

Key Concepts Explained

1. Async Streams (Stream Trait and .next().await)

Streams are the async equivalent of iterators—they yield items over time asynchronously.

The problem with synchronous iterators:

#![allow(unused)]
fn main() {
// Synchronous iterator (blocks thread)
fn process_items(items: Vec<i32>) {
    for item in items {
        // Process immediately (all items must be in memory)
        println!("{}", item);
    }
}

// Problem: All items must exist upfront
// Memory: 1M items × 4 bytes = 4MB
}

Async streams solution:

#![allow(unused)]
fn main() {
use tokio_stream::{Stream, StreamExt};

// Asynchronous stream (items arrive over time)
async fn process_stream<S>(mut stream: S)
where
    S: Stream<Item = i32> + Unpin,
{
    // Items arrive over time (e.g., from network, sensors)
    while let Some(item) = stream.next().await {
        println!("{}", item);
    }
}

// Memory: Constant (only current item in memory)
// Throughput: Processes items as they arrive (no waiting)
}

How it works:

• Stream<Item = T> trait defines async iteration
• .next().await returns Option<T> (Some(item) or None when done)
• Items can arrive from channels, timers, network, files, etc.
• Memory efficient: Process infinite streams with constant memory

Visual timeline:

Iterator:   [All items loaded] → Process → Done
            Memory: O(n)

Stream:     Item1 arrives → Process → Item2 arrives → Process → ...
            Memory: O(1) (constant)

Performance comparison:

#![allow(unused)]
fn main() {
// Load all into memory (iterator)
let items: Vec<i32> = (0..1_000_000).collect();  // 4MB memory
for item in items {
    process(item);
}

// Stream from channel (async stream)
let (tx, rx) = mpsc::channel(100);
let stream = ReceiverStream::new(rx);  // 100-item buffer = 400 bytes
stream.for_each(|item| async { process(item) }).await;

// Memory: 4MB vs 400 bytes = 10,000× less
}

When to use streams:

• Data arrives over time (network, sensors, logs)
• Infinite or very large data (can’t fit in memory)
• Real-time processing (process as data arrives)
• Backpressure needed (slow down producer if consumer falls behind)

2. Stream Combinators (map, filter, chunks_timeout)

Stream combinators transform streams declaratively without manual loops.

Manual loop approach (verbose, error-prone):

#![allow(unused)]
fn main() {
async fn process_manual(mut stream: impl Stream<Item = Event> + Unpin) {
    let mut results = Vec::new();

    while let Some(event) = stream.next().await {
        // Filter
        if event.value > 50.0 {
            // Transform
            let normalized = event.value / 100.0;
            results.push(normalized);
        }
    }

    // Problem: Manual state management, verbose, easy to introduce bugs
}
}

Combinator approach (declarative, composable):

#![allow(unused)]
fn main() {
async fn process_combinators(stream: impl Stream<Item = Event>) {
    let results: Vec<f64> = stream
        .filter(|e| e.value > 50.0)          // Keep events > 50
        .map(|e| e.value / 100.0)            // Normalize to [0, 1]
        .collect()                            // Gather into Vec
        .await;

    // Concise, clear intent, composable
}
}

Common combinators:

chunks_timeout (critical for batching):

#![allow(unused)]
fn main() {
use tokio::time::Duration;

// Problem: Process one at a time (inefficient)
stream.for_each(|item| async {
    write_to_db(vec![item]).await;  // 1 write per item
}).await;
// Throughput: 1,000 writes/sec (1ms per write)

// Solution: Batch by count OR time
let batches = stream.chunks_timeout(100, Duration::from_secs(1));
batches.for_each(|batch| async {
    write_to_db(batch).await;  // 1 write per 100 items
}).await;
// Throughput: 100,000 items/sec with same 1ms latency
// Speedup: 100× faster
}

chunks_timeout behavior:

• Yields Vec of items when:
  • Reached N items (count limit), OR
  • Timeout elapsed (time limit)
  • Whichever comes first
• Balances latency (timeout) vs throughput (batch size)

Example timeline:

chunks_timeout(3, 500ms):

Items arrive: [1]─(100ms)─[2]─(100ms)─[3]─(100ms)─[4]
Batches:      └─────────────────────────[1,2,3]  ← Count limit (3 items)

Items arrive: [5]─(600ms)─[6]
Batches:      └──────────[5,6]  ← Timeout (500ms elapsed, only 2 items)

Performance impact:

#![allow(unused)]
fn main() {
// Database writes (1ms overhead per write)
// One-at-a-time: 1,000 writes = 1,000ms
// Batched (100): 10 writes = 10ms
// Speedup: 100× faster
}

Zero-cost abstraction:

#![allow(unused)]
fn main() {
// This combinator chain:
stream
    .filter(|e| e.value > 50.0)
    .map(|e| e.value / 100.0)

// Compiles to efficient loop (zero overhead):
while let Some(e) = stream.next().await {
    if e.value > 50.0 {
        let normalized = e.value / 100.0;
        // ...
    }
}
}

3. MPSC Channels (Multi-Producer Single-Consumer)

MPSC channels enable async communication between tasks (producer → consumer).

The problem without channels:

#![allow(unused)]
fn main() {
// Shared mutable state (needs locking)
let data = Arc::new(Mutex<Vec<Event>>);

// Producer
let data_clone = Arc::clone(&data);
tokio::spawn(async move {
    let mut guard = data_clone.lock().unwrap();  // Blocks!
    guard.push(event);
});

// Consumer
let events = data.lock().unwrap();  // Blocks waiting for producer

// Problems: Locking overhead, blocking, complexity
}

MPSC channel solution:

#![allow(unused)]
fn main() {
use tokio::sync::mpsc;

// Create channel (bounded to 100 items)
let (tx, mut rx) = mpsc::channel::<Event>(100);

// Producer (send events)
tokio::spawn(async move {
    for i in 0..1000 {
        tx.send(event).await.unwrap();  // Async, non-blocking
    }
});

// Consumer (receive events)
tokio::spawn(async move {
    while let Some(event) = rx.recv().await {
        process(event);
    }
});

// Benefits: No locks, backpressure, simple API
}

Bounded vs unbounded:

Backpressure with bounded channels:

#![allow(unused)]
fn main() {
let (tx, rx) = mpsc::channel(10);  // Max 10 buffered items

// Producer
for i in 0..100 {
    tx.send(i).await.unwrap();  // Blocks when buffer is full
}
// Automatically slows down producer when consumer falls behind

// Without backpressure (unbounded):
let (tx, rx) = mpsc::unbounded_channel();
for i in 0..100_000_000 {
    tx.send(i).unwrap();  // Never blocks, accumulates in memory
}
// Memory: 100M items → OOM crash
}

ReceiverStream adapter (convert Receiver → Stream):

#![allow(unused)]
fn main() {
use tokio_stream::wrappers::ReceiverStream;

let (tx, rx) = mpsc::channel(100);
let stream = ReceiverStream::new(rx);  // Now it's a Stream!

// Can use stream combinators
stream
    .map(|x| x * 2)
    .filter(|x| x > 100)
    .collect()
    .await;
}

Performance:

• Send/receive: ~50ns per message (lock-free fast path)
• Throughput: ~20 million messages/sec (single thread)
• Memory: Bounded = N items × item size

Multi-producer example:

#![allow(unused)]
fn main() {
let (tx, mut rx) = mpsc::channel(100);

// Multiple producers (clone sender)
for i in 0..5 {
    let tx_clone = tx.clone();
    tokio::spawn(async move {
        tx_clone.send(format!("Producer {}", i)).await.unwrap();
    });
}
drop(tx);  // Drop original sender

// Single consumer
while let Some(msg) = rx.recv().await {
    println!("Received: {}", msg);
}
// Receives from all 5 producers (merged automatically)
}

4. Broadcast Channels (One-to-Many Fan-Out)

Broadcast channels enable one sender to send to multiple receivers (fan-out pattern).

The problem with MPSC (one consumer only):

#![allow(unused)]
fn main() {
let (tx, mut rx) = mpsc::channel(100);

// Only ONE receiver can consume
let consumer1 = tokio::spawn(async move {
    while let Some(msg) = rx.recv().await {
        process1(msg);
    }
});

// Can't create second receiver - rx is moved!
// let consumer2 = tokio::spawn(async move {
//     while let Some(msg) = rx.recv().await {  // ERROR: rx moved
//         process2(msg);
//     }
// });
}

Broadcast channel solution:

#![allow(unused)]
fn main() {
use tokio::sync::broadcast;

// Create broadcast channel (capacity 100)
let (tx, _rx) = broadcast::channel::<Event>(100);

// Multiple receivers (each gets ALL messages)
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();
let mut rx3 = tx.subscribe();

// Consumer 1: Count events
tokio::spawn(async move {
    let mut count = 0;
    while let Ok(event) = rx1.recv().await {
        count += 1;
    }
    println!("Consumer 1 counted: {}", count);
});

// Consumer 2: Alert on critical
tokio::spawn(async move {
    while let Ok(event) = rx2.recv().await {
        if event.severity == Severity::Critical {
            alert(event);
        }
    }
});

// Consumer 3: Write to database
tokio::spawn(async move {
    while let Ok(event) = rx3.recv().await {
        write_to_db(event).await;
    }
});

// Sender (all 3 receivers get this)
tx.send(event).unwrap();
}

Fan-out pattern (parallel processing):

         ┌─→ Consumer 1 (Analytics)
Sender ──┼─→ Consumer 2 (Alerting)
         └─→ Consumer 3 (Storage)

All consumers process SAME events in parallel

Performance benefits:

#![allow(unused)]
fn main() {
// Sequential processing (slow)
for event in events {
    analytics(event);     // 10ms
    alerting(event);      // 5ms
    storage(event);       // 20ms
}
// Total: 35ms per event

// Parallel fan-out (fast)
tx.send(event);
// Consumer 1: analytics (10ms) ┐
// Consumer 2: alerting (5ms)   ├─ Run in parallel
// Consumer 3: storage (20ms)   ┘
// Total: max(10, 5, 20) = 20ms per event
// Speedup: 1.75× faster
}

Lagging receivers (automatic dropping):

#![allow(unused)]
fn main() {
let (tx, _) = broadcast::channel(10);  // Buffer 10 messages
let mut rx = tx.subscribe();

// Send 20 messages (receiver can't keep up)
for i in 0..20 {
    tx.send(i).unwrap();
}

// Receiver tries to receive
match rx.recv().await {
    Ok(msg) => println!("Got: {}", msg),
    Err(RecvError::Lagged(n)) => {
        println!("Missed {} messages (dropped oldest)", n);
        // Automatically drops oldest 10 messages
    }
}

// Prevents slow consumers from blocking fast producers
}

MPSC vs Broadcast:

Real-world example (event processing):

#![allow(unused)]
fn main() {
// Stream processor with fan-out
let (broadcast_tx, _) = broadcast::channel(1000);

// Forward processed events to broadcast
let tx_clone = broadcast_tx.clone();
tokio::spawn(async move {
    while let Some(event) = stream.next().await {
        tx_clone.send(event).unwrap();
    }
});

// Multiple independent consumers
let mut analytics_rx = broadcast_tx.subscribe();
let mut alerting_rx = broadcast_tx.subscribe();
let mut storage_rx = broadcast_tx.subscribe();

// Each processes same events for different purposes
// Analytics: Compute statistics
// Alerting: Check for anomalies
// Storage: Persist to database
}

5. Windowed Aggregations (Tumbling and Sliding Windows)

Windowed aggregations compute statistics over time-based intervals for real-time analytics.

The problem with count-based batching:

#![allow(unused)]
fn main() {
// Batching by count (100 events per batch)
let batches = stream.chunks(100);

// Problem: Batch time varies with event rate
// High rate (1000 events/sec): Batch completes in 0.1s
// Low rate (10 events/sec): Batch completes in 10s
// "Events per second" is inaccurate!
}

Time-based windowing solution:

#![allow(unused)]
fn main() {
// Tumbling window (non-overlapping 1-second intervals)
let windows = stream.chunks_timeout(usize::MAX, Duration::from_secs(1));

// Guarantees: Each window is exactly 1 second
// Window 1: 0-1s (500 events)
// Window 2: 1-2s (520 events)
// Window 3: 2-3s (480 events)
// Accurate "events per second" metric!
}

Tumbling windows (non-overlapping):

Time:     0s      1s      2s      3s      4s
         ├───────┼───────┼───────┼───────┤
Window:  [  W1  ][  W2  ][  W3  ][  W4  ]

Each event belongs to exactly ONE window
Used for: Throughput metrics, event counts, periodic aggregations

Sliding windows (overlapping):

Time:     0s    0.5s    1s    1.5s    2s
         ├───────┼───────┼───────┼───────┤
Window:  [  W1 (1s)  ]
               [  W2 (1s)  ]
                     [  W3 (1s)  ]
                           [  W4 (1s)  ]

Each event belongs to MULTIPLE windows
Used for: Moving averages, trend detection, smooth metrics

Implementation example:

#![allow(unused)]
fn main() {
use std::collections::HashMap;

// Tumbling window: Group events by time bucket
pub fn create_tumbling_window<S>(
    stream: S,
    duration_ms: u64,
) -> impl Stream<Item = WindowedStats>
where
    S: Stream<Item = Event>,
{
    let mut current_window: Vec<Event> = Vec::new();
    let mut window_start = 0;

    stream.filter_map(move |event| {
        let event_window = event.timestamp / duration_ms;
        let current_window_id = window_start / duration_ms;

        if event_window > current_window_id {
            // Emit completed window
            let stats = calculate_stats(&current_window, window_start, duration_ms);
            current_window.clear();
            window_start = event_window * duration_ms;
            current_window.push(event);
            Some(stats)
        } else {
            current_window.push(event);
            None
        }
    })
}

// Calculate statistics for a window
pub fn calculate_stats(events: &[Event], start: u64, duration: u64) -> WindowedStats {
    let count = events.len();
    let avg = events.iter().map(|e| e.value).sum::<f64>() / count as f64;
    let max = events.iter().map(|e| e.value).fold(0.0, f64::max);

    WindowedStats {
        window_start: start,
        window_end: start + duration,
        event_count: count,
        avg_value: avg,
        max_value: max,
    }
}
}

Use cases:

Anomaly detection with windows:

#![allow(unused)]
fn main() {
// Compare current window to baseline
pub async fn detect_anomalies(
    current: &WindowedStats,
    baseline: &WindowedStats,
) -> Option<Alert> {
    // Spike detection (event count > 2× normal)
    if current.event_count > baseline.event_count * 2 {
        return Some(Alert {
            message: format!("Event spike: {} vs {} normal",
                current.event_count, baseline.event_count),
            severity: Severity::Critical,
        });
    }

    // Abnormal values (avg > 1.5× normal)
    if current.avg_value > baseline.avg_value * 1.5 {
        return Some(Alert {
            message: format!("Value anomaly: {:.2} vs {:.2} normal",
                current.avg_value, baseline.avg_value),
            severity: Severity::High,
        });
    }

    None
}

// Real-world example:
// Baseline: 100 events/sec, avg value 50
// Current: 500 events/sec, avg value 95
// Alert: "Event spike: 500 vs 100 normal" (5× increase)
}

Performance:

#![allow(unused)]
fn main() {
// Per-event aggregation (slow)
for event in events {
    calculate_stats_for_all_time();  // O(n) per event
}
// Complexity: O(n²)

// Windowed aggregation (fast)
for window in windows {
    calculate_stats(window);  // O(window_size) per window
}
// Complexity: O(n) total
// Speedup: n× faster
}

Memory efficiency:

#![allow(unused)]
fn main() {
// Store all events (unbounded)
let all_events: Vec<Event> = stream.collect().await;
calculate_stats(&all_events);
// Memory: O(n) - grows unbounded

// Windowed processing (constant)
stream.chunks_timeout(usize::MAX, Duration::from_secs(1))
    .for_each(|window| {
        calculate_stats(&window);  // Process and discard
    }).await;
// Memory: O(window_size) - constant
}

6. Backpressure Strategies (Drop Oldest, Sample, Bounded Buffers)

Backpressure prevents fast producers from overwhelming slow consumers and causing OOM crashes.

The problem without backpressure:

#![allow(unused)]
fn main() {
// Unbounded channel (no backpressure)
let (tx, rx) = mpsc::unbounded_channel();

// Fast producer (10,000 events/sec)
tokio::spawn(async move {
    loop {
        tx.send(generate_event()).unwrap();  // Never blocks
        tokio::time::sleep(Duration::from_micros(100)).await;
    }
});

// Slow consumer (1,000 events/sec)
tokio::spawn(async move {
    while let Some(event) = rx.recv().await {
        expensive_processing(event).await;  // 1ms per event
    }
});

// Result: 9,000 events/sec accumulate in channel
// At 1KB/event: 9MB/sec → 540MB/min → OOM crash in ~10 minutes
}

Strategy 1: Bounded buffer (automatic backpressure):

#![allow(unused)]
fn main() {
// Bounded channel (max 100 buffered)
let (tx, rx) = mpsc::channel(100);

// Producer automatically slows down when buffer fills
tx.send(event).await?;  // Blocks when 100 events buffered
// Prevents unbounded memory growth
}

Strategy 2: Drop oldest (ring buffer):

#![allow(unused)]
fn main() {
use std::collections::VecDeque;

pub struct DropOldestBuffer<T> {
    buffer: VecDeque<T>,
    capacity: usize,
}

impl<T> DropOldestBuffer<T> {
    pub fn push(&mut self, item: T) {
        if self.buffer.len() == self.capacity {
            self.buffer.pop_front();  // Drop oldest
        }
        self.buffer.push_back(item);
    }
}

// Use case: Real-time monitoring (recent data matters most)
// 100 events buffered, 101st arrives → drop 1st, keep 2nd-101st
}

Strategy 3: Drop newest (reject when full):

#![allow(unused)]
fn main() {
pub struct DropNewestBuffer<T> {
    buffer: VecDeque<T>,
    capacity: usize,
}

impl<T> DropNewestBuffer<T> {
    pub fn try_push(&mut self, item: T) -> Result<(), T> {
        if self.buffer.len() < self.capacity {
            self.buffer.push_back(item);
            Ok(())
        } else {
            Err(item)  // Reject new item
        }
    }
}

// Use case: Critical events (don't lose early data)
// Buffer full → reject new events, preserve existing
}

Strategy 4: Sampling (probabilistic dropping):

#![allow(unused)]
fn main() {
use rand::Rng;

pub fn should_sample(rate: f64) -> bool {
    rand::thread_rng().gen::<f64>() < rate
}

// Apply sampling to stream
let sampled = stream.filter(|_| should_sample(0.1));  // Keep 10%

// Use case: High-volume telemetry (statistical sampling okay)
// 10,000 events/sec × 10% = 1,000 events/sec processed
// Throughput: 10× reduction, memory: 10× less
}

Backpressure comparison:

Performance impact:

#![allow(unused)]
fn main() {
// Without backpressure
// Producer: 100,000 events/sec
// Consumer: 10,000 events/sec
// Memory growth: 90,000 events/sec × 1KB = 90MB/sec
// Time to OOM (8GB): ~90 seconds

// With sampling (10%)
// Producer: 100,000 events/sec
// After sampling: 10,000 events/sec
// Consumer: 10,000 events/sec
// Memory growth: 0 (balanced)
// Time to OOM: Never
}

Monitoring backpressure:

#![allow(unused)]
fn main() {
use std::sync::atomic::{AtomicUsize, Ordering};

pub struct BackpressureStats {
    received: AtomicUsize,
    dropped: AtomicUsize,
    processed: AtomicUsize,
}

impl BackpressureStats {
    pub fn get_report(&self) -> String {
        let received = self.received.load(Ordering::Relaxed);
        let dropped = self.dropped.load(Ordering::Relaxed);
        let drop_rate = (dropped as f64 / received as f64) * 100.0;

        format!(
            "Received: {}, Dropped: {} ({:.1}% drop rate)",
            received, dropped, drop_rate
        )
    }
}

// Output: "Received: 1000000, Dropped: 900000 (90.0% drop rate)"
// Action: Increase consumer capacity or reduce producer rate
}

Adaptive backpressure (dynamic sampling):

#![allow(unused)]
fn main() {
// Adjust sampling rate based on queue depth
pub fn adaptive_sample_rate(queue_len: usize, capacity: usize) -> f64 {
    let usage = queue_len as f64 / capacity as f64;

    match usage {
        x if x < 0.5 => 1.0,   // < 50% full: Accept all
        x if x < 0.8 => 0.5,   // 50-80% full: Sample 50%
        _ => 0.1,              // > 80% full: Sample 10%
    }
}

// Automatically reduces load as buffer fills
}

7. Stream Merging (select_all for Multi-Source)

Stream merging combines multiple streams into one, enabling unified processing of multi-source data.

The problem with separate streams:

#![allow(unused)]
fn main() {
// Separate processing for each source (code duplication)
tokio::spawn(async {
    while let Some(event) = sensor1_stream.next().await {
        process(event);  // Duplicate logic
    }
});

tokio::spawn(async {
    while let Some(event) = sensor2_stream.next().await {
        process(event);  // Duplicate logic
    }
});

tokio::spawn(async {
    while let Some(event) = sensor3_stream.next().await {
        process(event);  // Duplicate logic
    }
});

// Problems: Code duplication, harder to maintain, separate pipelines
}

Stream merging solution:

#![allow(unused)]
fn main() {
use futures::stream::{self, StreamExt};

// Merge all sources into one stream
let merged = stream::select_all(vec![
    sensor1_stream,
    sensor2_stream,
    sensor3_stream,
]);

// Single processing pipeline
merged.for_each(|event| async {
    process(event);  // One implementation
}).await;

// Benefits: DRY, unified pipeline, easier to maintain
}

How select_all works:

Stream 1: ───e1───────e4──────e7───
Stream 2: ──────e2──────e5────────e8
Stream 3: ────────e3──────e6───────

Merged:   ───e1─e2─e3─e4─e5─e6─e7─e8
          (emits items as they arrive, preserves relative order per stream)

Real-world example (IoT sensors):

#![allow(unused)]
fn main() {
// Create streams for multiple sensors
let mut sensor_streams = Vec::new();

for sensor_id in 0..10 {
    let (tx, rx) = mpsc::channel(100);

    // Each sensor sends data independently
    tokio::spawn(async move {
        loop {
            let reading = read_sensor(sensor_id).await;
            tx.send(Event::new(sensor_id, reading)).await.unwrap();
            sleep(Duration::from_millis(100)).await;
        }
    });

    sensor_streams.push(ReceiverStream::new(rx));
}

// Merge all sensor streams
let merged = stream::select_all(sensor_streams);

// Process all sensors with unified pipeline
let processed = merged
    .map(process_event)
    .filter(|e| e.severity > Severity::Medium)
    .chunks_timeout(100, Duration::from_secs(1));

// Single pipeline handles all 10 sensors!
}

Performance benefits:

#![allow(unused)]
fn main() {
// Separate tasks (high overhead)
// 10 sensors × 1KB stack = 10KB
// 10 separate pipelines = 10× code

// Merged stream (efficient)
// 1 merged stream = 1KB stack
// 1 unified pipeline = 1× code
// Memory: 10× less, Code: 10× less
}

Ordering guarantees:

#![allow(unused)]
fn main() {
// select_all preserves per-stream order
Stream A: [1, 2, 3]
Stream B: [4, 5, 6]

// Possible merged outputs (many valid orderings):
[1, 4, 2, 5, 3, 6]  ✓ Valid (preserves A: 1→2→3, B: 4→5→6)
[1, 2, 4, 5, 3, 6]  ✓ Valid
[4, 1, 5, 2, 6, 3]  ✓ Valid
[4, 1, 2, 5, 3, 6]  ✓ Valid
[2, 1, 4, 5, 3, 6]  ✗ Invalid (A order violated: 2 before 1)

// Per-stream order guaranteed, cross-stream order is interleaved
}

Merging typed sources:

#![allow(unused)]
fn main() {
#[derive(Debug, Clone)]
pub enum SourceType {
    Sensor(String),
    LogFile(String),
    API(String),
}

#[derive(Debug, Clone)]
pub struct Event {
    pub source_type: SourceType,
    pub data: String,
}

// Merge different source types
let sensors = sensor_stream.map(|data| Event {
    source_type: SourceType::Sensor("temp".into()),
    data,
});

let logs = log_stream.map(|data| Event {
    source_type: SourceType::LogFile("app.log".into()),
    data,
});

let api = api_stream.map(|data| Event {
    source_type: SourceType::API("metrics".into()),
    data,
});

let merged = stream::select_all(vec![sensors, logs, api]);

// Process based on source type
merged.for_each(|event| async {
    match event.source_type {
        SourceType::Sensor(name) => process_sensor(&name, &event.data),
        SourceType::LogFile(path) => process_log(&path, &event.data),
        SourceType::API(endpoint) => process_api(&endpoint, &event.data),
    }
}).await;
}

Dynamic stream addition:

#![allow(unused)]
fn main() {
// Start with initial streams
let (merge_tx, merge_rx) = mpsc::channel(100);
let merged = ReceiverStream::new(merge_rx);

// Add new streams dynamically
fn add_stream(tx: mpsc::Sender<Event>, stream: impl Stream<Item = Event>) {
    tokio::spawn(async move {
        tokio::pin!(stream);
        while let Some(event) = stream.next().await {
            tx.send(event).await.unwrap();
        }
    });
}

// Add streams at runtime
add_stream(merge_tx.clone(), new_sensor_stream());
// Merged stream now includes new source!
}

8. AtomicUsize (Lock-Free Counters)

AtomicUsize provides lock-free shared counters for concurrent statistics tracking.

The problem with Mutex (locking overhead):

#![allow(unused)]
fn main() {
use std::sync::{Arc, Mutex};

// Shared counter with mutex
let counter = Arc::new(Mutex::new(0));

// Multiple tasks increment counter
for _ in 0..10 {
    let counter_clone = Arc::clone(&counter);
    tokio::spawn(async move {
        for _ in 0..1000 {
            let mut guard = counter_clone.lock().unwrap();  // Acquire lock
            *guard += 1;
            drop(guard);  // Release lock
        }
    });
}

// Problems:
// 1. Blocking: Each increment waits for lock
// 2. Contention: High overhead with many threads
// 3. Performance: ~100ns per increment (lock overhead)
}

AtomicUsize solution (lock-free):

#![allow(unused)]
fn main() {
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

// Shared counter without locks
let counter = Arc::new(AtomicUsize::new(0));

// Multiple tasks increment atomically
for _ in 0..10 {
    let counter_clone = Arc::clone(&counter);
    tokio::spawn(async move {
        for _ in 0..1000 {
            counter_clone.fetch_add(1, Ordering::Relaxed);  // Atomic, no lock!
        }
    });
}

// Benefits:
// 1. No blocking: Never waits
// 2. No contention: Lock-free
// 3. Performance: ~5ns per increment (20× faster than Mutex)
}

Memory ordering (Relaxed vs Acquire/Release):

Backpressure stats example:

#![allow(unused)]
fn main() {
pub struct BackpressureStats {
    received: AtomicUsize,   // Total events received
    dropped: AtomicUsize,    // Events dropped
    processed: AtomicUsize,  // Events processed
}

impl BackpressureStats {
    pub fn new() -> Self {
        Self {
            received: AtomicUsize::new(0),
            dropped: AtomicUsize::new(0),
            processed: AtomicUsize::new(0),
        }
    }

    // Thread-safe increment (no lock needed!)
    pub fn increment_received(&self) {
        self.received.fetch_add(1, Ordering::Relaxed);
    }

    pub fn increment_dropped(&self) {
        self.dropped.fetch_add(1, Ordering::Relaxed);
    }

    // Thread-safe read
    pub fn get_report(&self) -> String {
        let received = self.received.load(Ordering::Relaxed);
        let dropped = self.dropped.load(Ordering::Relaxed);
        let drop_rate = (dropped as f64 / received as f64) * 100.0;

        format!("Received: {}, Dropped: {} ({:.1}%)", received, dropped, drop_rate)
    }
}

// Usage from multiple tasks
let stats = Arc::new(BackpressureStats::new());

// Producer increments received
stats.increment_received();

// Consumer increments processed or dropped
if should_drop {
    stats.increment_dropped();
} else {
    stats.increment_processed();
}

// Monitor prints stats
println!("{}", stats.get_report());
}

Performance comparison:

#![allow(unused)]
fn main() {
use std::time::Instant;

// Mutex (blocking)
let mutex_counter = Arc::new(Mutex::new(0));
let start = Instant::now();
for _ in 0..1_000_000 {
    *mutex_counter.lock().unwrap() += 1;
}
println!("Mutex: {:?}", start.elapsed());  // ~100ms

// AtomicUsize (lock-free)
let atomic_counter = Arc::new(AtomicUsize::new(0));
let start = Instant::now();
for _ in 0..1_000_000 {
    atomic_counter.fetch_add(1, Ordering::Relaxed);
}
println!("Atomic: {:?}", start.elapsed());  // ~5ms

// Speedup: 20× faster with AtomicUsize
}

When to use AtomicUsize:

• ✅ Counters, flags, simple statistics
• ✅ Lock-free progress tracking
• ✅ High-contention scenarios (many threads)
• ❌ Complex shared state (use Mutex)
• ❌ Bulk updates (use Mutex for batch)

Common atomic operations:

#![allow(unused)]
fn main() {
let atomic = AtomicUsize::new(0);

// Load (read)
let value = atomic.load(Ordering::Relaxed);

// Store (write)
atomic.store(42, Ordering::Relaxed);

// Fetch-add (increment)
let old = atomic.fetch_add(1, Ordering::Relaxed);

// Fetch-sub (decrement)
let old = atomic.fetch_sub(1, Ordering::Relaxed);

// Compare-and-swap (conditional update)
let result = atomic.compare_exchange(
    0,      // Expected value
    100,    // New value
    Ordering::Relaxed,
    Ordering::Relaxed,
);

// Fetch-max (update to max)
let old = atomic.fetch_max(50, Ordering::Relaxed);
}

Real-world monitoring example:

#![allow(unused)]
fn main() {
// Progress tracker
pub struct ProgressTracker {
    total: AtomicUsize,
    completed: AtomicUsize,
    failed: AtomicUsize,
}

impl ProgressTracker {
    pub fn record_completion(&self, success: bool) {
        self.completed.fetch_add(1, Ordering::Relaxed);
        if !success {
            self.failed.fetch_add(1, Ordering::Relaxed);
        }
    }

    pub fn report(&self) -> String {
        let total = self.total.load(Ordering::Relaxed);
        let completed = self.completed.load(Ordering::Relaxed);
        let failed = self.failed.load(Ordering::Relaxed);
        let percent = (completed as f64 / total as f64) * 100.0;

        format!(
            "[{}/{}] {:.1}% complete, {} failed",
            completed, total, percent, failed
        )
    }
}

// Update from any task, no locks needed!
tracker.record_completion(true);
}

9. Pin and Unpin (Stream Trait Bounds)

Pin and Unpin are critical for safe async stream handling in Rust.

The problem with movable futures:

#![allow(unused)]
fn main() {
// Async function creates a future
async fn fetch_data(url: &str) -> String {
    // 'url' is borrowed - future stores pointer to it
    reqwest::get(url).await.unwrap().text().await.unwrap()
}

let url = String::from("https://example.com");
let future = fetch_data(&url);

// If future moves in memory, stored pointer becomes invalid!
// Pin prevents this by guaranteeing the future won't move
}

Pin guarantees:

• Once pinned, T will never move in memory
• Self-referential structures are safe (async futures often are)
• Required for Stream::poll_next() to work correctly

Unpin marker trait:

#![allow(unused)]
fn main() {
// Most types are Unpin (safe to move)
impl Unpin for i32 {}
impl Unpin for String {}
impl Unpin for Vec<T> {}

// Futures and streams are NOT Unpin by default
// (they may contain self-references)
}

Stream trait bound with Unpin:

#![allow(unused)]
fn main() {
// Common pattern: Require Unpin for ergonomic APIs
async fn consume_stream<S>(mut stream: S)
where
    S: Stream<Item = Event> + Unpin,  // Unpin required!
{
    while let Some(event) = stream.next().await {
        process(event);
    }
}

// Without Unpin, you'd need to pin manually:
async fn consume_stream_pinned<S>(stream: S)
where
    S: Stream<Item = Event>,  // No Unpin
{
    tokio::pin!(stream);  // Pin to stack
    while let Some(event) = stream.next().await {
        process(event);
    }
}
}

Why ReceiverStream is Unpin:

#![allow(unused)]
fn main() {
use tokio_stream::wrappers::ReceiverStream;

// ReceiverStream wraps Receiver (which is Unpin)
let (tx, rx) = mpsc::channel(100);
let stream = ReceiverStream::new(rx);  // ReceiverStream<T>: Unpin

// Can use directly without pinning
consume_stream(stream).await;  // Works!
}

tokio::pin! macro (pin to stack):

#![allow(unused)]
fn main() {
use tokio::pin;

async fn process_complex_stream<S>(stream: S)
where
    S: Stream<Item = Event>,  // Not Unpin
{
    // Pin to stack (stack-allocated Pin<&mut S>)
    pin!(stream);

    while let Some(event) = stream.next().await {
        process(event);
    }
    // stream unpinned when it goes out of scope
}
}

Box::pin (pin to heap):

#![allow(unused)]
fn main() {
use futures::stream::{self, StreamExt};

// Create stream that's NOT Unpin
let stream = stream::iter(vec![1, 2, 3])
    .then(|x| async move {
        tokio::time::sleep(Duration::from_millis(100)).await;
        x * 2
    });

// Pin to heap (heap-allocated Pin<Box<dyn Stream>>)
let mut pinned = Box::pin(stream);

while let Some(value) = pinned.next().await {
    println!("{}", value);
}
}

Common pattern (generic Unpin bound):

#![allow(unused)]
fn main() {
// Accept any stream that is Unpin
pub fn process_stream<S>(stream: S) -> impl Future<Output = ()>
where
    S: Stream<Item = Event> + Unpin + Send + 'static,
{
    async move {
        tokio::pin!(stream);  // Or just use directly if Unpin
        while let Some(event) = stream.next().await {
            process(event);
        }
    }
}
}

When you need Pin:

• Implementing custom Stream types
• Working with combinators that return non-Unpin streams
• Storing futures/streams in structs

When you don’t need Pin:

• Using ReceiverStream, BroadcastStream (they’re Unpin)
• Using Vec/HashMap iterators (they’re Unpin)
• Simple stream::iter() (Unpin)

10. Error Handling in Streams (Result<T, E> Items)

Stream error handling differs from sync iterators—errors don’t stop the stream.

Synchronous iterator (error stops iteration):

#![allow(unused)]
fn main() {
// Iterator of Results
let results: Vec<Result<i32, &str>> = vec![
    Ok(1),
    Ok(2),
    Err("error"),  // Stops here
    Ok(4),
];

// collect() stops at first error
let values: Result<Vec<i32>, &str> = results.into_iter().collect();
assert_eq!(values, Err("error"));  // Lost Ok(4)!
}

Stream with errors (can continue processing):

#![allow(unused)]
fn main() {
use tokio_stream::{self as stream, StreamExt};

// Stream of Results
let results = stream::iter(vec![
    Ok(1),
    Ok(2),
    Err("error"),
    Ok(4),  // Can still process this!
]);

// Option 1: Stop at first error
let values: Result<Vec<i32>, &str> = results
    .try_collect()  // Stops at first Err
    .await;
// Result: Err("error")

// Option 2: Filter errors, keep successes
let values: Vec<i32> = stream::iter(vec![
    Ok(1),
    Ok(2),
    Err("error"),
    Ok(4),
])
    .filter_map(|r| r.ok())  // Keep only Ok values
    .collect()
    .await;
// Result: [1, 2, 4] (skipped error, continued processing)
}

filter_map for error handling:

#![allow(unused)]
fn main() {
// Parse events, skip invalid ones
async fn process_stream(stream: impl Stream<Item = String>) {
    let parsed = stream.filter_map(|s| {
        match parse_event(&s) {
            Ok(event) => Some(event),   // Keep valid
            Err(e) => {
                eprintln!("Parse error: {}", e);
                None  // Skip invalid
            }
        }
    });

    parsed.for_each(|event| async {
        process(event);
    }).await;
}

// Handles errors gracefully without stopping stream
}

Partial results pattern:

#![allow(unused)]
fn main() {
#[derive(Debug)]
pub struct ProcessingResult<T, E> {
    pub successes: Vec<T>,
    pub failures: Vec<E>,
}

// Collect both successes and failures
pub async fn process_with_errors<S>(stream: S) -> ProcessingResult<Event, String>
where
    S: Stream<Item = Result<Event, String>>,
{
    let mut successes = Vec::new();
    let mut failures = Vec::new();

    tokio::pin!(stream);

    while let Some(result) = stream.next().await {
        match result {
            Ok(event) => successes.push(event),
            Err(e) => failures.push(e),
        }
    }

    ProcessingResult { successes, failures }
}

// Usage
let result = process_with_errors(stream).await;
println!("Processed: {} events, {} errors",
    result.successes.len(), result.failures.len());
}

Retry on error (with exponential backoff):

#![allow(unused)]
fn main() {
use tokio::time::{sleep, Duration};

// Retry failed operations
pub fn retry_on_error<S, T, E>(
    stream: S,
    max_retries: usize,
) -> impl Stream<Item = Result<T, E>>
where
    S: Stream<Item = Result<T, E>>,
    E: std::fmt::Display,
{
    stream.then(move |result| async move {
        let mut attempts = 0;
        let mut current_result = result;

        while attempts < max_retries {
            match current_result {
                Ok(value) => return Ok(value),
                Err(e) => {
                    attempts += 1;
                    if attempts >= max_retries {
                        return Err(e);
                    }

                    eprintln!("Retry {} after error: {}", attempts, e);
                    let backoff = Duration::from_secs(2u64.pow(attempts as u32));
                    sleep(backoff).await;

                    // In real code, re-attempt the operation here
                    current_result = Err(e);
                }
            }
        }

        current_result
    })
}
}

Error rate monitoring:

#![allow(unused)]
fn main() {
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

pub struct ErrorStats {
    total: AtomicUsize,
    errors: AtomicUsize,
}

impl ErrorStats {
    pub fn record<T, E>(&self, result: &Result<T, E>) {
        self.total.fetch_add(1, Ordering::Relaxed);
        if result.is_err() {
            self.errors.fetch_add(1, Ordering::Relaxed);
        }
    }

    pub fn error_rate(&self) -> f64 {
        let total = self.total.load(Ordering::Relaxed);
        let errors = self.errors.load(Ordering::Relaxed);
        if total > 0 {
            (errors as f64 / total as f64) * 100.0
        } else {
            0.0
        }
    }
}

// Monitor error rate in stream
let stats = Arc::new(ErrorStats::new());
let stats_clone = Arc::clone(&stats);

stream
    .inspect(move |result| {
        stats_clone.record(result);
    })
    .for_each(|result| async {
        if let Ok(event) = result {
            process(event);
        }
    })
    .await;

println!("Error rate: {:.2}%", stats.error_rate());
}

Connection to This Project

This project demonstrates how async stream processing patterns enable production-ready real-time data pipelines.

Milestone 1: Basic Event Stream from Channel

Concepts applied:

• Async Streams: ReceiverStream converts mpsc::Receiver to Stream
• MPSC Channels: Producer-consumer communication
• Stream consumption: .next().await pattern

Why it matters: Transform channel into stream for composable processing:

#![allow(unused)]
fn main() {
let (tx, rx) = mpsc::channel(100);
let stream = ReceiverStream::new(rx);  // Now can use stream combinators!

stream
    .map(|e| process(e))
    .filter(|e| e.value > 50.0)
    .collect()
    .await;
}

Real-world impact:

#![allow(unused)]
fn main() {
// Manual channel consumption (verbose)
while let Some(event) = rx.recv().await {
    if event.value > 50.0 {
        let processed = process(event);
        results.push(processed);
    }
}

// Stream combinators (concise)
let results = ReceiverStream::new(rx)
    .filter(|e| e.value > 50.0)
    .map(process)
    .collect()
    .await;

// Lines of code: 6 → 4 (33% less)
// Clarity: Declarative > Imperative
}

Performance baseline:

• Sequential processing: 100 events/sec (10ms each)
• Stream doesn’t add overhead (zero-cost abstraction)
• Same performance, better ergonomics

Milestone 2: Stream Transformations and Filtering

Concepts applied:

• Stream Combinators: .map(), .filter(), .filter_map()
• Zero-cost abstractions: Compiles to efficient loops
• Composition: Chain multiple transformations

Why it matters: Build complex processing pipelines declaratively:

#![allow(unused)]
fn main() {
let processed = stream
    .map(|e| Event { value: e.value * 2.0, ..e })     // Transform
    .filter(|e| e.value > 50.0)                        // Filter
    .map(|e| ProcessedEvent {                          // Enrich
        normalized: normalize_value(e.value, 0.0, 100.0),
        severity: calculate_severity(e.value),
        original: e,
    })
    .collect()
    .await;
}

Real-world impact:

#![allow(unused)]
fn main() {
// Manual filtering + transformation (15 lines)
let mut results = Vec::new();
while let Some(event) = stream.next().await {
    if event.value > 50.0 {
        let normalized = normalize_value(event.value, 0.0, 100.0);
        let severity = calculate_severity(normalized);
        results.push(ProcessedEvent { normalized, severity, original: event });
    }
}

// Stream combinators (3 lines)
let results = stream
    .filter(|e| e.value > 50.0)
    .map(process_event)
    .collect().await;

// Code reduction: 15 → 3 lines (5× less)
// Maintainability: Clear intent, easier to modify
}

Performance:

• Lazy evaluation: Transformations only run for items passing filters
• 1M events, 10% pass filter: Process 100K instead of 1M
• Speedup: 10× fewer transformations (90% filtered out early)

Milestone 3: Buffering and Batching

Concepts applied:

• chunks_timeout: Batch by count OR time
• Amortized overhead: Batch operations reduce fixed costs
• Latency vs throughput tradeoff: Tune batch size and timeout

Why it matters: Batching amortizes expensive operations (DB writes, network sends):

#![allow(unused)]
fn main() {
// One-at-a-time (slow)
stream.for_each(|event| async {
    write_to_db(vec![event]).await;  // 1 write per event
}).await;
// 1,000 events × 1ms per write = 1,000ms

// Batched (fast)
stream
    .chunks_timeout(100, Duration::from_millis(500))
    .for_each(|batch| async {
        write_to_db(batch).await;  // 1 write per 100 events
    }).await;
// 1,000 events ÷ 100 per batch × 1ms per write = 10ms
// Speedup: 100× faster
}

Real-world impact:

#![allow(unused)]
fn main() {
// Database insertion benchmark
// Individual inserts: 1,000 events/sec (1ms overhead each)
for event in events {
    db.execute("INSERT INTO events VALUES (?)", event).await;
}
// Total: 1,000ms for 1,000 events

// Batch inserts: 100,000 events/sec (1ms overhead per batch of 100)
for batch in events.chunks(100) {
    db.execute_batch("INSERT INTO events VALUES (?)", batch).await;
}
// Total: 10ms for 1,000 events
// Speedup: 100× faster (same data, 1% of the time)
}

Latency-throughput tradeoff:

Production config (balanced):

#![allow(unused)]
fn main() {
stream.chunks_timeout(50, Duration::from_millis(200))
// Batch of 50: ~10ms latency per batch
// Timeout 200ms: Max 200ms delay for slow periods
// Throughput: 50,000 events/sec (high load)
// Latency: 200ms max (low load, timeout triggers)
}

Milestone 4: Windowed Aggregations

Concepts applied:

• Tumbling windows: Non-overlapping time intervals
• Time-based grouping: Assign events to windows by timestamp
• Aggregate statistics: Count, average, max per window

Why it matters: Enable accurate real-time analytics with time guarantees:

#![allow(unused)]
fn main() {
// Count-based batching (inaccurate)
stream.chunks(100).for_each(|batch| {
    println!("Batch of 100");  // How long did this take? Unknown!
});

// Time-based windowing (accurate)
create_tumbling_window(stream, 1000)  // 1-second windows
    .for_each(|stats| {
        println!("Events per second: {}", stats.event_count);  // Guaranteed 1s window
    }).await;
}

Real-world impact:

#![allow(unused)]
fn main() {
// Real-time monitoring dashboard
// Requirement: Display "requests per second" chart

// Wrong approach (count-based)
stream.chunks(100).for_each(|batch| {
    // Problem: 100 requests could take 0.1s (1000 req/s) or 10s (10 req/s)
    // Chart would show constant 100 req/batch (misleading!)
});

// Correct approach (time-based windows)
create_tumbling_window(stream, 1000).for_each(|stats| {
    // Guaranteed 1-second window
    println!("Requests per second: {}", stats.event_count);
    // Accurate chart: 1000, 1050, 980, ... (true throughput)
}).await;
}

Anomaly detection:

#![allow(unused)]
fn main() {
let mut baseline = None;

create_tumbling_window(stream, 1000).for_each(|stats| async {
    if let Some(ref base) = baseline {
        if stats.event_count > base.event_count * 2 {
            alert!("Traffic spike: {} vs {} normal", stats.event_count, base.event_count);
            // Example: Normal 100 req/s → Spike 250 req/s → Alert!
        }
    } else {
        baseline = Some(stats);  // Establish baseline
    }
}).await;
}

Performance:

#![allow(unused)]
fn main() {
// Per-event aggregation (slow)
let mut total = 0;
for event in all_events {
    total += event.value;
    avg = total / count;  // Recalculate for every event
}
// Complexity: O(n) work per event = O(n²) total

// Windowed aggregation (fast)
for window in windows {
    let sum: f64 = window.iter().map(|e| e.value).sum();
    let avg = sum / window.len();  // Once per window
}
// Complexity: O(window_size) per window = O(n) total
// Speedup: n× faster (1000 events = 1000× speedup)
}

Milestone 5: Backpressure Handling

Concepts applied:

• Bounded channels: Natural backpressure (blocks sender when full)
• Drop strategies: DropOldest, DropNewest, Sampling
• AtomicUsize: Lock-free statistics tracking

Why it matters: Prevent OOM crashes when producer outpaces consumer:

#![allow(unused)]
fn main() {
// Without backpressure (crashes)
let (tx, rx) = mpsc::unbounded_channel();

// Fast producer: 10,000 events/sec
loop {
    tx.send(generate_event()).unwrap();  // Never blocks
}

// Slow consumer: 1,000 events/sec
while let Some(event) = rx.recv().await {
    expensive_processing(event).await;  // 1ms each
}

// Result: 9,000 events/sec accumulate → OOM in 10 minutes

// With backpressure (stable)
let (tx, rx) = mpsc::channel(100);  // Bounded to 100

// Producer automatically slows to consumer's pace
tx.send(event).await?;  // Blocks when buffer full
// Memory: Constant (100 events max)
}

Real-world impact:

#![allow(unused)]
fn main() {
// Production scenario: Sensor data processing
// Input rate: 100,000 events/sec (100 sensors × 1,000 Hz each)
// Processing capacity: 10,000 events/sec (expensive ML inference)

// Strategy 1: Bounded buffer (slow down sensors)
let (tx, rx) = mpsc::channel(1000);
// Problem: Sensors must slow down (not always possible)

// Strategy 2: Sampling (keep 10%)
let (tx, rx) = mpsc::channel(10000);
let sampled = stream.filter(|_| should_sample(0.1));  // Random 10%
// Result: 10,000 events/sec processed
// Trade-off: Lose 90% of data, but system stable

// Strategy 3: Adaptive sampling (dynamic)
let usage = queue.len() as f64 / capacity as f64;
let rate = if usage < 0.5 { 1.0 } else if usage < 0.8 { 0.5 } else { 0.1 };
let sampled = stream.filter(|_| should_sample(rate));
// Result: Automatically adjusts sampling based on load
// < 50% full: Accept all
// 50-80% full: Sample 50%
// > 80% full: Sample 10%
}

Memory comparison:

#![allow(unused)]
fn main() {
// Unbounded (OOM)
// 100,000 events/sec input - 10,000 events/sec processing
// Accumulation: 90,000 events/sec × 1KB/event = 90MB/sec
// Time to OOM (8GB RAM): ~90 seconds

// Bounded (stable)
// Buffer size: 1000 events
// Memory: 1000 events × 1KB = 1MB (constant)
// Time to OOM: Never

// Sampling 10% (stable)
// After sampling: 10,000 events/sec (matches processing)
// Memory: Minimal buffer (100 events = 100KB)
// Time to OOM: Never
}

Milestone 6: Multi-Source Stream Merging and Fan-Out

Concepts applied:

• Stream merging: select_all combines multiple streams
• Broadcast channels: One-to-many fan-out
• Parallel consumers: Multiple tasks process same events

Why it matters: Unified processing for multiple sources, parallel consumers for same data:

#![allow(unused)]
fn main() {
// Merging: 3 sensor streams → 1 processing pipeline
let merged = stream::select_all(vec![
    sensor1_stream,
    sensor2_stream,
    sensor3_stream,
]);

let processed = merged.map(process_event);  // One implementation for all!

// Fan-out: 1 stream → 3 consumers
let (broadcast_tx, _) = broadcast::channel(1000);
let analytics_rx = broadcast_tx.subscribe();
let alerting_rx = broadcast_tx.subscribe();
let storage_rx = broadcast_tx.subscribe();

// Each processes same events independently in parallel
}

Real-world impact:

#![allow(unused)]
fn main() {
// Separate pipelines (inefficient)
tokio::spawn(async { process_sensor1().await });  // Duplicate code
tokio::spawn(async { process_sensor2().await });  // Duplicate code
tokio::spawn(async { process_sensor3().await });  // Duplicate code
// Problems: 3× code, hard to maintain, inconsistent processing

// Merged pipeline (efficient)
let merged = merge_streams(vec![sensor1, sensor2, sensor3]);
process_unified_pipeline(merged).await;  // One implementation
// Benefits: 1× code, easy to maintain, consistent processing
}

Parallel processing with fan-out:

#![allow(unused)]
fn main() {
// Sequential consumers (slow)
for event in events {
    analytics(event);   // 10ms
    alerting(event);    // 5ms
    storage(event);     // 20ms
}
// Total: 35ms per event

// Parallel fan-out (fast)
let (tx, _) = broadcast::channel(1000);
let rx1 = tx.subscribe();
let rx2 = tx.subscribe();
let rx3 = tx.subscribe();

tokio::spawn(async move { while let Ok(e) = rx1.recv().await { analytics(e); } });
tokio::spawn(async move { while let Ok(e) = rx2.recv().await { alerting(e); } });
tokio::spawn(async move { while let Ok(e) = rx3.recv().await { storage(e); } });

for event in events {
    tx.send(event);  // All 3 consumers process in parallel
}
// Total: max(10, 5, 20) = 20ms per event
// Speedup: 1.75× faster
}

Production architecture:

10 Sensors ─┬─ merge ─→ Processing Pipeline ─→ broadcast ─┬─→ Analytics DB
            │                                               ├─→ Alerting Service
            │                                               ├─→ Long-term Storage
            │                                               └─→ Real-time Dashboard

Simplified: 10 sources → 1 pipeline → 4 consumers
Benefits: DRY code, unified processing, parallel consumers

Milestone 1: Basic Event Stream from Channel

Introduction

Before processing streams, you need to understand async stream fundamentals. This milestone teaches you to create streams from channels, consume them with .next().await, and apply basic transformations.

Why Start Here: Streams are the async equivalent of iterators. Channels naturally produce streams (events arrive over time). Understanding stream consumption patterns is foundational—everything builds on .next().await and stream combinators.

Architecture

Structs:

• Event - Represents a single event

  • Field id: u64 - Unique event identifier
  • Field timestamp: u64 - Unix timestamp (milliseconds)
  • Field source: String - Event origin (e.g., “sensor-1”)
  • Field value: f64 - Event payload (temperature, metric, etc.)
  • Field event_type: EventType - Category of event

• EventType - Enum for event categories

  • Variant Metric, Log, Alert, Error

Key Functions:

• async fn create_event_stream(rx: mpsc::Receiver<Event>) -> impl Stream<Item = Event> - Wraps channel receiver as stream
• async fn generate_events(tx: mpsc::Sender<Event>, count: usize, delay_ms: u64) - Simulates event source
• async fn consume_stream(mut stream: impl Stream<Item = Event> + Unpin) - Consumes and prints events

Role Each Plays:

• tokio::sync::mpsc: Channel for sending events from producer to consumer
• ReceiverStream: Adapter that makes Receiver<T> into Stream<Item = T>
• Stream trait: Async iterator (yields items over time via .next().await)

Starter Code

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};
use tokio_stream::{Stream, StreamExt};
use tokio_stream::wrappers::ReceiverStream;

#[derive(Debug, Clone)]
pub enum EventType {
    Metric,
    Log,
    Alert,
    Error,
}

#[derive(Debug, Clone)]
pub struct Event {
    pub id: u64,
    pub timestamp: u64,
    pub source: String,
    pub value: f64,
    pub event_type: EventType,
}

impl Event {
    pub fn new(id: u64, source: String, value: f64, event_type: EventType) -> Self {
        // TODO: Create event with current timestamp
        // Hint: Use std::time::SystemTime::now()
        //       .duration_since(std::time::UNIX_EPOCH)
        //       .unwrap()
        //       .as_millis() as u64

        todo!("Implement Event::new")
    }
}

pub async fn generate_events(tx: mpsc::Sender<Event>, count: usize, delay_ms: u64) {
    // TODO: Generate 'count' events with 'delay_ms' between each
    // TODO: Send events through the channel
    // TODO: Vary the source and value to simulate different sensors

    todo!("Implement event generator")
}

pub async fn consume_stream<S>(mut stream: S)
where
    S: Stream<Item = Event> + Unpin,
{
    // TODO: Use while let Some(event) = stream.next().await
    // TODO: Print each event

    todo!("Implement stream consumer")
}

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(100);

    // Spawn event generator
    tokio::spawn(generate_events(tx, 20, 100));

    // Convert receiver to stream and consume
    let stream = ReceiverStream::new(rx);
    consume_stream(stream).await;
}

Implementation Hints:

1. Use std::time::SystemTime::now().duration_since(UNIX_EPOCH) for timestamps
2. Use tx.send(event).await? to send events (returns Result)
3. Use sleep(Duration::from_millis(delay_ms)).await for delays
4. Use stream.next().await to get next item (returns Option<T>)
5. Channel automatically closes when sender is dropped

Checkpoint Tests

#![allow(unused)]
fn main() {
#[tokio::test]
async fn test_event_creation() {
    let event = Event::new(1, "sensor-1".to_string(), 23.5, EventType::Metric);

    assert_eq!(event.id, 1);
    assert_eq!(event.source, "sensor-1");
    assert_eq!(event.value, 23.5);
}

#[tokio::test]
async fn test_channel_to_stream() {
    use tokio::sync::mpsc;
    use tokio_stream::wrappers::ReceiverStream;
    use tokio_stream::StreamExt;

    let (tx, rx) = mpsc::channel(10);

    // Send some events
    tx.send(Event::new(1, "test".into(), 1.0, EventType::Metric)).await.unwrap();
    tx.send(Event::new(2, "test".into(), 2.0, EventType::Metric)).await.unwrap();
    drop(tx); // Close channel

    let mut stream = ReceiverStream::new(rx);

    let event1 = stream.next().await.unwrap();
    let event2 = stream.next().await.unwrap();
    let event3 = stream.next().await; // Should be None (stream closed)

    assert_eq!(event1.id, 1);
    assert_eq!(event2.id, 2);
    assert!(event3.is_none());
}

#[tokio::test]
async fn test_event_generator() {
    use tokio::sync::mpsc;

    let (tx, mut rx) = mpsc::channel(100);

    tokio::spawn(generate_events(tx, 10, 1));

    let mut count = 0;
    while let Some(_event) = rx.recv().await {
        count += 1;
    }

    assert_eq!(count, 10);
}

#[tokio::test]
async fn test_stream_consumption() {
    use tokio::sync::mpsc;
    use tokio_stream::wrappers::ReceiverStream;
    use tokio_stream::StreamExt;

    let (tx, rx) = mpsc::channel(10);

    tokio::spawn(async move {
        for i in 0..5 {
            tx.send(Event::new(i, "sensor".into(), i as f64, EventType::Metric))
                .await
                .unwrap();
        }
    });

    let stream = ReceiverStream::new(rx);
    let events: Vec<Event> = stream.collect().await;

    assert_eq!(events.len(), 5);
    assert_eq!(events[0].id, 0);
    assert_eq!(events[4].id, 4);
}
}

Milestone 2: Stream Transformations and Filtering

Introduction

Why Milestone 1 Isn’t Enough: Raw event streams need transformation—unit conversion, normalization, filtering. Doing this manually with loops is verbose and error-prone.

The Improvement: Use stream combinators (.map(), .filter(), .filter_map()) to transform and filter events declaratively. These combinators compose—chain multiple transformations without intermediate collections.

Performance: Stream combinators are zero-cost abstractions. They compile to efficient loops with no overhead. Lazy evaluation means transformations only run for items that pass filters.

Architecture

Structs:

• Reuse Event and EventType from Milestone 1

• ProcessedEvent - Enriched event after processing

  • Field original: Event - The source event
  • Field normalized_value: f64 - Value scaled to [0, 1]
  • Field severity: Severity - Computed severity level

• Severity - Enum for event severity

  • Variant Low, Medium, High, Critical

Key Functions:

• fn normalize_value(value: f64, min: f64, max: f64) -> f64 - Scales value to [0, 1] range
• fn calculate_severity(value: f64) -> Severity - Maps normalized value to severity
• async fn process_stream(stream: impl Stream<Item = Event>) -> impl Stream<Item = ProcessedEvent> - Transformation pipeline

Role Each Plays:

• map: Transforms each item (like iterator map)
• filter: Keeps items matching predicate
• filter_map: Combines filter + map (map returns Option)

Starter Code

#[derive(Debug, Clone, PartialEq)]
pub enum Severity {
    Low,
    Medium,
    High,
    Critical,
}

#[derive(Debug, Clone)]
pub struct ProcessedEvent {
    pub original: Event,
    pub normalized_value: f64,
    pub severity: Severity,
}

pub fn normalize_value(value: f64, min: f64, max: f64) -> f64 {
    // TODO: Normalize value to [0, 1] range
    // Formula: (value - min) / (max - min)
    // Handle edge case where min == max

    todo!("Implement normalization")
}

pub fn calculate_severity(normalized: f64) -> Severity {
    // TODO: Map normalized value [0, 1] to severity
    // 0.0 - 0.3: Low
    // 0.3 - 0.6: Medium
    // 0.6 - 0.9: High
    // 0.9 - 1.0: Critical

    todo!("Implement severity calculation")
}

pub fn process_stream<S>(stream: S) -> impl Stream<Item = ProcessedEvent>
where
    S: Stream<Item = Event>,
{
    // TODO: Use .map() to transform Event -> ProcessedEvent
    // TODO: For each event, normalize value (assume range 0-100)
    // TODO: Calculate severity from normalized value
    // TODO: Return ProcessedEvent

    todo!("Implement stream processing")
}

pub fn filter_high_severity<S>(stream: S) -> impl Stream<Item = ProcessedEvent>
where
    S: Stream<Item = ProcessedEvent>,
{
    // TODO: Use .filter() to keep only High and Critical severity events

    todo!("Implement severity filtering")
}

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(100);

    tokio::spawn(generate_events(tx, 50, 50));

    let stream = ReceiverStream::new(rx);

    let processed = process_stream(stream);
    let high_severity = filter_high_severity(processed);

    high_severity
        .for_each(|event| async move {
            println!("HIGH SEVERITY: {:?} -> {:?}", event.original.source, event.severity);
        })
        .await;
}

Implementation Hints:

1. Use stream.map(|event| { ... }) to transform
2. Normalize: (value - min) / (max - min), handle divide by zero
3. Severity: Use if/else or match on ranges
4. Filter: stream.filter(|pe| matches!(pe.severity, Severity::High | Severity::Critical))
5. Chain combinators: stream.map(...).filter(...).map(...)

Checkpoint Tests

#![allow(unused)]
fn main() {
#[test]
fn test_normalize_value() {
    assert_eq!(normalize_value(50.0, 0.0, 100.0), 0.5);
    assert_eq!(normalize_value(0.0, 0.0, 100.0), 0.0);
    assert_eq!(normalize_value(100.0, 0.0, 100.0), 1.0);
    assert_eq!(normalize_value(75.0, 50.0, 100.0), 0.5);
}

#[test]
fn test_calculate_severity() {
    assert!(matches!(calculate_severity(0.2), Severity::Low));
    assert!(matches!(calculate_severity(0.5), Severity::Medium));
    assert!(matches!(calculate_severity(0.8), Severity::High));
    assert!(matches!(calculate_severity(0.95), Severity::Critical));
}

#[tokio::test]
async fn test_stream_map() {
    use tokio_stream::{self as stream};

    let events = vec![
        Event::new(1, "s1".into(), 10.0, EventType::Metric),
        Event::new(2, "s1".into(), 20.0, EventType::Metric),
    ];

    let result: Vec<f64> = stream::iter(events)
        .map(|e| e.value * 2.0)
        .collect()
        .await;

    assert_eq!(result, vec![20.0, 40.0]);
}

#[tokio::test]
async fn test_stream_filter() {
    use tokio_stream::{self as stream};

    let events = vec![
        Event::new(1, "s1".into(), 10.0, EventType::Metric),
        Event::new(2, "s1".into(), 50.0, EventType::Metric),
        Event::new(3, "s1".into(), 100.0, EventType::Metric),
    ];

    let result: Vec<Event> = stream::iter(events)
        .filter(|e| e.value > 30.0)
        .collect()
        .await;

    assert_eq!(result.len(), 2);
    assert_eq!(result[0].value, 50.0);
}

#[tokio::test]
async fn test_process_stream() {
    use tokio_stream::{self as stream};

    let events = vec![
        Event::new(1, "s1".into(), 10.0, EventType::Metric),
        Event::new(2, "s1".into(), 90.0, EventType::Metric),
    ];

    let processed: Vec<ProcessedEvent> = process_stream(stream::iter(events))
        .collect()
        .await;

    assert_eq!(processed.len(), 2);
    assert!(processed[0].normalized_value < 0.5);
    assert!(processed[1].normalized_value > 0.5);
}
}

Milestone 3: Buffering and Batching

Introduction

Why Milestone 2 Isn’t Enough: Processing events one-at-a-time is inefficient. Database writes, network sends, and many operations have fixed overhead—batching amortizes this cost.

The Improvement: Use .chunks_timeout() to batch events by count OR time, whichever comes first. This balances throughput (batch size) with latency (timeout).

Performance (Optimization): Writing to database 1 event at a time = 1000 writes/sec (1ms per write). Batching 100 events = 100,000 events/sec with same 1ms latency per batch (100x throughput improvement).

Architecture

Structs:

• EventBatch - Collection of events for batch processing
  • Field events: Vec<ProcessedEvent> - Events in this batch
  • Field batch_id: u64 - Unique batch identifier
  • Field created_at: u64 - When batch was created

Key Functions:

• fn batch_events(stream: impl Stream<Item = ProcessedEvent>, size: usize, timeout: Duration) -> impl Stream<Item = EventBatch> - Creates batches
• async fn process_batch(batch: EventBatch) - Handles a complete batch
• async fn write_batch_to_storage(batch: &EventBatch) - Simulates batch write

Role Each Plays:

• chunks_timeout: Buffers items until N items OR timeout, yields Vec
• EventBatch: Encapsulates batch metadata + events
• Batch processing: Amortizes fixed costs (DB connection, network roundtrip)

Starter Code

use futures::stream::StreamExt; // Note: different StreamExt
use tokio::time::Duration;

#[derive(Debug, Clone)]
pub struct EventBatch {
    pub events: Vec<ProcessedEvent>,
    pub batch_id: u64,
    pub created_at: u64,
}

impl EventBatch {
    pub fn new(batch_id: u64, events: Vec<ProcessedEvent>) -> Self {
        // TODO: Create batch with current timestamp

        todo!("Implement EventBatch::new")
    }
}

pub fn batch_events<S>(
    stream: S,
    size: usize,
    timeout: Duration,
) -> impl Stream<Item = EventBatch>
where
    S: Stream<Item = ProcessedEvent>,
{
    // TODO: Use .chunks_timeout(size, timeout) to create batches
    // TODO: Map Vec<ProcessedEvent> to EventBatch with unique IDs

    todo!("Implement batching")
}

pub async fn process_batch(batch: EventBatch) {
    // TODO: Simulate batch processing
    // TODO: Print batch info (id, size, avg severity)
    // TODO: Call write_batch_to_storage

    todo!("Implement batch processing")
}

pub async fn write_batch_to_storage(batch: &EventBatch) {
    // TODO: Simulate database write (sleep for 10ms)
    // TODO: In real code, this would be DB INSERT with prepared statement

    todo!("Implement batch write simulation")
}

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(1000);

    // High-rate event generator
    tokio::spawn(generate_events(tx, 1000, 5));

    let stream = ReceiverStream::new(rx);
    let processed = process_stream(stream);

    let batched = batch_events(processed, 50, Duration::from_millis(200));

    batched
        .for_each(|batch| async move {
            process_batch(batch).await;
        })
        .await;
}

Implementation Hints:

1. Use use futures::stream::StreamExt; for chunks_timeout (different from tokio_stream)
2. Use atomic counter for batch IDs: static BATCH_ID: AtomicU64 = AtomicU64::new(0);
3. .chunks_timeout(size, timeout) returns Stream<Item = Vec<T>>
4. Use .enumerate() or atomic to generate batch IDs
5. Batch processing: sum values, count by severity, etc.

Checkpoint Tests

#![allow(unused)]
fn main() {
#[tokio::test]
async fn test_batch_by_count() {
    use tokio_stream::{self as stream, StreamExt};

    let events: Vec<i32> = (0..10).collect();
    let batches: Vec<Vec<i32>> = stream::iter(events)
        .chunks_timeout(3, Duration::from_secs(10))
        .collect()
        .await;

    assert_eq!(batches.len(), 4); // [0,1,2], [3,4,5], [6,7,8], [9]
    assert_eq!(batches[0].len(), 3);
    assert_eq!(batches[3].len(), 1);
}

#[tokio::test]
async fn test_batch_by_timeout() {
    use tokio_stream::{self as stream, StreamExt};

    let (tx, rx) = mpsc::channel(10);

    tokio::spawn(async move {
        // Send 2 events, then wait (batch should timeout)
        tx.send(1).await.unwrap();
        tx.send(2).await.unwrap();
        sleep(Duration::from_millis(150)).await;
        tx.send(3).await.unwrap();
    });

    let batches: Vec<Vec<i32>> = ReceiverStream::new(rx)
        .chunks_timeout(10, Duration::from_millis(100))
        .take(2)
        .collect()
        .await;

    assert_eq!(batches.len(), 2);
    assert_eq!(batches[0].len(), 2); // Timed out after 100ms
    assert_eq!(batches[1].len(), 1);
}

#[tokio::test]
async fn test_event_batch_creation() {
    let events = vec![
        ProcessedEvent {
            original: Event::new(1, "s1".into(), 10.0, EventType::Metric),
            normalized_value: 0.1,
            severity: Severity::Low,
        },
    ];

    let batch = EventBatch::new(1, events);

    assert_eq!(batch.batch_id, 1);
    assert_eq!(batch.events.len(), 1);
}

#[tokio::test]
async fn test_batch_processing() {
    let events = vec![
        ProcessedEvent {
            original: Event::new(1, "s1".into(), 10.0, EventType::Metric),
            normalized_value: 0.1,
            severity: Severity::Low,
        },
    ];

    let batch = EventBatch::new(1, events);

    // Should not panic
    process_batch(batch).await;
}
}

Milestone 4: Windowed Aggregations

Introduction

Why Milestone 3 Isn’t Enough: Real-time analytics require time-based aggregations—events per second, moving averages, anomaly detection. Batching by count doesn’t respect time boundaries.

The Improvement: Implement tumbling windows (non-overlapping time intervals) and sliding windows (overlapping intervals) for time-based aggregations.

Optimization (Accuracy): Time-based windows ensure accurate rate calculations. “Events per second” with count-based batching is inaccurate—batch size varies with event rate. Time windows guarantee consistent intervals.

Architecture

Structs:

• WindowedStats - Aggregated statistics for a time window
  • Field window_start: u64 - Window start timestamp
  • Field window_end: u64 - Window end timestamp
  • Field event_count: usize - Events in window
  • Field avg_value: f64 - Average event value
  • Field max_value: f64 - Maximum value
  • Field severity_counts: HashMap<Severity, usize> - Counts by severity

Key Functions:

• fn create_tumbling_window(stream: impl Stream<Item = ProcessedEvent>, duration_ms: u64) -> impl Stream<Item = WindowedStats> - Creates fixed windows
• fn calculate_window_stats(events: Vec<ProcessedEvent>, window_start: u64, duration_ms: u64) -> WindowedStats - Aggregates events
• async fn detect_anomalies(stats: &WindowedStats, baseline: &WindowedStats) -> Option<Alert> - Compares windows

Role Each Plays:

• Tumbling window: Divides time into non-overlapping intervals (0-1s, 1-2s, 2-3s)
• Window stats: Aggregates metrics for analysis
• Anomaly detection: Compares current window to historical baseline

Starter Code

use std::collections::HashMap;

#[derive(Debug, Clone)]
pub struct WindowedStats {
    pub window_start: u64,
    pub window_end: u64,
    pub event_count: usize,
    pub avg_value: f64,
    pub max_value: f64,
    pub severity_counts: HashMap<Severity, usize>,
}

#[derive(Debug, Clone)]
pub struct Alert {
    pub message: String,
    pub severity: Severity,
    pub window: WindowedStats,
}

pub fn calculate_window_stats(
    events: Vec<ProcessedEvent>,
    window_start: u64,
    duration_ms: u64,
) -> WindowedStats {
    // TODO: Calculate aggregate statistics
    // TODO: Count events by severity
    // TODO: Compute avg and max values

    todo!("Implement window stats calculation")
}

pub fn create_tumbling_window<S>(
    stream: S,
    duration_ms: u64,
) -> impl Stream<Item = WindowedStats>
where
    S: Stream<Item = ProcessedEvent>,
{
    // TODO: Group events by time window
    // TODO: Use .chunks_timeout or manual window tracking
    // TODO: For each window, calculate stats

    todo!("Implement tumbling window")
}

pub async fn detect_anomalies(
    current: &WindowedStats,
    baseline: &WindowedStats,
) -> Option<Alert> {
    // TODO: Compare current window to baseline
    // TODO: Check if event_count > 2x baseline (spike)
    // TODO: Check if avg_value > 1.5x baseline (abnormal values)
    // TODO: Return Alert if anomaly detected

    todo!("Implement anomaly detection")
}

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(1000);

    tokio::spawn(generate_events(tx, 10000, 10));

    let stream = ReceiverStream::new(rx);
    let processed = process_stream(stream);

    let windows = create_tumbling_window(processed, 1000); // 1-second windows

    let mut baseline: Option<WindowedStats> = None;

    windows
        .for_each(|stats| async {
            println!(
                "Window [{} - {}]: {} events, avg={:.2}, max={:.2}",
                stats.window_start, stats.window_end, stats.event_count, stats.avg_value, stats.max_value
            );

            if let Some(ref base) = baseline {
                if let Some(alert) = detect_anomalies(&stats, base).await {
                    println!("ALERT: {}", alert.message);
                }
            } else {
                baseline = Some(stats.clone());
            }
        })
        .await;
}

Implementation Hints:

1. Window assignment: let window_id = event.timestamp / duration_ms;
2. Use HashMap<u64, Vec<ProcessedEvent>> to group by window
3. For tumbling: emit window when next event is in different window
4. Stats: use .fold() or manual accumulation
5. Anomaly: current.event_count > baseline.event_count * 2

Checkpoint Tests

#![allow(unused)]
fn main() {
#[tokio::test]
async fn test_tumbling_window() {
    use tokio_stream::{self as stream};

    // Create events with timestamps 0, 500, 1000, 1500, 2000 ms
    let mut events = Vec::new();
    for i in 0..5 {
        let mut event = Event::new(i, "s1".into(), (i * 10) as f64, EventType::Metric);
        event.timestamp = i * 500;
        events.push(ProcessedEvent {
            original: event,
            normalized_value: 0.5,
            severity: Severity::Medium,
        });
    }

    let windows: Vec<WindowedStats> = create_tumbling_window(
        stream::iter(events),
        1000, // 1-second windows
    )
    .collect()
    .await;

    // Should create 3 windows: [0-1000), [1000-2000), [2000-3000)
    assert!(windows.len() >= 2);
}

#[test]
fn test_calculate_window_stats() {
    let events = vec![
        ProcessedEvent {
            original: Event::new(1, "s1".into(), 10.0, EventType::Metric),
            normalized_value: 0.1,
            severity: Severity::Low,
        },
        ProcessedEvent {
            original: Event::new(2, "s1".into(), 30.0, EventType::Metric),
            normalized_value: 0.3,
            severity: Severity::Medium,
        },
    ];

    let stats = calculate_window_stats(events, 0, 1000);

    assert_eq!(stats.event_count, 2);
    assert_eq!(stats.avg_value, 20.0); // (10 + 30) / 2
    assert_eq!(stats.max_value, 30.0);
}

#[tokio::test]
async fn test_anomaly_detection() {
    let baseline = WindowedStats {
        window_start: 0,
        window_end: 1000,
        event_count: 100,
        avg_value: 50.0,
        max_value: 80.0,
        severity_counts: HashMap::new(),
    };

    let normal = WindowedStats {
        window_start: 1000,
        window_end: 2000,
        event_count: 105, // Within tolerance
        avg_value: 52.0,
        max_value: 82.0,
        severity_counts: HashMap::new(),
    };

    let anomaly = WindowedStats {
        window_start: 2000,
        window_end: 3000,
        event_count: 500, // 5x normal
        avg_value: 95.0,  // Much higher
        max_value: 120.0,
        severity_counts: HashMap::new(),
    };

    assert!(detect_anomalies(&normal, &baseline).await.is_none());
    assert!(detect_anomalies(&anomaly, &baseline).await.is_some());
}
}

Milestone 5: Backpressure Handling

Introduction

Why Milestone 4 Isn’t Enough: Fast producers can overwhelm slow consumers. Without backpressure, memory usage grows unbounded → OOM crash. Bounded channels provide backpressure but can still accumulate events.

The Improvement: Implement explicit backpressure strategies—drop oldest, drop newest, or apply sampling when consumer falls behind.

Optimization (Memory): Unbounded buffer with 10,000 events/sec input, 1,000 events/sec processing = 9,000 events/sec accumulation. At 1KB/event, that’s 9MB/sec → 540MB/minute → crash. Bounded buffer + drop strategy keeps memory constant.

Architecture

Structs:

• BackpressureConfig - Configuration for backpressure handling

  • Field strategy: BackpressureStrategy - How to handle overload
  • Field buffer_size: usize - Maximum buffered events
  • Field sample_rate: f64 - Sampling probability (0.0-1.0)

• BackpressureStrategy - Enum for strategies

  • Variant DropOldest - Remove oldest events when full
  • Variant DropNewest - Reject new events when full
  • Variant Sample - Randomly sample events

• BackpressureStats - Metrics about drops

  • Field received: AtomicUsize - Total events received
  • Field dropped: AtomicUsize - Events dropped
  • Field processed: AtomicUsize - Events processed

Key Functions:

• async fn apply_backpressure(stream: impl Stream<Item = Event>, config: BackpressureConfig) -> (impl Stream<Item = Event>, BackpressureStats) - Wraps stream with backpressure
• fn should_sample(rate: f64) -> bool - Random sampling decision
• async fn monitor_backpressure(stats: Arc<BackpressureStats>) - Reports drop rate

Role Each Plays:

• Bounded channel: Provides natural backpressure (send blocks when full)
• Drop strategies: Choose which events to discard under load
• Sampling: Reduces load while maintaining statistical properties

Starter Code

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use rand::Rng;

#[derive(Debug, Clone)]
pub enum BackpressureStrategy {
    DropOldest,
    DropNewest,
    Sample,
}

#[derive(Debug, Clone)]
pub struct BackpressureConfig {
    pub strategy: BackpressureStrategy,
    pub buffer_size: usize,
    pub sample_rate: f64,
}

pub struct BackpressureStats {
    received: AtomicUsize,
    dropped: AtomicUsize,
    processed: AtomicUsize,
}

impl BackpressureStats {
    pub fn new() -> Self {
        // TODO: Initialize atomic counters

        todo!("Implement BackpressureStats::new")
    }

    pub fn increment_received(&self) {
        self.received.fetch_add(1, Ordering::Relaxed);
    }

    pub fn increment_dropped(&self) {
        self.dropped.fetch_add(1, Ordering::Relaxed);
    }

    pub fn increment_processed(&self) {
        self.processed.fetch_add(1, Ordering::Relaxed);
    }

    pub fn received(&self) -> usize {
        self.received.load(Ordering::Relaxed)
    }

    pub fn dropped(&self) -> usize {
        self.dropped.load(Ordering::Relaxed)
    }

    pub fn processed(&self) -> usize {
        self.processed.load(Ordering::Relaxed)
    }

    pub fn get_report(&self) -> String {
        // TODO: Format statistics report
        // Include drop rate percentage

        todo!("Implement stats report")
    }
}

pub fn should_sample(rate: f64) -> bool {
    // TODO: Return true with probability 'rate'
    // Hint: rand::thread_rng().gen::<f64>() < rate

    todo!("Implement sampling decision")
}

pub fn apply_backpressure<S>(
    stream: S,
    config: BackpressureConfig,
) -> (impl Stream<Item = Event>, Arc<BackpressureStats>)
where
    S: Stream<Item = Event>,
{
    // TODO: Wrap stream with backpressure logic
    // TODO: Track events in stats
    // TODO: Apply strategy (drop or sample)

    todo!("Implement backpressure")
}

pub async fn monitor_backpressure(stats: Arc<BackpressureStats>) {
    // TODO: Periodically print stats
    // TODO: Loop with sleep, print every second

    todo!("Implement monitoring")
}

#[tokio::main]
async fn main() {
    let config = BackpressureConfig {
        strategy: BackpressureStrategy::Sample,
        buffer_size: 100,
        sample_rate: 0.1,
    };

    let (tx, rx) = mpsc::channel(1000);

    // Very fast event generator
    tokio::spawn(generate_events(tx, 100000, 1));

    let stream = ReceiverStream::new(rx);
    let (backpressured, stats) = apply_backpressure(stream, config);

    // Spawn monitor
    let stats_clone = Arc::clone(&stats);
    tokio::spawn(monitor_backpressure(stats_clone));

    let processed = process_stream(backpressured);

    // Slow consumer (simulates heavy processing)
    processed
        .for_each(|event| async move {
            sleep(Duration::from_millis(10)).await; // Slow processing
            // Process event...
        })
        .await;

    println!("\nFinal stats: {}", stats.get_report());
}

Implementation Hints:

1. Use VecDeque with capacity for DropOldest buffer
2. For DropOldest: if buffer.len() == capacity { buffer.pop_front(); }
3. For Sample: stream.filter(|_| should_sample(rate))
4. Wrap stream in struct with stats tracking
5. Use .inspect() combinator to track events

Checkpoint Tests

#![allow(unused)]
fn main() {
#[test]
fn test_should_sample() {
    // Sample at 50% - roughly half should pass
    let mut passed = 0;
    for _ in 0..1000 {
        if should_sample(0.5) {
            passed += 1;
        }
    }

    // Should be around 500 (allow some variance)
    assert!(passed > 400 && passed < 600);
}

#[tokio::test]
async fn test_drop_oldest_strategy() {
    let config = BackpressureConfig {
        strategy: BackpressureStrategy::DropOldest,
        buffer_size: 5,
        sample_rate: 1.0,
    };

    let (tx, rx) = mpsc::channel(100);

    // Send 10 events quickly
    for i in 0..10 {
        tx.send(Event::new(i, "s1".into(), i as f64, EventType::Metric))
            .await
            .unwrap();
    }
    drop(tx);

    let stream = ReceiverStream::new(rx);
    let (processed_stream, stats) = apply_backpressure(stream, config);

    let events: Vec<Event> = processed_stream.collect().await;

    // Should keep last 5 events (dropped first 5)
    assert_eq!(stats.received.load(Ordering::Relaxed), 10);
    assert_eq!(stats.dropped.load(Ordering::Relaxed), 5);
}

#[tokio::test]
async fn test_sampling_strategy() {
    let config = BackpressureConfig {
        strategy: BackpressureStrategy::Sample,
        buffer_size: 1000,
        sample_rate: 0.1, // Keep 10%
    };

    let events: Vec<Event> = (0..1000)
        .map(|i| Event::new(i, "s1".into(), i as f64, EventType::Metric))
        .collect();

    let stream = tokio_stream::iter(events);
    let (processed_stream, stats) = apply_backpressure(stream, config);

    let result: Vec<Event> = processed_stream.collect().await;

    // Should keep roughly 10% (allow variance)
    assert!(result.len() > 50 && result.len() < 150);
    assert_eq!(stats.received.load(Ordering::Relaxed), 1000);
}

#[tokio::test]
async fn test_backpressure_stats() {
    let stats = BackpressureStats::new();

    stats.increment_received();
    stats.increment_received();
    stats.increment_dropped();
    stats.increment_processed();

    assert_eq!(stats.received(), 2);
    assert_eq!(stats.dropped(), 1);
    assert_eq!(stats.processed(), 1);

    let report = stats.get_report();
    assert!(report.contains("50.0%")); // 1 dropped out of 2
}
}

Milestone 6: Multi-Source Stream Merging and Fan-Out

Introduction

Why Milestone 5 Isn’t Enough: Real systems have multiple event sources (sensors, logs, APIs). Processing them separately duplicates code. We need to merge streams and distribute results.

The Improvement: Implement stream merging (combine multiple sources) and fan-out (send results to multiple consumers). Use select_all for merging and broadcast channels for fan-out.

Optimization (Parallelism): Fan-out enables parallel processing—same events processed by analytics engine AND alerting system simultaneously. Without fan-out, you’d process sequentially (2x latency) or duplicate the event stream (2x bandwidth).

Architecture

Structs:

• StreamSource - Identifies event source

  • Field id: String - Source identifier
  • Field source_type: SourceType - Category

• SourceType - Enum for source types

  • Variant Sensor(String), LogFile(String), API(String)

• ProcessingPipeline - Manages multi-stream processing

  • Field sources: Vec<StreamSource> - Active sources
  • Field outputs: Vec<broadcast::Sender<ProcessedEvent>> - Output channels

Key Functions:

• fn merge_streams(streams: Vec<impl Stream<Item = Event>>) -> impl Stream<Item = Event> - Combines multiple streams
• fn create_fanout(input: impl Stream<Item = ProcessedEvent>, output_count: usize) -> Vec<broadcast::Receiver<ProcessedEvent>> - Distributes to multiple outputs
• async fn run_pipeline(sources: Vec<impl Stream<Item = Event>>) -> ProcessingPipeline - Full pipeline

Role Each Plays:

• select_all: Merges streams, yields items as they arrive (unordered)
• broadcast channel: One sender, many receivers (all get same events)
• Fan-out: Enables parallel consumers without duplication

Starter Code

use tokio::sync::broadcast;
use tokio_stream::StreamExt;

#[derive(Debug, Clone)]
pub enum SourceType {
    Sensor(String),
    LogFile(String),
    API(String),
}

#[derive(Debug, Clone)]
pub struct StreamSource {
    pub id: String,
    pub source_type: SourceType,
}

pub struct ProcessingPipeline {
    pub sources: Vec<StreamSource>,
    pub broadcast_tx: broadcast::Sender<ProcessedEvent>,
}

pub fn merge_streams<S>(streams: Vec<S>) -> impl Stream<Item = Event>
where
    S: Stream<Item = Event> + Send + 'static,
{
    // TODO: Use StreamExt::merge or select_all to combine streams
    // Hint: futures::stream::select_all(streams)

    todo!("Implement stream merging")
}

pub fn create_fanout(
    input: impl Stream<Item = ProcessedEvent> + Send + 'static,
    output_count: usize,
) -> (tokio::task::JoinHandle<()>, Vec<broadcast::Receiver<ProcessedEvent>>) {
    // TODO: Create broadcast channel
    // TODO: Spawn task to forward input stream to broadcast
    // TODO: Create N receivers
    // TODO: Return handle and receivers

    todo!("Implement fan-out")
}

pub async fn run_pipeline(
    sources: Vec<(StreamSource, impl Stream<Item = Event> + Send + 'static)>,
) -> ProcessingPipeline {
    // TODO: Extract streams from sources
    // TODO: Merge all streams
    // TODO: Process merged stream
    // TODO: Create fan-out for processed events
    // TODO: Return pipeline with broadcast sender

    todo!("Implement full pipeline")
}

#[tokio::main]
async fn main() {
    // Create multiple event sources
    let mut sources = Vec::new();

    for i in 0..3 {
        let (tx, rx) = mpsc::channel(100);
        let source_id = format!("sensor-{}", i);

        tokio::spawn({
            let source = source_id.clone();
            async move {
                generate_events(tx, 100, 50).await;
            }
        });

        sources.push((
            StreamSource {
                id: source_id.clone(),
                source_type: SourceType::Sensor(source_id),
            },
            ReceiverStream::new(rx),
        ));
    }

    let pipeline = run_pipeline(sources).await;

    // Create multiple consumers
    let mut consumer1 = pipeline.broadcast_tx.subscribe();
    let mut consumer2 = pipeline.broadcast_tx.subscribe();

    // Consumer 1: Count events
    let counter = tokio::spawn(async move {
        let mut count = 0;
        while consumer1.recv().await.is_ok() {
            count += 1;
        }
        println!("Consumer 1 processed: {} events", count);
    });

    // Consumer 2: Alert on high severity
    let alerter = tokio::spawn(async move {
        while let Ok(event) = consumer2.recv().await {
            if matches!(event.severity, Severity::High | Severity::Critical) {
                println!("ALERT: High severity event from {}", event.original.source);
            }
        }
        println!("Consumer 2 done");
    });

    tokio::join!(counter, alerter);
}

Implementation Hints:

1. Use futures::stream::select_all(streams) for merging
2. For fan-out: let (tx, _rx) = broadcast::channel(capacity);
3. Subscribe: let rx = tx.subscribe();
4. Forward stream: while let Some(item) = stream.next().await { tx.send(item)?; }
5. Handle lagging receivers: broadcast drops oldest if receiver falls behind

Checkpoint Tests

#![allow(unused)]
fn main() {
#[tokio::test]
async fn test_merge_streams() {
    use tokio_stream::{self as stream};

    let stream1 = stream::iter(vec![1, 2, 3]);
    let stream2 = stream::iter(vec![4, 5, 6]);
    let stream3 = stream::iter(vec![7, 8, 9]);

    let merged = merge_streams(vec![stream1, stream2, stream3]);
    let results: Vec<i32> = merged.collect().await;

    assert_eq!(results.len(), 9);
    assert!(results.contains(&1));
    assert!(results.contains(&9));
}

#[tokio::test]
async fn test_fanout() {
    use tokio::sync::broadcast;

    let (tx, _) = broadcast::channel(100);

    let rx1 = tx.subscribe();
    let rx2 = tx.subscribe();
    let rx3 = tx.subscribe();

    // Send events
    for i in 0..5 {
        tx.send(i).unwrap();
    }
    drop(tx);

    // All receivers should get all events
    let results1: Vec<i32> = ReceiverStream::new(rx1).collect().await;
    let results2: Vec<i32> = ReceiverStream::new(rx2).collect().await;
    let results3: Vec<i32> = ReceiverStream::new(rx3).collect().await;

    assert_eq!(results1, results2);
    assert_eq!(results2, results3);
    assert_eq!(results1.len(), 5);
}

#[tokio::test]
async fn test_multi_source_processing() {
    // Create 3 event sources
    let (tx1, rx1) = mpsc::channel(10);
    let (tx2, rx2) = mpsc::channel(10);
    let (tx3, rx3) = mpsc::channel(10);

    tokio::spawn(async move {
        for i in 0..5 {
            tx1.send(Event::new(i, "source1".into(), i as f64, EventType::Metric))
                .await
                .unwrap();
        }
    });

    tokio::spawn(async move {
        for i in 5..10 {
            tx2.send(Event::new(i, "source2".into(), i as f64, EventType::Log))
                .await
                .unwrap();
        }
    });

    tokio::spawn(async move {
        for i in 10..15 {
            tx3.send(Event::new(i, "source3".into(), i as f64, EventType::Alert))
                .await
                .unwrap();
        }
    });

    let streams = vec![
        ReceiverStream::new(rx1),
        ReceiverStream::new(rx2),
        ReceiverStream::new(rx3),
    ];

    let merged = merge_streams(streams);
    let events: Vec<Event> = merged.collect().await;

    assert_eq!(events.len(), 15);
}
}

Complete Working Example

// Cargo.toml:
// [dependencies]
// tokio = { version = "1.35", features = ["full"] }
// tokio-stream = "0.1"
// futures = "0.3"
// serde = { version = "1.0", features = ["derive"] }
// rand = "0.8"

use tokio::sync::{mpsc, broadcast};
use tokio::time::{sleep, Duration};
use tokio_stream::{Stream, StreamExt};
use tokio_stream::wrappers::ReceiverStream;
use futures::stream;
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

// Event types
#[derive(Debug, Clone, PartialEq)]
pub enum EventType {
    Metric,
    Log,
    Alert,
    Error,
}

#[derive(Debug, Clone)]
pub struct Event {
    pub id: u64,
    pub timestamp: u64,
    pub source: String,
    pub value: f64,
    pub event_type: EventType,
}

impl Event {
    pub fn new(id: u64, source: String, value: f64, event_type: EventType) -> Self {
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;

        Self {
            id,
            timestamp,
            source,
            value,
            event_type,
        }
    }
}

// Processing types
#[derive(Debug, Clone, PartialEq)]
pub enum Severity {
    Low,
    Medium,
    High,
    Critical,
}

#[derive(Debug, Clone)]
pub struct ProcessedEvent {
    pub original: Event,
    pub normalized_value: f64,
    pub severity: Severity,
}

// Backpressure types
pub struct BackpressureStats {
    received: AtomicUsize,
    dropped: AtomicUsize,
    processed: AtomicUsize,
}

impl BackpressureStats {
    pub fn new() -> Self {
        Self {
            received: AtomicUsize::new(0),
            dropped: AtomicUsize::new(0),
            processed: AtomicUsize::new(0),
        }
    }

    pub fn increment_received(&self) {
        self.received.fetch_add(1, Ordering::Relaxed);
    }

    pub fn increment_dropped(&self) {
        self.dropped.fetch_add(1, Ordering::Relaxed);
    }

    pub fn increment_processed(&self) {
        self.processed.fetch_add(1, Ordering::Relaxed);
    }

    pub fn get_report(&self) -> String {
        let received = self.received.load(Ordering::Relaxed);
        let dropped = self.dropped.load(Ordering::Relaxed);
        let processed = self.processed.load(Ordering::Relaxed);
        let drop_rate = if received > 0 {
            (dropped as f64 / received as f64) * 100.0
        } else {
            0.0
        };

        format!(
            "Received: {}, Processed: {}, Dropped: {} ({:.1}% drop rate)",
            received, processed, dropped, drop_rate
        )
    }
}

// Event generation
pub async fn generate_events(tx: mpsc::Sender<Event>, count: usize, delay_ms: u64) {
    for i in 0..count {
        let source = format!("sensor-{}", i % 5);
        let value = (i as f64 * 3.7) % 100.0;
        let event_type = match i % 4 {
            0 => EventType::Metric,
            1 => EventType::Log,
            2 => EventType::Alert,
            _ => EventType::Error,
        };

        let event = Event::new(i as u64, source, value, event_type);

        if tx.send(event).await.is_err() {
            break;
        }

        sleep(Duration::from_millis(delay_ms)).await;
    }
}

// Processing functions
pub fn normalize_value(value: f64, min: f64, max: f64) -> f64 {
    if (max - min).abs() < 0.001 {
        0.5
    } else {
        (value - min) / (max - min)
    }
}

pub fn calculate_severity(normalized: f64) -> Severity {
    match normalized {
        x if x < 0.3 => Severity::Low,
        x if x < 0.6 => Severity::Medium,
        x if x < 0.9 => Severity::High,
        _ => Severity::Critical,
    }
}

pub fn process_event(event: Event) -> ProcessedEvent {
    let normalized = normalize_value(event.value, 0.0, 100.0);
    let severity = calculate_severity(normalized);

    ProcessedEvent {
        original: event,
        normalized_value: normalized,
        severity,
    }
}

// Stream merging
pub fn merge_event_streams<S>(streams: Vec<S>) -> impl Stream<Item = Event>
where
    S: Stream<Item = Event> + Send + 'static,
{
    stream::select_all(streams)
}

// Windowed aggregation
#[derive(Debug, Clone)]
pub struct WindowedStats {
    pub window_start: u64,
    pub window_end: u64,
    pub event_count: usize,
    pub avg_value: f64,
    pub max_value: f64,
}

impl WindowedStats {
    pub fn from_events(events: Vec<ProcessedEvent>, window_start: u64, duration_ms: u64) -> Self {
        let count = events.len();
        let avg = if count > 0 {
            events.iter().map(|e| e.original.value).sum::<f64>() / count as f64
        } else {
            0.0
        };
        let max = events
            .iter()
            .map(|e| e.original.value)
            .fold(0.0, f64::max);

        Self {
            window_start,
            window_end: window_start + duration_ms,
            event_count: count,
            avg_value: avg,
            max_value: max,
        }
    }
}

// Complete pipeline
pub async fn run_complete_pipeline() {
    println!("=== Real-Time Event Stream Processor ===\n");

    // Create multiple event sources
    let mut stream_handles = Vec::new();

    for i in 0..3 {
        let (tx, rx) = mpsc::channel(100);

        tokio::spawn(async move {
            generate_events(tx, 50, 20 + i * 10).await;
        });

        stream_handles.push(ReceiverStream::new(rx));
    }

    // Merge all sources
    let merged = merge_event_streams(stream_handles);

    // Process events
    let processed = merged.map(process_event);

    // Apply backpressure with sampling
    let stats = Arc::new(BackpressureStats::new());
    let stats_clone = Arc::clone(&stats);

    let with_backpressure = processed.filter(move |event| {
        stats_clone.increment_received();

        // Sample at 50% under load
        let should_process = rand::random::<f64>() < 0.5;

        if should_process {
            stats_clone.increment_processed();
        } else {
            stats_clone.increment_dropped();
        }

        should_process
    });

    // Create fan-out for multiple consumers
    let (broadcast_tx, _) = broadcast::channel(1000);
    let mut rx1 = broadcast_tx.subscribe();
    let mut rx2 = broadcast_tx.subscribe();

    // Forward processed events to broadcast
    let broadcast_clone = broadcast_tx.clone();
    tokio::spawn(async move {
        tokio::pin!(with_backpressure);

        while let Some(event) = with_backpressure.next().await {
            let _ = broadcast_clone.send(event);
        }
    });

    // Consumer 1: Count by severity
    let counter = tokio::spawn(async move {
        let mut counts: HashMap<String, usize> = HashMap::new();

        while let Ok(event) = rx1.recv().await {
            let severity = format!("{:?}", event.severity);
            *counts.entry(severity).or_insert(0) += 1;
        }

        println!("\nSeverity counts:");
        for (sev, count) in counts {
            println!("  {}: {}", sev, count);
        }
    });

    // Consumer 2: Alert on critical
    let alerter = tokio::spawn(async move {
        let mut alert_count = 0;

        while let Ok(event) = rx2.recv().await {
            if matches!(event.severity, Severity::Critical) {
                alert_count += 1;
                println!(
                    "CRITICAL ALERT: {} = {:.2}",
                    event.original.source, event.original.value
                );
            }
        }

        println!("\nTotal critical alerts: {}", alert_count);
    });

    // Wait for all consumers
    tokio::join!(counter, alerter);

    println!("\n{}", stats.get_report());
}

#[tokio::main]
async fn main() {
    run_complete_pipeline().await;
}

Project-Wide Benefits

Async stream patterns enable high-throughput, low-latency data processing:

End-to-end performance (10,000 events/sec, 3 sources, 3 consumers):

Real-world production metrics:

• Throughput: 100K-1M events/sec (depends on processing complexity)
• Memory: 5-50MB (constant, regardless of stream length)
• Latency: 100-500ms (tunable via batch timeout)
• Error rate: < 0.1% (with backpressure and retries)

Comparison to other approaches:

When to use this approach:

• ✅ Real-time data processing (sensors, logs, metrics)
• ✅ High-throughput pipelines (100K+ events/sec)
• ✅ Multi-source aggregation (merge streams)
• ✅ Multi-consumer fan-out (broadcast)
• ✅ Backpressure critical (prevent OOM)
• ❌ Batch processing (use Rayon for parallel batches)
• ❌ Simple scripts (overkill for low volume)