1
use std::collections::VecDeque;
2
use std::time::Duration;
3
 
4
use str0m::format::Codec;
5
use str0m::media::MediaKind;
6
use str0m::rtp::{ExtensionValues, RtpWrite, Ssrc};
7
use str0m::{Event, RtcError};
8
 
9
mod common;
10
use common::{connect_l_r, init_crypto_default, init_log, progress};
11
 
12
#[test]
13
pub fn rtp_direct_reset_ssrc() -> Result<(), RtcError> {
14
    init_log();
15
    init_crypto_default();
16
 
17
    let (mut l, mut r) = connect_l_r();
18
 
19
    let mid = "aud".into();
20
    let rid = "hi".into();
21
 
22
    // Initial SSRC
23
    let ssrc_tx_initial: Ssrc = 42.into();
24
    // New SSRC to use after reset
25
    let ssrc_tx_new: Ssrc = 84.into();
26
 
27
    l.direct_api().declare_media(mid, MediaKind::Audio);
28
    l.direct_api()
29
        .declare_stream_tx(ssrc_tx_initial, None, mid, Some(rid));
30
 
31
    r.direct_api()
32
        .declare_media(mid, MediaKind::Audio)
33
        .expect_rid_rx(rid);
34
 
35
    // Set initial timing
36
    let max = l.last.max(r.last);
37
    l.last = max;
38
    r.last = max;
39
 
40
    let params = l.params_opus();
41
    let ssrc = l
42
        .direct_api()
43
        .stream_tx_by_mid(mid, Some(rid))
44
        .unwrap()
45
        .ssrc();
46
    assert_eq!(
47
        ssrc, ssrc_tx_initial,
48
        "Initial SSRC should match what we set"
49
    );
50
    assert_eq!(params.spec().codec, Codec::Opus);
51
    let pt = params.pt();
52
 
53
    // First batch of packets with initial SSRC
54
    let first_batch: Vec<&[u8]> = vec![&[0x1, 0x2, 0x3, 0x4], &[0x5, 0x6, 0x7, 0x8]];
55
 
56
    let mut first_batch: VecDeque<_> = first_batch.into();
57
    let mut first_counts: Vec<u64> = vec![0, 1];
58
    let mut write_at = l.last + Duration::from_millis(300);
59
 
60
    // Send first batch with initial SSRC
61
    loop {
62
        if l.start + l.duration() > write_at {
63
            write_at = l.last + Duration::from_millis(300);
64
            if let Some(packet) = first_batch.pop_front() {
65
                let wallclock = l.start + l.duration();
66
                let mut direct = l.direct_api();
67
                let stream = direct.stream_tx_by_mid(mid, Some(rid)).unwrap();
68
 
69
                let count = first_counts.remove(0);
70
                let time = (count * 1000 + 47_000_000) as u32;
71
                let seq_no = (47_000 + count).into();
72
 
73
                let exts = ExtensionValues {
74
                    audio_level: Some(-42 - count as i8),
75
                    voice_activity: Some(false),
76
                    ..Default::default()
77
                };
78
 
79
                stream.write_rtp(RtpWrite::new(pt, seq_no, time, wallclock, packet).ext_vals(exts));
80
            }
81
        }
82
 
83
        progress(&mut l, &mut r)?;
84
 
85
        if first_batch.is_empty() && first_counts.is_empty() {
86
            break;
87
        }
88
    }
89
 
90
    // Run a bit longer to ensure first batch of packets arrive
91
    for _ in 0..20 {
92
        progress(&mut l, &mut r)?;
93
    }
94
 
95
    // Try resetting to the same SSRC (should fail and return None)
96
    let mut api = l.direct_api();
97
    let result = api.reset_stream_tx(mid, Some(rid), ssrc_tx_initial, None);
98
    assert!(
99
        result.is_none(),
100
        "Resetting to the same SSRC should return None"
101
    );
102
 
103
    // Reset the SSRC with None for RTX since the stream doesn't use RTX
104
    let mut api = l.direct_api();
105
    let result = api.reset_stream_tx(mid, Some(rid), ssrc_tx_new, None);
106
    assert!(result.is_some(), "Reset should succeed with valid new SSRC");
107
 
108
    // Verify the SSRC was changed
109
    let updated_ssrc = l
110
        .direct_api()
111
        .stream_tx_by_mid(mid, Some(rid))
112
        .unwrap()
113
        .ssrc();
114
    assert_eq!(
115
        updated_ssrc, ssrc_tx_new,
116
        "SSRC should be updated to new value"
117
    );
118
 
119
    // Second batch of packets with new SSRC
120
    let second_batch: Vec<&[u8]> = vec![&[0x9, 0xa, 0xb, 0xc], &[0xd, 0xe, 0xf, 0x10]];
121
 
122
    let mut second_batch: VecDeque<_> = second_batch.into();
123
    let mut second_counts: Vec<u64> = vec![0, 1];
124
    write_at = l.last + Duration::from_millis(300);
125
 
126
    // Send second batch with new SSRC
127
    loop {
128
        if l.start + l.duration() > write_at {
129
            write_at = l.last + Duration::from_millis(300);
130
            if let Some(packet) = second_batch.pop_front() {
131
                let wallclock = l.start + l.duration();
132
                let mut direct = l.direct_api();
133
                let stream = direct.stream_tx_by_mid(mid, Some(rid)).unwrap();
134
 
135
                let count = second_counts.remove(0);
136
                let time = (count * 1000 + 48_000_000) as u32;
137
                let seq_no = (48_000 + count).into(); // Different seq range to verify reset
138
 
139
                let exts = ExtensionValues {
140
                    audio_level: Some(-52 - count as i8),
141
                    voice_activity: Some(false),
142
                    ..Default::default()
143
                };
144
 
145
                stream.write_rtp(RtpWrite::new(pt, seq_no, time, wallclock, packet).ext_vals(exts));
146
            }
147
        }
148
 
149
        progress(&mut l, &mut r)?;
150
 
151
        if second_batch.is_empty() && second_counts.is_empty() {
152
            // Run a bit longer to ensure packets arrive
153
            for _ in 0..20 {
154
                progress(&mut l, &mut r)?;
155
            }
156
            break;
157
        }
158
    }
159
 
160
    // Collect all received media packets
161
    let media: Vec<_> = r
162
        .events
163
        .iter()
164
        .filter_map(|(_, e)| {
165
            if let Event::RtpPacket(v) = e {
166
                Some(v)
167
            } else {
168
                None
169
            }
170
        })
171
        .collect();
172
 
173
    // Should have received all 4 packets
174
    assert_eq!(media.len(), 4, "Should have received all 4 packets");
175
 
176
    // Verify that we have packets with both SSRCs
177
    let first_ssrc_packets: Vec<_> = media
178
        .iter()
179
        .filter(|p| p.header.ssrc == ssrc_tx_initial)
180
        .collect();
181
    let second_ssrc_packets: Vec<_> = media
182
        .iter()
183
        .filter(|p| p.header.ssrc == ssrc_tx_new)
184
        .collect();
185
 
186
    assert_eq!(
187
        first_ssrc_packets.len(),
188
        2,
189
        "Should have 2 packets with initial SSRC"
190
    );
191
    assert_eq!(
192
        second_ssrc_packets.len(),
193
        2,
194
        "Should have 2 packets with new SSRC"
195
    );
196
 
197
    // Verify specific packet properties
198
    // First SSRC packets
199
    let h0 = &first_ssrc_packets[0].header;
200
    let h1 = &first_ssrc_packets[1].header;
201
 
202
    assert_eq!(h0.sequence_number, 47000);
203
    assert_eq!(h1.sequence_number, 47001);
204
 
205
    assert_eq!(h0.timestamp, 47_000_000);
206
    assert_eq!(h1.timestamp, 47_001_000);
207
 
208
    assert_eq!(h0.ext_vals.audio_level, Some(-42));
209
    assert_eq!(h1.ext_vals.audio_level, Some(-43));
210
 
211
    // Second SSRC packets - sequence numbers should be reset
212
    let h2 = &second_ssrc_packets[0].header;
213
    let h3 = &second_ssrc_packets[1].header;
214
 
215
    assert_eq!(h2.sequence_number, 48000);
216
    assert_eq!(h3.sequence_number, 48001);
217
 
218
    assert_eq!(h2.timestamp, 48_000_000);
219
    assert_eq!(h3.timestamp, 48_001_000);
220
 
221
    assert_eq!(h2.ext_vals.audio_level, Some(-52));
222
    assert_eq!(h3.ext_vals.audio_level, Some(-53));
223
 
224
    // Clean up
225
    l.direct_api().remove_media(mid);
226
    r.direct_api().remove_media(mid);
227
 
228
    Ok(())
229
}
230
 
