rs2-stream 0.3.3

A high-performance, production-ready async streaming library for Rust.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

use futures_util::stream::StreamExt;
use rs2_stream::rs2::*;
use tokio::runtime::Runtime;

// Define a CPU-intensive operation
fn compute_fibonacci(n: u64) -> u64 {
    match n {
        0 => 0,
        1 => 1,
        _ => compute_fibonacci(n - 1) + compute_fibonacci(n - 2),
    }
}

fn main() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        println!("🚀 Demonstrating parallel mapping operations");

        // Create a stream of numbers
        let numbers = (20..30).collect::<Vec<u64>>();
        println!(
            "Processing Fibonacci calculations for numbers: {:?}",
            numbers
        );

        // Example 1: Using map_parallel_rs2 for automatic concurrency
        println!("\n📊 Example 1: Using map_parallel_rs2 with automatic concurrency");
        let start = std::time::Instant::now();

        let results = from_iter(numbers.clone())
            .map_parallel_rs2(|n| {
                println!(
                    "  Computing Fibonacci for {} on thread {:?}",
                    n,
                    std::thread::current().id()
                );
                compute_fibonacci(n)
            })
            .collect::<Vec<_>>()
            .await;

        let elapsed = start.elapsed();
        println!("✅ Completed in {:.2?}", elapsed);
        println!("Results: {:?}", results);

        // Example 2: Using map_parallel_with_concurrency_rs2 for custom concurrency
        println!("\n📊 Example 2: Using map_parallel_with_concurrency_rs2 with custom concurrency");
        let concurrency = 2; // Deliberately use fewer threads than available cores
        println!("  Using concurrency level: {}", concurrency);

        let start = std::time::Instant::now();

        let results = from_iter(numbers.clone())
            .map_parallel_with_concurrency_rs2(concurrency, |n| {
                println!(
                    "  Computing Fibonacci for {} on thread {:?}",
                    n,
                    std::thread::current().id()
                );
                compute_fibonacci(n)
            })
            .collect::<Vec<_>>()
            .await;

        let elapsed = start.elapsed();
        println!("✅ Completed in {:.2?}", elapsed);
        println!("Results: {:?}", results);

        // Example 3: Comparing with sequential processing
        println!("\n📊 Example 3: Sequential processing for comparison");
        let start = std::time::Instant::now();

        let results = from_iter(numbers.clone())
            .map_rs2(|n| {
                println!(
                    "  Computing Fibonacci for {} on thread {:?}",
                    n,
                    std::thread::current().id()
                );
                compute_fibonacci(n)
            })
            .collect::<Vec<_>>()
            .await;

        let elapsed = start.elapsed();
        println!("✅ Completed in {:.2?}", elapsed);
        println!("Results: {:?}", results);
    });
}