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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

use rs2_stream::resource_manager::{ResourceConfig, ResourceManager};

#[tokio::test]
async fn test_resource_manager_creation() {
    let manager = ResourceManager::new();
    let metrics = manager.get_metrics().await;
    assert_eq!(metrics.current_memory_bytes, 0);
    assert_eq!(metrics.active_keys, 0);
}

#[tokio::test]
async fn test_memory_tracking() {
    let manager = ResourceManager::new();
    
    manager.track_memory_allocation(1000).await.unwrap();
    let metrics = manager.get_metrics().await;
    assert_eq!(metrics.current_memory_bytes, 1000);
    assert_eq!(metrics.peak_memory_bytes, 1000);
    
    manager.track_memory_deallocation(500).await;
    let metrics = manager.get_metrics().await;
    assert_eq!(metrics.current_memory_bytes, 500);
    assert_eq!(metrics.peak_memory_bytes, 1000);
}

#[tokio::test]
async fn test_key_tracking() {
    let manager = ResourceManager::new();
    
    manager.track_key_creation().await.unwrap();
    let metrics = manager.get_metrics().await;
    assert_eq!(metrics.active_keys, 1);
    
    manager.track_key_removal().await;
    let metrics = manager.get_metrics().await;
    assert_eq!(metrics.active_keys, 0);
}

#[tokio::test]
async fn test_circuit_breaker() {
    let manager = ResourceManager::new();
    
    assert!(!manager.is_circuit_open().await);
    
    // Note: trip_circuit_breaker is private, so we'll test the public interface
    // by checking the initial state and metrics
    let metrics = manager.get_metrics().await;
    assert_eq!(metrics.circuit_breaker_trips, 0);
    
    manager.reset_circuit_breaker().await;
    // We can't directly test the internal state, but we can test that the method doesn't panic
}

#[tokio::test]
async fn test_resource_config() {
    let config = ResourceConfig::default();
    assert_eq!(config.max_memory_bytes, 1024 * 1024 * 1024); // 1GB
    assert_eq!(config.max_keys, 100_000);
    assert_eq!(config.memory_threshold_percent, 80);
    assert_eq!(config.buffer_overflow_threshold, 10_000);
}

#[tokio::test]
async fn test_custom_resource_config() {
    let config = ResourceConfig {
        max_memory_bytes: 512 * 1024 * 1024, // 512MB
        max_keys: 50_000,
        memory_threshold_percent: 70,
        buffer_overflow_threshold: 5_000,
        cleanup_interval: std::time::Duration::from_secs(60),
        emergency_cleanup_threshold: 90,
    };
    
    let manager = ResourceManager::with_config(config);
    let metrics = manager.get_metrics().await;
    assert_eq!(metrics.current_memory_bytes, 0);
    assert_eq!(metrics.active_keys, 0);
}

#[tokio::test]
async fn test_buffer_overflow_tracking() {
    let manager = ResourceManager::new();
    
    // Track some buffer overflows
    manager.track_buffer_overflow().await.unwrap();
    manager.track_buffer_overflow().await.unwrap();
    
    let metrics = manager.get_metrics().await;
    assert_eq!(metrics.buffer_overflow_count, 2);
}

#[tokio::test]
async fn test_memory_pressure() {
    let manager = ResourceManager::new();
    
    // Initially no pressure
    assert!(!manager.is_under_memory_pressure().await);
    
    // Allocate memory to trigger pressure - use a large amount to ensure we exceed threshold
    let large_allocation = 1024 * 1024 * 1024; // 1GB
    manager.track_memory_allocation(large_allocation).await.unwrap();
    
    assert!(manager.is_under_memory_pressure().await);
    
    // Check memory pressure percentage
    let pressure = manager.get_memory_pressure().await;
    assert!(pressure > 0);
}

#[tokio::test]
async fn test_resource_tracking() {
    let manager = ResourceManager::new();
    
    // Track a resource
    manager.track_resource("test_resource".to_string(), 1024).await;
    
    // Untrack the resource
    let size = manager.untrack_resource("test_resource").await;
    assert_eq!(size, Some(1024));
    
    // Try to untrack non-existent resource
    let size = manager.untrack_resource("non_existent").await;
    assert_eq!(size, None);
}

#[tokio::test]
async fn test_global_resource_manager() {
    use rs2_stream::resource_manager::get_global_resource_manager;
    
    let manager = get_global_resource_manager();
    let metrics = manager.get_metrics().await;
    
    // Global manager should be initialized with default values
    assert_eq!(metrics.current_memory_bytes, 0);
    assert_eq!(metrics.active_keys, 0);
    assert_eq!(metrics.circuit_breaker_trips, 0);
    assert_eq!(metrics.emergency_cleanups, 0);
}