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

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

fn main() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        println!("\n=== Basic Sliding Window Example ===");

        // Create a stream of numbers
        let numbers = from_iter(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);

        // Apply a sliding window of size 3
        let windows = numbers.sliding_window_rs2(3).collect::<Vec<_>>().await;

        // Print each window
        println!("Sliding windows of size 3:");
        for (i, window) in windows.iter().enumerate() {
            println!("  Window {}: {:?}", i + 1, window);
        }

        println!("\n=== Sliding Window with Strings Example ===");

        // Create a stream of words
        let words = from_iter(vec![
            "The".to_string(),
            "quick".to_string(),
            "brown".to_string(),
            "fox".to_string(),
            "jumps".to_string(),
            "over".to_string(),
            "the".to_string(),
            "lazy".to_string(),
            "dog".to_string(),
        ]);

        // Apply a sliding window of size 4 to create phrases
        let phrases = words.sliding_window_rs2(4).collect::<Vec<_>>().await;

        // Print each phrase
        println!("Phrases (sliding windows of size 4):");
        for (i, phrase) in phrases.iter().enumerate() {
            println!("  Phrase {}: {}", i + 1, phrase.join(" "));
        }

        println!("\n=== Sliding Window for Time Series Analysis ===");

        // Create a stream of time series data (timestamp, value)
        let time_series = from_iter(vec![
            (0, 10.5),
            (1, 11.2),
            (2, 10.8),
            (3, 12.3),
            (4, 13.5),
            (5, 14.1),
            (6, 13.8),
            (7, 15.2),
            (8, 16.0),
            (9, 15.7),
        ]);

        // Apply a sliding window of size 3 for moving average calculation
        let moving_averages = time_series
            .sliding_window_rs2(3)
            .map_rs2(|window| {
                let sum: f64 = window.iter().map(|(_, value)| value).sum();
                let avg = sum / window.len() as f64;
                (window.last().unwrap().0, avg) // Use the last timestamp with the average
            })
            .collect::<Vec<_>>()
            .await;

        // Print the moving averages
        println!("Moving averages (window size 3):");
        for (timestamp, avg) in moving_averages {
            println!("  Time {}: {:.2}", timestamp, avg);
        }
    });
}