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
139

use futures_util::StreamExt;
use rs2_stream::media::streaming::{MediaStreamingService, StreamingServiceFactory};
use rs2_stream::media::types::{MediaStream, MediaType, QualityLevel};
use std::collections::HashMap;
use tokio::runtime::Runtime;

#[test]
fn test_streaming_service_creation() {
    // Test factory methods
    let live_service = StreamingServiceFactory::create_live_streaming_service();
    let file_service = StreamingServiceFactory::create_file_streaming_service();
    let low_latency_service = StreamingServiceFactory::create_low_latency_service();

    // Basic creation test - just ensure they don't panic
    assert!(true);
}

#[test]
fn test_streaming_service_metrics() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        // Create service
        let service = MediaStreamingService::new(512);

        // Get initial metrics
        let metrics = service.get_metrics().await;

        // Verify initial metrics
        assert_eq!(metrics.items_processed, 0);
        assert_eq!(metrics.bytes_processed, 0);
        assert_eq!(metrics.errors, 0);
        assert_eq!(metrics.average_item_size, 0.0);
    });
}

#[test]
fn test_streaming_service_live_stream() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        // Create service
        let service = MediaStreamingService::new(512);

        // Create stream config
        let mut metadata = HashMap::new();
        metadata.insert("max_chunks".to_string(), "10".to_string());

        let stream_config = MediaStream {
            id: "test_live_stream".to_string(),
            user_id: 0,
            content_type: MediaType::Video,
            quality: QualityLevel::Low,
            chunk_size: 1024,
            created_at: Default::default(),
            metadata,
        };

        // Start live stream
        let mut stream = service.start_live_stream(stream_config).await;

        // Collect chunks
        let mut chunks = Vec::new();
        for _ in 0..5 {
            if let Some(chunk) = stream.next().await {
                chunks.push(chunk);
            }
        }

        // Verify we got chunks
        assert!(!chunks.is_empty());
        assert!(chunks.len() <= 5); // May be less due to backpressure

        // Verify chunk properties
        for chunk in chunks {
            assert_eq!(chunk.stream_id, "test_live_stream");
            assert_eq!(chunk.data.len(), 1024);
            assert!(!chunk.is_final);
        }

        // Get metrics
        let metrics = service.get_metrics().await;

        // Verify metrics were updated
        assert!(metrics.items_processed > 0);
        assert!(metrics.bytes_processed > 0);
    });
}

#[test]
fn test_streaming_service_metrics_stream() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        // Create service
        let service = MediaStreamingService::new(512);

        // Get metrics stream
        let mut metrics_stream = service.get_metrics_stream();

        // Get first metrics update
        let metrics = metrics_stream.next().await.unwrap();

        // Basic verification
        assert_eq!(metrics.items_processed, 0);
        assert_eq!(metrics.bytes_processed, 0);
    });
}

#[test]
fn test_streaming_service_shutdown() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        // Create service
        let service = MediaStreamingService::new(512);

        // Shutdown service
        service.shutdown().await;

        // Try to get chunk stream after shutdown
        let chunk_stream = service.get_chunk_stream();

        // Verify we can still get a stream (but it should be empty/closed)
        assert!(true);
    });
}

// Helper function to create a test stream config
fn create_test_stream_config(id: &str, chunk_size: usize) -> MediaStream {
    let mut metadata = HashMap::new();
    metadata.insert("test_key".to_string(), "test_value".to_string());

    MediaStream {
        id: id.to_string(),
        user_id: 0,
        content_type: MediaType::Video,
        quality: QualityLevel::Low,
        chunk_size,
        created_at: Default::default(),
        metadata,
    }
}