effect-rs 0.1.0

use effect_rs::{Effect, EffectStream, Exit, Runtime};

fn process_item(n: i32) -> Effect<(), (), i32> {
    Effect::sync(move || {
        // Simulate CPU work
        let start = std::time::Instant::now();
        while start.elapsed().as_micros() < 500 {}
        n * 2
    })
}

fn main() {
    let rt = Runtime::new();

    // Source: 1000 items
    let data: Vec<i32> = (0..1000).collect();
    let stream = EffectStream::<(), (), i32>::from_iter(data);

    println!("Starting Pipeline...");
    let start = std::time::Instant::now();

    // Pipeline:
    // 1. Filter even numbers
    // 2. Map Concurrently (Process)
    // 3. Buffer (Decouple)
    // 4. Collect
    let program = stream
        .filter(|n| n % 2 == 0)
        .map_par(10, |n| process_item(n)) // 10 workers
        .buffer(100)
        .run_collect();

    let res = rt.block_on(program, ());

    match res {
        Exit::Success(items) => {
            let duration = start.elapsed();
            println!("Pipeline Completed in {:?}", duration);
            println!("Processed {} items", items.len());
            println!("Sum: {}", items.iter().sum::<i32>());

            // Validation
            // 0..1000 -> 500 evens. 500 * 2 = 1000 items? No 500 items.
            // Sum of (2x) where x in 0, 2, 4 ... 998.
            // Sum = 2 * (0 + 2 + ... + 998)
            // = 4 * (0 + 1 + ... + 499)
            // = 4 * (499 * 500 / 2) = 2 * 499 * 500 = 499000

            let expected_sum = 499000;
            if items.iter().sum::<i32>() == expected_sum {
                println!("SUCCESS: Result is correct.");
            } else {
                println!(
                    "FAILURE: Expected {}, got {}",
                    expected_sum,
                    items.iter().sum::<i32>()
                );
            }
        }
        Exit::Failure(c) => println!("Pipeline Failed: {:?}", c),
    }
}