1
use std::collections::VecDeque;
2
use std::time::{Duration, Instant};
3
 
4
use netem::{NetemConfig, Probability, RandomLoss};
5
use str0m::format::Codec;
6
use str0m::media::MediaKind;
7
use str0m::media::Pt;
8
use str0m::rtp::{ExtensionValues, RawPacket, RtpWrite, Ssrc, Vp8Descriptor, Vp8Patch};
9
use str0m::{Event, RtcError};
10
 
11
mod common;
12
use common::{connect_l_r, init_crypto_default, init_log, progress};
13
 
14
const VP8_PAYLOAD: [u8; 6] = [0x90, 0xf0, 0x01, 0x02, 0xa3, 0x00];
15
const VP8_REWRITTEN_PAYLOAD: [u8; 6] = [0x90, 0xf0, 0x7e, 0x44, 0xbf, 0x00];
16
 
17
#[test]
18
pub fn rtp_direct_ssrc() -> Result<(), RtcError> {
19
    init_log();
20
    init_crypto_default();
21
 
22
    let (mut l, mut r) = connect_l_r();
23
 
24
    let mid = "aud".into();
25
 
26
    // In this example we are not using RID to identify the stream, we are simply
27
    // using SSRC 1 as knowledge shared between sending and receiving side.
28
    let ssrc: Ssrc = 1.into();
29
 
30
    l.direct_api().declare_media(mid, MediaKind::Audio);
31
 
32
    l.direct_api().declare_stream_tx(ssrc, None, mid, None);
33
 
34
    r.direct_api().declare_media(mid, MediaKind::Audio);
35
 
36
    r.direct_api().expect_stream_rx(ssrc, None, mid, None);
37
 
38
    let max = l.last.max(r.last);
39
    l.last = max;
40
    r.last = max;
41
 
42
    let params = l.params_opus();
43
    let ssrc = l.direct_api().stream_tx_by_mid(mid, None).unwrap().ssrc();
44
    assert_eq!(params.spec().codec, Codec::Opus);
45
    let pt = params.pt();
46
 
47
    let to_write: Vec<&[u8]> = vec![
48
        // 1
49
        &[0x1, 0x2, 0x3, 0x4],
50
        // 3
51
        &[0x9, 0xa, 0xb, 0xc],
52
        // 2
53
        &[0x5, 0x6, 0x7, 0x8],
54
    ];
55
 
56
    let mut to_write: VecDeque<_> = to_write.into();
57
 
58
    let mut write_at = l.last + Duration::from_millis(300);
59
 
60
    let mut counts: Vec<u64> = vec![0, 3, 1];
61
 
62
    loop {
63
        if l.start + l.duration() > write_at {
64
            write_at = l.last + Duration::from_millis(300);
65
            if let Some(packet) = to_write.pop_front() {
66
                let wallclock = l.start + l.duration();
67
 
68
                let mut direct = l.direct_api();
69
                let stream = direct.stream_tx(&ssrc).unwrap();
70
 
71
                let count = counts.remove(0);
72
                let time = (count * 1000 + 47_000_000) as u32;
73
                let seq_no = (47_000 + count).into();
74
 
75
                let exts = ExtensionValues {
76
                    audio_level: Some(-42 - count as i8),
77
                    voice_activity: Some(false),
78
                    ..Default::default()
79
                };
80
 
81
                stream.write_rtp(RtpWrite::new(pt, seq_no, time, wallclock, packet).ext_vals(exts));
82
            }
83
        }
84
 
85
        progress(&mut l, &mut r)?;
86
 
87
        if l.duration() > Duration::from_secs(10) {
88
            break;
89
        }
90
    }
91
 
92
    let media: Vec<_> = r
93
        .events
94
        .iter()
95
        .filter_map(|(_, e)| {
96
            if let Event::RtpPacket(v) = e {
97
                Some(v)
98
            } else {
99
                None
100
            }
101
        })
102
        .collect();
103
 
104
    assert_eq!(media.len(), 3);
105
 
106
    let h0 = media[0].header.clone();
107
    let h1 = media[1].header.clone();
108
    let h2 = media[2].header.clone();
109
 
110
    assert_eq!(h0.sequence_number, 47000);
111
    assert_eq!(h1.sequence_number, 47003);
112
    assert_eq!(h2.sequence_number, 47001);
113
 
114
    assert_eq!(h0.timestamp, 47_000_000);
115
    assert_eq!(h1.timestamp, 47_003_000);
116
    assert_eq!(h2.timestamp, 47_001_000);
117
 
118
    assert_eq!(h0.ext_vals.audio_level, Some(-42));
119
    assert_eq!(h1.ext_vals.audio_level, Some(-45));
120
    assert_eq!(h2.ext_vals.audio_level, Some(-43));
121
 
122
    assert!(!h0.marker);
123
    assert!(!h1.marker);
124
    assert!(!h2.marker);
125
 
126
    assert!(l.media(mid).is_some());
127
    assert!(l.direct_api().stream_tx_by_mid(mid, None).is_some());
128
    l.direct_api().remove_media(mid);
129
    assert!(l.media(mid).is_none());
130
    assert!(l.direct_api().stream_tx_by_mid(mid, None).is_none());
131
 
132
    assert!(r.media(mid).is_some());
133
    assert!(r.direct_api().stream_rx_by_mid(mid, None).is_some());
134
    r.direct_api().remove_media(mid);
135
    assert!(r.media(mid).is_none());
136
    assert!(r.direct_api().stream_rx_by_mid(mid, None).is_none());
137
 
138
    Ok(())
139
}
140
 