231
#[test]
232
pub fn rtp_direct_reset_ssrc_with_rtx() -> Result<(), RtcError> {
233
    init_log();
234
    init_crypto_default();
235
 
236
    let (mut l, mut r) = connect_l_r();
237
 
238
    let mid = "vid".into();
239
    let rid = "hi".into();
240
 
241
    // Initial SSRCs
242
    let ssrc_tx_initial: Ssrc = 100.into();
243
    let ssrc_rtx_initial: Ssrc = 101.into();
244
 
245
    // New SSRCs to use after reset
246
    let ssrc_tx_new: Ssrc = 200.into();
247
    let ssrc_rtx_new: Ssrc = 201.into();
248
 
249
    // Create media and stream with both main and RTX SSRCs
250
    l.direct_api().declare_media(mid, MediaKind::Video);
251
    l.direct_api()
252
        .declare_stream_tx(ssrc_tx_initial, Some(ssrc_rtx_initial), mid, Some(rid));
253
 
254
    r.direct_api()
255
        .declare_media(mid, MediaKind::Video)
256
        .expect_rid_rx(rid);
257
 
258
    // Set initial timing
259
    let max = l.last.max(r.last);
260
    l.last = max;
261
    r.last = max;
262
 
263
    // Verify the initial SSRCs
264
    {
265
        let mut api = l.direct_api();
266
        let stream = api.stream_tx_by_mid(mid, Some(rid)).unwrap();
267
        assert_eq!(
268
            stream.ssrc(),
269
            ssrc_tx_initial,
270
            "Initial SSRC should match what we set"
271
        );
272
        assert_eq!(
273
            stream.rtx(),
274
            Some(ssrc_rtx_initial),
275
            "Initial RTX SSRC should match what we set"
276
        );
277
    }
278
 
279
    // Try resetting with the same RTX SSRC (should fail)
280
    let mut api = l.direct_api();
281
    let result = api.reset_stream_tx(mid, Some(rid), ssrc_tx_new, Some(ssrc_rtx_initial));
282
    assert!(result.is_none(), "Reset with same RTX SSRC should fail");
283
 
284
    // Reset with new main and RTX SSRCs (should succeed)
285
    let mut api = l.direct_api();
286
    let result = api.reset_stream_tx(mid, Some(rid), ssrc_tx_new, Some(ssrc_rtx_new));
287
    assert!(
288
        result.is_some(),
289
        "Reset with new main and RTX SSRCs should succeed"
290
    );
291
 
292
    // Verify the SSRCs were changed
293
    {
294
        let mut api = l.direct_api();
295
        let stream = api.stream_tx_by_mid(mid, Some(rid)).unwrap();
296
        assert_eq!(
297
            stream.ssrc(),
298
            ssrc_tx_new,
299
            "SSRC should be updated to new value"
300
        );
301
        assert_eq!(
302
            stream.rtx(),
303
            Some(ssrc_rtx_new),
304
            "RTX SSRC should be updated to new value"
305
        );
306
    }
307
 
308
    // Clean up
309
    l.direct_api().remove_media(mid);
310
    r.direct_api().remove_media(mid);
311
 
312
    Ok(())
313
}