1 use super::*;
2 use crate::mock::mock_stream::MockStream;
3 use crate::stream_info::RTPHeaderExtension;
4 use rtp::packet::Packet;
5 use tokio::sync::mpsc;
6 use tokio::time::Duration;
7 use util::Unmarshal;
8 use waitgroup::WaitGroup;
9
10 #[tokio::test]
test_twcc_sender_interceptor() -> Result<()>11 async fn test_twcc_sender_interceptor() -> Result<()> {
12 // "add transport wide cc to each packet"
13 let builder = Sender::builder().with_init_sequence_nr(0);
14 let icpr = builder.build("")?;
15
16 let (p_chan_tx, mut p_chan_rx) = mpsc::channel::<Packet>(10 * 5);
17 tokio::spawn(async move {
18 // start some parallel streams using the same interceptor to test for race conditions
19 let wg = WaitGroup::new();
20 for i in 0..10 {
21 let w = wg.worker();
22 let p_chan_tx2 = p_chan_tx.clone();
23 let icpr2 = Arc::clone(&icpr);
24 tokio::spawn(async move {
25 let _d = w;
26 let stream = MockStream::new(
27 &StreamInfo {
28 rtp_header_extensions: vec![RTPHeaderExtension {
29 uri: TRANSPORT_CC_URI.to_owned(),
30 id: 1,
31 }],
32 ..Default::default()
33 },
34 icpr2,
35 )
36 .await;
37
38 let id = i + 1;
39 #[allow(clippy::identity_op)]
40 for seq_num in [id * 1, id * 2, id * 3, id * 4, id * 5] {
41 stream
42 .write_rtp(&rtp::packet::Packet {
43 header: rtp::header::Header {
44 sequence_number: seq_num,
45 ..Default::default()
46 },
47 ..Default::default()
48 })
49 .await
50 .unwrap();
51
52 let timeout = tokio::time::sleep(Duration::from_millis(10));
53 tokio::pin!(timeout);
54
55 tokio::select! {
56 p = stream.written_rtp() =>{
57 if let Some(p) = p {
58 assert_eq!(p.header.sequence_number, seq_num);
59 let _ = p_chan_tx2.send(p).await;
60 }else{
61 panic!("stream.written_rtp none");
62 }
63 }
64 _ = timeout.as_mut()=>{
65 panic!("written rtp packet not found");
66 }
67 };
68 }
69
70 let _ = stream.close().await;
71 });
72 }
73 wg.wait().await;
74 });
75
76 while let Some(p) = p_chan_rx.recv().await {
77 // Can't check for increasing transport cc sequence number, since we can't ensure ordering between the streams
78 // on pChan is same as in the interceptor, but at least make sure each packet has a seq nr.
79 let mut extension_header = p.header.get_extension(1).unwrap();
80 let _twcc = TransportCcExtension::unmarshal(&mut extension_header)?;
81 }
82
83 Ok(())
84 }
85