141
#[test]
142
pub fn rtp_direct_vp8_patch() -> Result<(), RtcError> {
143
    init_log();
144
    init_crypto_default();
145
 
146
    let (mut l, mut r) = connect_l_r();
147
 
148
    let mid = "vid".into();
149
    let ssrc: Ssrc = 1.into();
150
 
151
    l.direct_api().declare_media(mid, MediaKind::Video);
152
    l.direct_api().declare_stream_tx(ssrc, None, mid, None);
153
 
154
    r.direct_api().declare_media(mid, MediaKind::Video);
155
    r.direct_api().expect_stream_rx(ssrc, None, mid, None);
156
 
157
    let max = l.last.max(r.last);
158
    l.last = max;
159
    r.last = max;
160
 
161
    let params = l.params_vp8();
162
    assert_eq!(params.spec().codec, Codec::Vp8);
163
    let pt = params.pt();
164
    let wallclock = l.start + l.duration();
165
 
166
    {
167
        let mut direct = l.direct_api();
168
        let stream = direct.stream_tx(&ssrc).unwrap();
169
 
170
        stream.write_rtp(
171
            RtpWrite::new(pt, 47_000.into(), 47_000_000, wallclock, VP8_PAYLOAD)
172
                .vp8_patch(vp8_patch()),
173
        );
174
 
175
        stream.write_rtp(RtpWrite::new(
176
            pt,
177
            47_001.into(),
178
            47_001_000,
179
            wallclock,
180
            [20, 21, 22],
181
        ));
182
    }
183
 
184
    loop {
185
        progress(&mut l, &mut r)?;
186
 
187
        let has_two_media_packets = r
188
            .events
189
            .iter()
190
            .filter(|(_, e)| matches!(e, Event::RtpPacket(_)))
191
            .take(2)
192
            .count()
193
            == 2;
194
 
195
        if has_two_media_packets || l.duration() > Duration::from_secs(10) {
196
            break;
197
        }
198
    }
199
 
200
    let mut media: Vec<_> = r
201
        .events
202
        .iter()
203
        .filter_map(|(_, e)| {
204
            if let Event::RtpPacket(v) = e {
205
                Some(v)
206
            } else {
207
                None
208
            }
209
        })
210
        .collect();
211
 
212
    media.sort_by_key(|packet| packet.header.sequence_number);
213
 
214
    assert_eq!(media.len(), 2);
215
    assert_eq!(media[0].header.sequence_number, 47_000);
216
    assert_eq!(media[0].payload.as_ref(), VP8_REWRITTEN_PAYLOAD.as_slice());
217
    assert_eq!(media[1].header.sequence_number, 47_001);
218
    assert_eq!(media[1].payload.as_ref(), &[20, 21, 22]);
219
 
220
    Ok(())
221
}
222
 
