mecha10-core 0.8.2

//! Example usage of backpressure handling for Receivers
//!
//! This example demonstrates how to use bounded channels to prevent memory
//! exhaustion when publishers send messages faster than subscribers can process them.

use mecha10::prelude::*;
use std::time::Duration;

/// Example 6: Backpressure monitoring and alerting
///
/// Monitor queue depth and trigger alerts when thresholds are exceeded.
pub async fn monitoring_example(ctx: &Context) -> Result<()> {
    info!("Example 6: Backpressure monitoring");

    let mut images = ctx.subscribe(sensor::CAMERA_RGB).await?.with_capacity(100);

    let monitor_interval = Duration::from_secs(1);
    let mut monitor = tokio::time::interval(monitor_interval);

    tokio::spawn(async move {
        loop {
            monitor.tick().await;

            let depth = images.len();
            let cap = images.capacity().unwrap_or(0);
            let util = images.utilization().unwrap_or(0.0);

            info!(
                "Queue metrics - Depth: {}, Capacity: {}, Utilization: {:.1}%",
                depth,
                cap,
                util * 100.0
            );

            if util > 0.95 {
                error!("CRITICAL: Queue at {:.1}% capacity!", util * 100.0);
            } else if util > 0.8 {
                warn!("WARNING: Queue at {:.1}% capacity", util * 100.0);
            }
        }
    });

    while let Some(image) = images.recv().await {
        process_image_normal(&image)?;
    }

    Ok(())
}

fn process_image_normal(image: &mecha10_core::types::Image) -> Result<()> {
    std::thread::sleep(Duration::from_millis(50));
    debug!("Normal: {}x{}", image.width, image.height);
    Ok(())
}

fn main() {
    println!("Backpressure Handling Examples");
    println!("===============================");
    println!();
    println!("Key Concepts:");
    println!();
    println!("1. Unbounded Channels (Default)");
    println!("   - No limit on queued messages");
    println!("   - Risk: Memory exhaustion from fast publishers");
    println!("   - Use when: Publisher and subscriber rates are balanced");
    println!();
    println!("2. Bounded Channels (.with_capacity(N))");
    println!("   - Limit to N messages in queue");
    println!("   - Backpressure applied when full");
    println!("   - Use when: Need to prevent memory exhaustion");
    println!();
    println!("3. Drop Oldest Strategy (.with_drop_oldest(N))");
    println!("   - Keep only N most recent messages");
    println!("   - Old messages dropped automatically");
    println!("   - Use when: Real-time data, recent > historical");
    println!();
    println!("4. Monitoring");
    println!("   - .len() - Current queue depth");
    println!("   - .capacity() - Maximum capacity (if bounded)");
    println!("   - .utilization() - Usage percentage (0.0-1.0)");
    println!();
    println!("Best Practices:");
    println!("- Sensor data: Use .with_drop_oldest() for latest values");
    println!("- Command queues: Use .with_capacity() to prevent loss");
    println!("- Monitor utilization and adjust processing speed");
    println!("- Alert when utilization > 80% (approaching backpressure)");
    println!();
    println!("Examples:");
    println!("  cargo run --example backpressure_unbounded");
    println!("  cargo run --example backpressure_bounded");
    println!("  cargo run --example backpressure_drop_oldest");
    println!("  cargo run --example backpressure_adaptive");
    println!("  cargo run --example backpressure_monitoring");
}