1 use super::*;
2 use crate::mock::mock_stream::MockStream;
3 use crate::stream_info::RTPHeaderExtension;
4 use rtcp::transport_feedbacks::transport_layer_cc::{
5     PacketStatusChunk, RunLengthChunk, StatusChunkTypeTcc, StatusVectorChunk, SymbolSizeTypeTcc,
6     SymbolTypeTcc, TransportLayerCc,
7 };
8 use util::Marshal;
9 
10 #[tokio::test]
11 async fn test_twcc_receiver_interceptor_before_any_packets() -> Result<()> {
12     let builder = Receiver::builder();
13     let icpr = builder.build("")?;
14 
15     let stream = MockStream::new(
16         &StreamInfo {
17             ssrc: 1,
18             rtp_header_extensions: vec![RTPHeaderExtension {
19                 uri: TRANSPORT_CC_URI.to_owned(),
20                 id: 1,
21                 ..Default::default()
22             }],
23             ..Default::default()
24         },
25         icpr,
26     )
27     .await;
28 
29     tokio::select! {
30         pkts = stream.written_rtcp() => {
31             assert!(pkts.map(|p| p.is_empty()).unwrap_or(true), "Should not have sent an RTCP packet before receiving the first RTP packets")
32         }
33         _ = tokio::time::sleep(Duration::from_millis(300)) => {
34             // All good
35         }
36     }
37 
38     stream.close().await?;
39 
40     Ok(())
41 }
42 
43 #[tokio::test]
44 async fn test_twcc_receiver_interceptor_after_rtp_packets() -> Result<()> {
45     let builder = Receiver::builder();
46     let icpr = builder.build("")?;
47 
48     let stream = MockStream::new(
49         &StreamInfo {
50             ssrc: 1,
51             rtp_header_extensions: vec![RTPHeaderExtension {
52                 uri: TRANSPORT_CC_URI.to_owned(),
53                 id: 1,
54                 ..Default::default()
55             }],
56             ..Default::default()
57         },
58         icpr,
59     )
60     .await;
61 
62     for i in 0..10 {
63         let mut hdr = rtp::header::Header::default();
64         let tcc = TransportCcExtension {
65             transport_sequence: i,
66         }
67         .marshal()?;
68         hdr.set_extension(1, tcc)?;
69         stream
70             .receive_rtp(rtp::packet::Packet {
71                 header: hdr,
72                 ..Default::default()
73             })
74             .await;
75     }
76 
77     let pkts = stream.written_rtcp().await.unwrap();
78     assert_eq!(pkts.len(), 1);
79     if let Some(cc) = pkts[0].as_any().downcast_ref::<TransportLayerCc>() {
80         assert_eq!(1, cc.media_ssrc);
81         assert_eq!(0, cc.base_sequence_number);
82         assert_eq!(
83             vec![PacketStatusChunk::RunLengthChunk(RunLengthChunk {
84                 type_tcc: StatusChunkTypeTcc::RunLengthChunk,
85                 packet_status_symbol: SymbolTypeTcc::PacketReceivedSmallDelta,
86                 run_length: 10,
87             })],
88             cc.packet_chunks
89         );
90     } else {
91         assert!(false);
92     }
93 
94     stream.close().await?;
95 
96     Ok(())
97 }
98 
99 #[tokio::test(start_paused = true)]
100 async fn test_twcc_receiver_interceptor_different_delays_between_rtp_packets() -> Result<()> {
101     let builder = Receiver::builder().with_interval(Duration::from_millis(500));
102     let icpr = builder.build("")?;
103 
104     let stream = MockStream::new(
105         &StreamInfo {
106             ssrc: 1,
107             rtp_header_extensions: vec![RTPHeaderExtension {
108                 uri: TRANSPORT_CC_URI.to_owned(),
109                 id: 1,
110                 ..Default::default()
111             }],
112             ..Default::default()
113         },
114         icpr,
115     )
116     .await;
117 
118     let delays = vec![0, 10, 100, 200];
119     for (i, d) in delays.iter().enumerate() {
120         tokio::time::advance(Duration::from_millis(*d)).await;
121 
122         let mut hdr = rtp::header::Header::default();
123         let tcc = TransportCcExtension {
124             transport_sequence: i as u16,
125         }
126         .marshal()?;
127 
128         hdr.set_extension(1, tcc)?;
129         stream
130             .receive_rtp(rtp::packet::Packet {
131                 header: hdr,
132                 ..Default::default()
133             })
134             .await;
135 
136         // Yield so this packet can be processed
137         tokio::task::yield_now().await;
138     }
139 
140     // Force a packet to be generated
141     tokio::time::advance(Duration::from_millis(2001)).await;
142     tokio::task::yield_now().await;
143 
144     let pkts = stream.written_rtcp().await.unwrap();
145 
146     assert_eq!(pkts.len(), 1);
147     if let Some(cc) = pkts[0].as_any().downcast_ref::<TransportLayerCc>() {
148         assert_eq!(0, cc.base_sequence_number);
149         assert_eq!(
150             vec![PacketStatusChunk::StatusVectorChunk(StatusVectorChunk {
151                 type_tcc: StatusChunkTypeTcc::StatusVectorChunk,
152                 symbol_size: SymbolSizeTypeTcc::TwoBit,
153                 symbol_list: vec![
154                     SymbolTypeTcc::PacketReceivedSmallDelta,
155                     SymbolTypeTcc::PacketReceivedSmallDelta,
156                     SymbolTypeTcc::PacketReceivedLargeDelta,
157                     SymbolTypeTcc::PacketReceivedLargeDelta,
158                 ],
159             })],
160             cc.packet_chunks
161         );
162     } else {
163         assert!(false);
164     }
165 
166     stream.close().await?;
167 
168     Ok(())
169 }
170 
171 #[tokio::test(start_paused = true)]
172 async fn test_twcc_receiver_interceptor_packet_loss() -> Result<()> {
173     let builder = Receiver::builder().with_interval(Duration::from_secs(2));
174     let icpr = builder.build("")?;
175 
176     let stream = MockStream::new(
177         &StreamInfo {
178             ssrc: 1,
179             rtp_header_extensions: vec![RTPHeaderExtension {
180                 uri: TRANSPORT_CC_URI.to_owned(),
181                 id: 1,
182                 ..Default::default()
183             }],
184             ..Default::default()
185         },
186         icpr,
187     )
188     .await;
189 
190     let sequence_number_to_delay = &[
191         (0, 0),
192         (1, 10),
193         (4, 100),
194         (8, 200),
195         (9, 20),
196         (10, 20),
197         (30, 300),
198     ];
199 
200     for (i, d) in sequence_number_to_delay {
201         tokio::time::advance(Duration::from_millis(*d)).await;
202         let mut hdr = rtp::header::Header::default();
203         let tcc = TransportCcExtension {
204             transport_sequence: *i,
205         }
206         .marshal()?;
207         hdr.set_extension(1, tcc)?;
208         stream
209             .receive_rtp(rtp::packet::Packet {
210                 header: hdr,
211                 ..Default::default()
212             })
213             .await;
214 
215         // Yield so this packet can be processed
216         tokio::task::yield_now().await;
217     }
218 
219     // Force a packet to be generated
220     tokio::time::advance(Duration::from_millis(2001)).await;
221     tokio::task::yield_now().await;
222 
223     let pkts = stream.written_rtcp().await.unwrap();
224 
225     assert_eq!(pkts.len(), 1);
226     if let Some(cc) = pkts[0].as_any().downcast_ref::<TransportLayerCc>() {
227         assert_eq!(0, cc.base_sequence_number);
228         assert_eq!(
229             vec![
230                 PacketStatusChunk::StatusVectorChunk(StatusVectorChunk {
231                     type_tcc: StatusChunkTypeTcc::StatusVectorChunk,
232                     symbol_size: SymbolSizeTypeTcc::TwoBit,
233                     symbol_list: vec![
234                         SymbolTypeTcc::PacketReceivedSmallDelta,
235                         SymbolTypeTcc::PacketReceivedSmallDelta,
236                         SymbolTypeTcc::PacketNotReceived,
237                         SymbolTypeTcc::PacketNotReceived,
238                         SymbolTypeTcc::PacketReceivedLargeDelta,
239                         SymbolTypeTcc::PacketNotReceived,
240                         SymbolTypeTcc::PacketNotReceived,
241                     ],
242                 }),
243                 PacketStatusChunk::StatusVectorChunk(StatusVectorChunk {
244                     type_tcc: StatusChunkTypeTcc::StatusVectorChunk,
245                     symbol_size: SymbolSizeTypeTcc::TwoBit,
246                     symbol_list: vec![
247                         SymbolTypeTcc::PacketNotReceived,
248                         SymbolTypeTcc::PacketReceivedLargeDelta,
249                         SymbolTypeTcc::PacketReceivedSmallDelta,
250                         SymbolTypeTcc::PacketReceivedSmallDelta,
251                         SymbolTypeTcc::PacketNotReceived,
252                         SymbolTypeTcc::PacketNotReceived,
253                         SymbolTypeTcc::PacketNotReceived,
254                     ],
255                 }),
256                 PacketStatusChunk::RunLengthChunk(RunLengthChunk {
257                     type_tcc: StatusChunkTypeTcc::RunLengthChunk,
258                     packet_status_symbol: SymbolTypeTcc::PacketNotReceived,
259                     run_length: 16,
260                 }),
261                 PacketStatusChunk::RunLengthChunk(RunLengthChunk {
262                     type_tcc: StatusChunkTypeTcc::RunLengthChunk,
263                     packet_status_symbol: SymbolTypeTcc::PacketReceivedLargeDelta,
264                     run_length: 1,
265                 }),
266             ],
267             cc.packet_chunks
268         );
269     } else {
270         assert!(false);
271     }
272 
273     stream.close().await?;
274 
275     Ok(())
276 }
277 
278 #[tokio::test]
279 async fn test_twcc_receiver_interceptor_overflow() -> Result<()> {
280     let builder = Receiver::builder();
281     let icpr = builder.build("")?;
282 
283     let stream = MockStream::new(
284         &StreamInfo {
285             ssrc: 1,
286             rtp_header_extensions: vec![RTPHeaderExtension {
287                 uri: TRANSPORT_CC_URI.to_owned(),
288                 id: 1,
289                 ..Default::default()
290             }],
291             ..Default::default()
292         },
293         icpr,
294     )
295     .await;
296 
297     for i in [65530, 65534, 65535, 1, 2, 10] {
298         let mut hdr = rtp::header::Header::default();
299         let tcc = TransportCcExtension {
300             transport_sequence: i,
301         }
302         .marshal()?;
303         hdr.set_extension(1, tcc)?;
304         stream
305             .receive_rtp(rtp::packet::Packet {
306                 header: hdr,
307                 ..Default::default()
308             })
309             .await;
310     }
311 
312     let pkts = stream.written_rtcp().await.unwrap();
313     assert_eq!(pkts.len(), 1);
314     if let Some(cc) = pkts[0].as_any().downcast_ref::<TransportLayerCc>() {
315         assert_eq!(65530, cc.base_sequence_number);
316         assert_eq!(
317             vec![
318                 PacketStatusChunk::StatusVectorChunk(StatusVectorChunk {
319                     type_tcc: StatusChunkTypeTcc::StatusVectorChunk,
320                     symbol_size: SymbolSizeTypeTcc::OneBit,
321                     symbol_list: vec![
322                         SymbolTypeTcc::PacketReceivedSmallDelta,
323                         SymbolTypeTcc::PacketNotReceived,
324                         SymbolTypeTcc::PacketNotReceived,
325                         SymbolTypeTcc::PacketNotReceived,
326                         SymbolTypeTcc::PacketReceivedSmallDelta,
327                         SymbolTypeTcc::PacketReceivedSmallDelta,
328                         SymbolTypeTcc::PacketNotReceived,
329                         SymbolTypeTcc::PacketReceivedSmallDelta,
330                         SymbolTypeTcc::PacketReceivedSmallDelta,
331                         SymbolTypeTcc::PacketNotReceived,
332                         SymbolTypeTcc::PacketNotReceived,
333                         SymbolTypeTcc::PacketNotReceived,
334                         SymbolTypeTcc::PacketNotReceived,
335                         SymbolTypeTcc::PacketNotReceived,
336                     ],
337                 }),
338                 PacketStatusChunk::StatusVectorChunk(StatusVectorChunk {
339                     type_tcc: StatusChunkTypeTcc::StatusVectorChunk,
340                     symbol_size: SymbolSizeTypeTcc::TwoBit,
341                     symbol_list: vec![
342                         SymbolTypeTcc::PacketNotReceived,
343                         SymbolTypeTcc::PacketNotReceived,
344                         SymbolTypeTcc::PacketReceivedSmallDelta,
345                     ],
346                 }),
347             ],
348             cc.packet_chunks
349         );
350     } else {
351         assert!(false);
352     }
353 
354     stream.close().await?;
355 
356     Ok(())
357 }
358