223
#[test]
224
pub fn rtp_direct_vp8_patch_survives_rtx() -> Result<(), RtcError> {
225
    init_log();
226
    init_crypto_default();
227
 
228
    let (mut l, mut r) = connect_l_r();
229
 
230
    let mid = "vid".into();
231
    let ssrc: Ssrc = 42.into();
232
    let ssrc_rtx: Ssrc = 44.into();
233
 
234
    l.direct_api().declare_media(mid, MediaKind::Video);
235
    l.direct_api()
236
        .declare_stream_tx(ssrc, Some(ssrc_rtx), mid, None)
237
        .set_rtx_cache(32, Duration::from_secs(3), None);
238
 
239
    r.direct_api().declare_media(mid, MediaKind::Video);
240
    r.direct_api()
241
        .expect_stream_rx(ssrc, Some(ssrc_rtx), mid, None);
242
 
243
    let max = l.last.max(r.last);
244
    l.last = max;
245
    r.last = max;
246
 
247
    let params = l.params_vp8();
248
    assert_eq!(params.spec().codec, Codec::Vp8);
249
    let pt = params.pt();
250
    let rtx_pt = params.resend().unwrap();
251
    let rewritten_seq = 47_002_u64;
252
    let original_seq_bytes = (rewritten_seq as u16).to_be_bytes();
253
 
254
    for index in 0_u64..=1 {
255
        let wallclock = l.start + l.duration();
256
        l.direct_api().stream_tx(&ssrc).unwrap().write_rtp(
257
            RtpWrite::new(
258
                pt,
259
                (47_000 + index).into(),
260
                47_000_000 + index as u32 * 1_000,
261
                wallclock,
262
                [index as u8],
263
            )
264
            .nackable(true),
265
        );
266
 
267
        progress(&mut l, &mut r)?;
268
    }
269
 
270
    r.set_netem(NetemConfig::new().loss(RandomLoss::new(Probability::ONE)));
271
 
272
    let wallclock = l.start + l.duration();
273
    l.direct_api().stream_tx(&ssrc).unwrap().write_rtp(
274
        RtpWrite::new(pt, rewritten_seq.into(), 47_002_000, wallclock, VP8_PAYLOAD)
275
            .nackable(true)
276
            .vp8_patch(vp8_patch()),
277
    );
278
 
279
    progress(&mut l, &mut r)?;
280
    r.set_netem(NetemConfig::new());
281
 
282
    for index in 3_u64..=7 {
283
        let wallclock = l.start + l.duration();
284
        l.direct_api().stream_tx(&ssrc).unwrap().write_rtp(
285
            RtpWrite::new(
286
                pt,
287
                (47_000 + index).into(),
288
                47_000_000 + index as u32 * 1_000,
289
                wallclock,
290
                [index as u8],
291
            )
292
            .nackable(true),
293
        );
294
 
295
        progress(&mut l, &mut r)?;
296
    }
297
 
298
    let rtx_payload = loop {
299
        progress(&mut l, &mut r)?;
300
 
301
        if let Some(payload) = rtx_payload_for_seq(&r.events, rtx_pt, &original_seq_bytes) {
302
            break payload;
303
        }
304
 
305
        if l.duration() > Duration::from_secs(10) {
306
            panic!("rewritten packet should be retransmitted over RTX");
307
        }
308
    };
309
 
310
    assert_eq!(rtx_payload.get(2..), Some(VP8_REWRITTEN_PAYLOAD.as_slice()));
311
 
312
    let recovered_packet = r.events.iter().find_map(|(_, event)| match event {
313
        Event::RtpPacket(packet) if packet.header.sequence_number == rewritten_seq as u16 => {
314
            Some(packet)
315
        }
316
        _ => None,
317
    });
318
 
319
    assert_eq!(
320
        recovered_packet.map(|packet| packet.payload.as_ref()),
321
        Some(VP8_REWRITTEN_PAYLOAD.as_slice())
322
    );
323
 
324
    Ok(())
325
}
326
 
327
fn vp8_patch() -> Vp8Patch {
328
    Vp8Descriptor::parse(&VP8_PAYLOAD)
329
        .expect("valid VP8 descriptor")
330
        .patch()
331
        .picture_id(0x7e)
332
        .tl0_pic_idx(0x44)
333
        .key_idx(0x1f)
334
        .build()
335
        .expect("valid VP8 patch")
336
}
337
 
338
fn rtx_payload_for_seq<'a>(
339
    events: &'a [(Instant, Event)],
340
    rtx_pt: Pt,
341
    original_seq_bytes: &[u8; 2],
342
) -> Option<&'a [u8]> {
343
    events
344
        .iter()
345
        .find_map(|(_, event)| match event.as_raw_packet() {
346
            Some(RawPacket::RtpRx(header, payload))
347
                if header.payload_type == rtx_pt && payload.starts_with(original_seq_bytes) =>
348
            {
349
                Some(payload.as_slice())
350
            }
351
            _ => None,
352
        })
353
}