1 use std::collections::HashMap; 2 use std::fmt; 3 use std::sync::Arc; 4 use std::time::SystemTime; 5 6 use super::{inbound, outbound, StatsContainer}; 7 use async_trait::async_trait; 8 use rtcp::extended_report::{DLRRReportBlock, ExtendedReport}; 9 use rtcp::payload_feedbacks::full_intra_request::FullIntraRequest; 10 use rtcp::payload_feedbacks::picture_loss_indication::PictureLossIndication; 11 use rtcp::receiver_report::ReceiverReport; 12 use rtcp::sender_report::SenderReport; 13 use rtcp::transport_feedbacks::transport_layer_nack::TransportLayerNack; 14 use rtp::extension::abs_send_time_extension::unix2ntp; 15 use tokio::sync::{mpsc, oneshot}; 16 use tokio::time::Duration; 17 18 use util::sync::Mutex; 19 use util::{MarshalSize, Unmarshal}; 20 21 use crate::error::Result; 22 use crate::stream_info::StreamInfo; 23 use crate::{Attributes, Interceptor, RTCPReader, RTCPWriter, RTPReader, RTPWriter}; 24 25 #[derive(Debug)] 26 enum Message { 27 StatUpdate { 28 ssrc: u32, 29 update: StatsUpdate, 30 }, 31 RequestInboundSnapshot { 32 ssrcs: Vec<u32>, 33 chan: oneshot::Sender<Vec<Option<inbound::StatsSnapshot>>>, 34 }, 35 RequestOutboundSnapshot { 36 ssrcs: Vec<u32>, 37 chan: oneshot::Sender<Vec<Option<outbound::StatsSnapshot>>>, 38 }, 39 } 40 41 #[derive(Debug)] 42 enum StatsUpdate { 43 /// Stats collected on the receiving end(inbound) of an RTP stream. 44 InboundRTP { 45 packets: u64, 46 header_bytes: u64, 47 payload_bytes: u64, 48 last_packet_timestamp: SystemTime, 49 }, 50 /// Stats collected on the sending end(outbound) of an RTP stream. 51 OutboundRTP { 52 packets: u64, 53 header_bytes: u64, 54 payload_bytes: u64, 55 last_packet_timestamp: SystemTime, 56 }, 57 /// Stats collected from received RTCP packets. 58 InboundRTCP { 59 fir_count: Option<u64>, 60 pli_count: Option<u64>, 61 nack_count: Option<u64>, 62 }, 63 /// Stats collected from sent RTCP packets. 64 OutboundRTCP { 65 fir_count: Option<u64>, 66 pli_count: Option<u64>, 67 nack_count: Option<u64>, 68 }, 69 /// An extended sequence number sent in an SR. 70 OutboundSRExtSeqNum { seq_num: u32 }, 71 /// Stats collected from received Receiver Reports i.e. where we have an outbound RTP stream. 72 InboundRecieverReport { 73 ext_seq_num: u32, 74 total_lost: u32, 75 jitter: u32, 76 rtt_ms: Option<f64>, 77 fraction_lost: u8, 78 }, 79 /// Stats collected from recieved Sender Reports i.e. where we have an inbound RTP stream. 80 InboundSenderRerport { 81 packets_and_bytes_sent: Option<(u32, u32)>, 82 rtt_ms: Option<f64>, 83 }, 84 } 85 86 pub struct StatsInterceptor { 87 // Wrapped RTP streams 88 recv_streams: Mutex<HashMap<u32, Arc<RTPReadRecorder>>>, 89 send_streams: Mutex<HashMap<u32, Arc<RTPWriteRecorder>>>, 90 91 tx: mpsc::Sender<Message>, 92 93 id: String, 94 now_gen: Arc<dyn Fn() -> SystemTime + Send + Sync>, 95 } 96 97 impl StatsInterceptor { 98 pub fn new(id: String) -> Self { 99 let (tx, rx) = mpsc::channel(100); 100 101 tokio::spawn(run_stats_reducer(rx)); 102 103 Self { 104 id, 105 recv_streams: Default::default(), 106 send_streams: Default::default(), 107 tx, 108 now_gen: Arc::new(SystemTime::now), 109 } 110 } 111 112 fn with_time_gen<F>(id: String, now_gen: F) -> Self 113 where 114 F: Fn() -> SystemTime + Send + Sync + 'static, 115 { 116 let (tx, rx) = mpsc::channel(100); 117 tokio::spawn(run_stats_reducer(rx)); 118 119 Self { 120 id, 121 recv_streams: Default::default(), 122 send_streams: Default::default(), 123 tx, 124 now_gen: Arc::new(now_gen), 125 } 126 } 127 128 pub async fn fetch_inbound_stats( 129 &self, 130 ssrcs: Vec<u32>, 131 ) -> Vec<Option<inbound::StatsSnapshot>> { 132 let (tx, rx) = oneshot::channel(); 133 134 if let Err(e) = self 135 .tx 136 .send(Message::RequestInboundSnapshot { ssrcs, chan: tx }) 137 .await 138 { 139 log::debug!( 140 "Failed to fetch inbound RTP stream stats from stats task with error: {}", 141 e 142 ); 143 144 return vec![]; 145 } 146 147 rx.await.unwrap_or_default() 148 } 149 150 pub async fn fetch_outbound_stats( 151 &self, 152 ssrcs: Vec<u32>, 153 ) -> Vec<Option<outbound::StatsSnapshot>> { 154 let (tx, rx) = oneshot::channel(); 155 156 if let Err(e) = self 157 .tx 158 .send(Message::RequestOutboundSnapshot { ssrcs, chan: tx }) 159 .await 160 { 161 log::debug!( 162 "Failed to fetch outbound RTP stream stats from stats task with error: {}", 163 e 164 ); 165 166 return vec![]; 167 } 168 169 rx.await.unwrap_or_default() 170 } 171 } 172 173 async fn run_stats_reducer(mut rx: mpsc::Receiver<Message>) { 174 let mut ssrc_stats: StatsContainer = Default::default(); 175 let mut cleanup_ticker = tokio::time::interval(Duration::from_secs(10)); 176 177 loop { 178 tokio::select! { 179 maybe_msg = rx.recv() => { 180 let msg = match maybe_msg { 181 Some(m) => m, 182 None => break, 183 }; 184 185 match msg { 186 Message::StatUpdate { ssrc, update } => { 187 handle_stats_update(&mut ssrc_stats, ssrc, update); 188 } 189 Message::RequestInboundSnapshot { ssrcs, chan} => { 190 let result = ssrcs 191 .into_iter() 192 .map(|ssrc| ssrc_stats.get_inbound_stats(ssrc).map(inbound::StreamStats::snapshot)) 193 .collect(); 194 195 let _ = chan.send(result); 196 } 197 Message::RequestOutboundSnapshot { ssrcs, chan} => { 198 let result = ssrcs 199 .into_iter() 200 .map(|ssrc| ssrc_stats.get_outbound_stats(ssrc).map(outbound::StreamStats::snapshot)) 201 .collect(); 202 203 let _ = chan.send(result); 204 205 } 206 } 207 208 } 209 _ = cleanup_ticker.tick() => { 210 ssrc_stats.remove_stale_entries(); 211 } 212 } 213 } 214 } 215 216 fn handle_stats_update(ssrc_stats: &mut StatsContainer, ssrc: u32, update: StatsUpdate) { 217 match update { 218 StatsUpdate::InboundRTP { 219 packets, 220 header_bytes, 221 payload_bytes, 222 last_packet_timestamp, 223 } => { 224 let stats = ssrc_stats.get_or_create_inbound_stream_stats(ssrc); 225 226 stats 227 .rtp_stats 228 .update(header_bytes, payload_bytes, packets, last_packet_timestamp); 229 stats.mark_updated(); 230 } 231 StatsUpdate::OutboundRTP { 232 packets, 233 header_bytes, 234 payload_bytes, 235 last_packet_timestamp, 236 } => { 237 let stats = ssrc_stats.get_or_create_outbound_stream_stats(ssrc); 238 stats 239 .rtp_stats 240 .update(header_bytes, payload_bytes, packets, last_packet_timestamp); 241 stats.mark_updated(); 242 } 243 StatsUpdate::InboundRTCP { 244 fir_count, 245 pli_count, 246 nack_count, 247 } => { 248 let stats = ssrc_stats.get_or_create_outbound_stream_stats(ssrc); 249 stats.rtcp_stats.update(fir_count, pli_count, nack_count); 250 stats.mark_updated(); 251 } 252 StatsUpdate::OutboundRTCP { 253 fir_count, 254 pli_count, 255 nack_count, 256 } => { 257 let stats = ssrc_stats.get_or_create_inbound_stream_stats(ssrc); 258 stats.rtcp_stats.update(fir_count, pli_count, nack_count); 259 stats.mark_updated(); 260 } 261 StatsUpdate::OutboundSRExtSeqNum { seq_num } => { 262 let stats = ssrc_stats.get_or_create_outbound_stream_stats(ssrc); 263 stats.record_sr_ext_seq_num(seq_num); 264 stats.mark_updated(); 265 } 266 StatsUpdate::InboundRecieverReport { 267 ext_seq_num, 268 total_lost, 269 jitter, 270 rtt_ms, 271 fraction_lost, 272 } => { 273 let stats = ssrc_stats.get_or_create_outbound_stream_stats(ssrc); 274 stats.record_remote_round_trip_time(rtt_ms); 275 stats.update_remote_fraction_lost(fraction_lost); 276 stats.update_remote_total_lost(total_lost); 277 stats.update_remote_inbound_packets_received(ext_seq_num, total_lost); 278 stats.update_remote_jitter(jitter); 279 280 stats.mark_updated(); 281 } 282 StatsUpdate::InboundSenderRerport { 283 rtt_ms, 284 packets_and_bytes_sent, 285 } => { 286 // This is a sender report we received, as such it concerns an RTP stream that's 287 // outbound at the remote. 288 let stats = ssrc_stats.get_or_create_inbound_stream_stats(ssrc); 289 290 if let Some((packets_sent, bytes_sent)) = packets_and_bytes_sent { 291 stats.record_sender_report(packets_sent, bytes_sent); 292 } 293 stats.record_remote_round_trip_time(rtt_ms); 294 295 stats.mark_updated(); 296 } 297 } 298 } 299 300 #[async_trait] 301 impl Interceptor for StatsInterceptor { 302 /// bind_remote_stream lets you modify any incoming RTP packets. It is called once for per RemoteStream. The returned method 303 /// will be called once per rtp packet. 304 async fn bind_remote_stream( 305 &self, 306 info: &StreamInfo, 307 reader: Arc<dyn RTPReader + Send + Sync>, 308 ) -> Arc<dyn RTPReader + Send + Sync> { 309 let mut lock = self.recv_streams.lock(); 310 311 let e = lock 312 .entry(info.ssrc) 313 .or_insert_with(|| Arc::new(RTPReadRecorder::new(reader, self.tx.clone()))); 314 315 e.clone() 316 } 317 318 /// unbind_remote_stream is called when the Stream is removed. It can be used to clean up any data related to that track. 319 async fn unbind_remote_stream(&self, info: &StreamInfo) { 320 let mut lock = self.recv_streams.lock(); 321 322 lock.remove(&info.ssrc); 323 } 324 325 /// bind_local_stream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method 326 /// will be called once per rtp packet. 327 async fn bind_local_stream( 328 &self, 329 info: &StreamInfo, 330 writer: Arc<dyn RTPWriter + Send + Sync>, 331 ) -> Arc<dyn RTPWriter + Send + Sync> { 332 let mut lock = self.send_streams.lock(); 333 334 let e = lock 335 .entry(info.ssrc) 336 .or_insert_with(|| Arc::new(RTPWriteRecorder::new(writer, self.tx.clone()))); 337 338 e.clone() 339 } 340 341 /// unbind_local_stream is called when the Stream is removed. It can be used to clean up any data related to that track. 342 async fn unbind_local_stream(&self, info: &StreamInfo) { 343 let mut lock = self.send_streams.lock(); 344 345 lock.remove(&info.ssrc); 346 } 347 348 async fn close(&self) -> Result<()> { 349 Ok(()) 350 } 351 352 /// bind_rtcp_writer lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method 353 /// will be called once per packet batch. 354 async fn bind_rtcp_writer( 355 &self, 356 writer: Arc<dyn RTCPWriter + Send + Sync>, 357 ) -> Arc<dyn RTCPWriter + Send + Sync> { 358 let now = self.now_gen.clone(); 359 360 Arc::new(RTCPWriteInterceptor { 361 rtcp_writer: writer, 362 tx: self.tx.clone(), 363 now_gen: move || now(), 364 }) 365 } 366 367 /// bind_rtcp_reader lets you modify any incoming RTCP packets. It is called once per sender/receiver, however this might 368 /// change in the future. The returned method will be called once per packet batch. 369 async fn bind_rtcp_reader( 370 &self, 371 reader: Arc<dyn RTCPReader + Send + Sync>, 372 ) -> Arc<dyn RTCPReader + Send + Sync> { 373 let now = self.now_gen.clone(); 374 375 Arc::new(RTCPReadInterceptor { 376 rtcp_reader: reader, 377 tx: self.tx.clone(), 378 now_gen: move || now(), 379 }) 380 } 381 } 382 383 pub struct RTCPReadInterceptor<F> { 384 rtcp_reader: Arc<dyn RTCPReader + Send + Sync>, 385 tx: mpsc::Sender<Message>, 386 now_gen: F, 387 } 388 389 #[async_trait] 390 impl<F> RTCPReader for RTCPReadInterceptor<F> 391 where 392 F: Fn() -> SystemTime + Send + Sync, 393 { 394 /// read a batch of rtcp packets 395 async fn read(&self, buf: &mut [u8], attributes: &Attributes) -> Result<(usize, Attributes)> { 396 let (n, attributes) = self.rtcp_reader.read(buf, attributes).await?; 397 398 let mut b = &buf[..n]; 399 let pkts = rtcp::packet::unmarshal(&mut b)?; 400 // Middle 32 bits 401 let now = (unix2ntp((self.now_gen)()) >> 16) as u32; 402 403 #[derive(Default, Debug)] 404 struct GenericRTCP { 405 fir_count: Option<u64>, 406 pli_count: Option<u64>, 407 nack_count: Option<u64>, 408 } 409 410 #[derive(Default, Debug)] 411 struct ReceiverReportEntry { 412 /// Extended sequence number value from Receiver Report, used to calculate remote 413 /// stats. 414 ext_seq_num: u32, 415 /// Total loss value from Receiver Report, used to calculate remote 416 /// stats. 417 total_lost: u32, 418 /// Jitter from Receiver Report. 419 jitter: u32, 420 /// Round Trip Time calculated from Receiver Report. 421 rtt_ms: Option<f64>, 422 /// Fraction of packets lost. 423 fraction_lost: u8, 424 } 425 426 #[derive(Default, Debug)] 427 struct SenderReportEntry { 428 /// NTP timestamp(from Sender Report). 429 sr_ntp_time: Option<u64>, 430 /// Packets Sent(from Sender Report). 431 sr_packets_sent: Option<u32>, 432 /// Bytes Sent(from Sender Report). 433 sr_bytes_sent: Option<u32>, 434 /// Last RR timestamp(middle bits) from DLRR extended report block. 435 dlrr_last_rr: Option<u32>, 436 /// Delay since last RR from DLRR extended report block. 437 dlrr_delay_rr: Option<u32>, 438 } 439 440 #[derive(Default, Debug)] 441 struct Entry { 442 generic_rtcp: GenericRTCP, 443 receiver_reports: Vec<ReceiverReportEntry>, 444 sender_reports: Vec<SenderReportEntry>, 445 } 446 let updates = pkts 447 .iter() 448 .fold(HashMap::<u32, Entry>::new(), |mut acc, p| { 449 if let Some(rr) = p.as_any().downcast_ref::<ReceiverReport>() { 450 for recp in &rr.reports { 451 let e = acc.entry(recp.ssrc).or_default(); 452 453 let rtt_ms = (recp.delay != 0) 454 .then(|| calculate_rtt_ms(now, recp.delay, recp.last_sender_report)); 455 456 e.receiver_reports.push(ReceiverReportEntry { 457 ext_seq_num: recp.last_sequence_number, 458 total_lost: recp.total_lost, 459 jitter: recp.jitter, 460 rtt_ms, 461 fraction_lost: recp.fraction_lost, 462 }); 463 } 464 } else if let Some(fir) = p.as_any().downcast_ref::<FullIntraRequest>() { 465 for fir_entry in &fir.fir { 466 let e = acc.entry(fir_entry.ssrc).or_default(); 467 e.generic_rtcp.fir_count = 468 e.generic_rtcp.fir_count.map(|v| v + 1).or(Some(1)); 469 } 470 } else if let Some(pli) = p.as_any().downcast_ref::<PictureLossIndication>() { 471 let e = acc.entry(pli.media_ssrc).or_default(); 472 e.generic_rtcp.pli_count = e.generic_rtcp.pli_count.map(|v| v + 1).or(Some(1)); 473 } else if let Some(nack) = p.as_any().downcast_ref::<TransportLayerNack>() { 474 let count = nack.nacks.iter().flat_map(|p| p.into_iter()).count() as u64; 475 476 let e = acc.entry(nack.media_ssrc).or_default(); 477 e.generic_rtcp.nack_count = 478 e.generic_rtcp.nack_count.map(|v| v + count).or(Some(count)); 479 } else if let Some(sr) = p.as_any().downcast_ref::<SenderReport>() { 480 let e = acc.entry(sr.ssrc).or_default(); 481 let sr_e = { 482 let need_new_entry = e 483 .sender_reports 484 .last() 485 .map(|e| e.sr_packets_sent.is_some()) 486 .unwrap_or(true); 487 488 if need_new_entry { 489 e.sender_reports.push(Default::default()); 490 } 491 492 // SAFETY: Unrwap ok because we just added an entry above 493 e.sender_reports.last_mut().unwrap() 494 }; 495 496 sr_e.sr_ntp_time = Some(sr.ntp_time); 497 sr_e.sr_packets_sent = Some(sr.packet_count); 498 sr_e.sr_bytes_sent = Some(sr.octet_count); 499 } else if let Some(xr) = p.as_any().downcast_ref::<ExtendedReport>() { 500 // Extended Report(XR) 501 502 // We only care about DLRR reports 503 let dlrrs = xr.reports.iter().flat_map(|report| { 504 let dlrr = report.as_any().downcast_ref::<DLRRReportBlock>(); 505 506 dlrr.map(|b| b.reports.iter()).into_iter().flatten() 507 }); 508 509 for dlrr in dlrrs { 510 let e = acc.entry(dlrr.ssrc).or_default(); 511 let sr_e = { 512 let need_new_entry = e 513 .sender_reports 514 .last() 515 .map(|e| e.dlrr_last_rr.is_some()) 516 .unwrap_or(true); 517 518 if need_new_entry { 519 e.sender_reports.push(Default::default()); 520 } 521 522 // SAFETY: Unrwap ok because we just added an entry above 523 e.sender_reports.last_mut().unwrap() 524 }; 525 526 sr_e.dlrr_last_rr = Some(dlrr.last_rr); 527 sr_e.dlrr_delay_rr = Some(dlrr.dlrr); 528 } 529 } 530 531 acc 532 }); 533 534 for ( 535 ssrc, 536 Entry { 537 generic_rtcp, 538 mut receiver_reports, 539 mut sender_reports, 540 }, 541 ) in updates.into_iter() 542 { 543 // Sort RR by seq number low to high 544 receiver_reports.sort_by(|a, b| a.ext_seq_num.cmp(&b.ext_seq_num)); 545 // Sort SR by ntp time, low to high 546 sender_reports 547 .sort_by(|a, b| a.sr_ntp_time.unwrap_or(0).cmp(&b.sr_ntp_time.unwrap_or(0))); 548 549 let _ = self 550 .tx 551 .send(Message::StatUpdate { 552 ssrc, 553 update: StatsUpdate::InboundRTCP { 554 fir_count: generic_rtcp.fir_count, 555 pli_count: generic_rtcp.pli_count, 556 nack_count: generic_rtcp.nack_count, 557 }, 558 }) 559 .await; 560 561 let futures = receiver_reports.into_iter().map(|rr| { 562 self.tx.send(Message::StatUpdate { 563 ssrc, 564 update: StatsUpdate::InboundRecieverReport { 565 ext_seq_num: rr.ext_seq_num, 566 total_lost: rr.total_lost, 567 jitter: rr.jitter, 568 rtt_ms: rr.rtt_ms, 569 fraction_lost: rr.fraction_lost, 570 }, 571 }) 572 }); 573 for fut in futures { 574 // TODO: Use futures::join_all 575 let _ = fut.await; 576 } 577 578 let futures = sender_reports.into_iter().map(|sr| { 579 let rtt_ms = match (sr.dlrr_last_rr, sr.dlrr_delay_rr, sr.sr_packets_sent) { 580 (Some(last_rr), Some(delay_rr), Some(_)) if last_rr != 0 && delay_rr != 0 => { 581 Some(calculate_rtt_ms(now, delay_rr, last_rr)) 582 } 583 _ => None, 584 }; 585 586 self.tx.send(Message::StatUpdate { 587 ssrc, 588 update: StatsUpdate::InboundSenderRerport { 589 packets_and_bytes_sent: sr 590 .sr_packets_sent 591 .and_then(|ps| sr.sr_bytes_sent.map(|bs| (ps, bs))), 592 rtt_ms, 593 }, 594 }) 595 }); 596 for fut in futures { 597 // TODO: Use futures::join_all 598 let _ = fut.await; 599 } 600 } 601 602 Ok((n, attributes)) 603 } 604 } 605 606 pub struct RTCPWriteInterceptor<F> { 607 rtcp_writer: Arc<dyn RTCPWriter + Send + Sync>, 608 tx: mpsc::Sender<Message>, 609 now_gen: F, 610 } 611 612 #[async_trait] 613 impl<F> RTCPWriter for RTCPWriteInterceptor<F> 614 where 615 F: Fn() -> SystemTime + Send + Sync, 616 { 617 async fn write( 618 &self, 619 pkts: &[Box<dyn rtcp::packet::Packet + Send + Sync>], 620 attributes: &Attributes, 621 ) -> Result<usize> { 622 #[derive(Default, Debug)] 623 struct Entry { 624 fir_count: Option<u64>, 625 pli_count: Option<u64>, 626 nack_count: Option<u64>, 627 sr_ext_seq_num: Option<u32>, 628 } 629 let updates = pkts 630 .iter() 631 .fold(HashMap::<u32, Entry>::new(), |mut acc, p| { 632 if let Some(fir) = p.as_any().downcast_ref::<FullIntraRequest>() { 633 for fir_entry in &fir.fir { 634 let e = acc.entry(fir_entry.ssrc).or_default(); 635 e.fir_count = e.fir_count.map(|v| v + 1).or(Some(1)); 636 } 637 } else if let Some(pli) = p.as_any().downcast_ref::<PictureLossIndication>() { 638 let e = acc.entry(pli.media_ssrc).or_default(); 639 e.pli_count = e.pli_count.map(|v| v + 1).or(Some(1)); 640 } else if let Some(nack) = p.as_any().downcast_ref::<TransportLayerNack>() { 641 let count = nack.nacks.iter().flat_map(|p| p.into_iter()).count() as u64; 642 643 let e = acc.entry(nack.media_ssrc).or_default(); 644 e.nack_count = e.nack_count.map(|v| v + count).or(Some(count)); 645 } else if let Some(sr) = p.as_any().downcast_ref::<SenderReport>() { 646 for rep in &sr.reports { 647 let e = acc.entry(rep.ssrc).or_default(); 648 649 match e.sr_ext_seq_num { 650 // We want the initial value for `last_sequence_number` from the first 651 // SR. It's possible that an RTCP batch contains more than one SR, in 652 // which case we should use the lowest value. 653 Some(seq_num) if seq_num > rep.last_sequence_number => { 654 e.sr_ext_seq_num = Some(rep.last_sequence_number) 655 } 656 None => e.sr_ext_seq_num = Some(rep.last_sequence_number), 657 _ => {} 658 } 659 } 660 } 661 662 acc 663 }); 664 665 for ( 666 ssrc, 667 Entry { 668 fir_count, 669 pli_count, 670 nack_count, 671 sr_ext_seq_num, 672 }, 673 ) in updates.into_iter() 674 { 675 let _ = self 676 .tx 677 .send(Message::StatUpdate { 678 ssrc, 679 update: StatsUpdate::OutboundRTCP { 680 fir_count, 681 pli_count, 682 nack_count, 683 }, 684 }) 685 .await; 686 687 if let Some(seq_num) = sr_ext_seq_num { 688 let _ = self 689 .tx 690 .send(Message::StatUpdate { 691 ssrc, 692 update: StatsUpdate::OutboundSRExtSeqNum { seq_num }, 693 }) 694 .await; 695 } 696 } 697 698 self.rtcp_writer.write(pkts, attributes).await 699 } 700 } 701 702 pub struct RTPReadRecorder { 703 rtp_reader: Arc<dyn RTPReader + Send + Sync>, 704 tx: mpsc::Sender<Message>, 705 } 706 707 impl RTPReadRecorder { 708 fn new(rtp_reader: Arc<dyn RTPReader + Send + Sync>, tx: mpsc::Sender<Message>) -> Self { 709 Self { rtp_reader, tx } 710 } 711 } 712 713 impl fmt::Debug for RTPReadRecorder { 714 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 715 f.debug_struct("RTPReadRecorder").finish() 716 } 717 } 718 719 #[async_trait] 720 impl RTPReader for RTPReadRecorder { 721 async fn read(&self, buf: &mut [u8], attributes: &Attributes) -> Result<(usize, Attributes)> { 722 let (bytes_read, attributes) = self.rtp_reader.read(buf, attributes).await?; 723 // TODO: This parsing happens redundantly in several interceptors, would be good if we 724 // could not do this. 725 let mut b = &buf[..bytes_read]; 726 let packet = rtp::packet::Packet::unmarshal(&mut b)?; 727 728 let _ = self 729 .tx 730 .send(Message::StatUpdate { 731 ssrc: packet.header.ssrc, 732 update: StatsUpdate::InboundRTP { 733 packets: 1, 734 header_bytes: (bytes_read - packet.payload.len()) as u64, 735 payload_bytes: packet.payload.len() as u64, 736 last_packet_timestamp: SystemTime::now(), 737 }, 738 }) 739 .await; 740 741 Ok((bytes_read, attributes)) 742 } 743 } 744 745 pub struct RTPWriteRecorder { 746 rtp_writer: Arc<dyn RTPWriter + Send + Sync>, 747 tx: mpsc::Sender<Message>, 748 } 749 750 impl RTPWriteRecorder { 751 fn new(rtp_writer: Arc<dyn RTPWriter + Send + Sync>, tx: mpsc::Sender<Message>) -> Self { 752 Self { rtp_writer, tx } 753 } 754 } 755 756 impl fmt::Debug for RTPWriteRecorder { 757 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 758 f.debug_struct("RTPWriteRecorder").finish() 759 } 760 } 761 762 #[async_trait] 763 impl RTPWriter for RTPWriteRecorder { 764 /// write a rtp packet 765 async fn write(&self, pkt: &rtp::packet::Packet, attributes: &Attributes) -> Result<usize> { 766 let n = self.rtp_writer.write(pkt, attributes).await?; 767 768 let _ = self 769 .tx 770 .send(Message::StatUpdate { 771 ssrc: pkt.header.ssrc, 772 update: StatsUpdate::OutboundRTP { 773 packets: 1, 774 header_bytes: pkt.header.marshal_size() as u64, 775 payload_bytes: pkt.payload.len() as u64, 776 last_packet_timestamp: SystemTime::now(), 777 }, 778 }) 779 .await; 780 781 Ok(n) 782 } 783 } 784 785 /// Calculate the round trip time for a given peer as described in 786 /// [RFC3550 6.4.1](https://datatracker.ietf.org/doc/html/rfc3550#section-6.4.1). 787 /// 788 /// ## Params 789 /// 790 /// - `now` the current middle 32 bits of an NTP timestamp for the current time. 791 /// - `delay` the delay(`DLSR`) since last sender report expressed as fractions of a second in 32 bits. 792 /// - `last_report` the middle 32 bits of an NTP timestamp for the most recent sender report(LSR) or Receiver Report(LRR). 793 fn calculate_rtt_ms(now: u32, delay: u32, last_report: u32) -> f64 { 794 // [10 Nov 1995 11:33:25.125 UTC] [10 Nov 1995 11:33:36.5 UTC] 795 // n SR(n) A=b710:8000 (46864.500 s) 796 // ----------------------------------------------------------------> 797 // v ^ 798 // ntp_sec =0xb44db705 v ^ dlsr=0x0005:4000 ( 5.250s) 799 // ntp_frac=0x20000000 v ^ lsr =0xb705:2000 (46853.125s) 800 // (3024992005.125 s) v ^ 801 // r v ^ RR(n) 802 // ----------------------------------------------------------------> 803 // |<-DLSR->| 804 // (5.250 s) 805 // 806 // A 0xb710:8000 (46864.500 s) 807 // DLSR -0x0005:4000 ( 5.250 s) 808 // LSR -0xb705:2000 (46853.125 s) 809 // ------------------------------- 810 // delay 0x0006:2000 ( 6.125 s) 811 812 let rtt = now - delay - last_report; 813 let rtt_seconds = rtt >> 16; 814 let rtt_fraction = (rtt & (u16::MAX as u32)) as f64 / (u16::MAX as u32) as f64; 815 816 rtt_seconds as f64 * 1000.0 + (rtt_fraction as f64) * 1000.0 817 } 818 819 #[cfg(test)] 820 mod test { 821 macro_rules! assert_feq { 822 ($left: expr, $right: expr) => { 823 assert_feq!($left, $right, 0.01); 824 }; 825 ($left: expr, $right: expr, $eps: expr) => { 826 if ($left - $right).abs() >= $eps { 827 assert!( 828 false, 829 "{:?} was not within {:?} of {:?}", 830 $left, $eps, $right 831 ); 832 } 833 }; 834 } 835 836 use bytes::Bytes; 837 use rtcp::extended_report::{DLRRReport, DLRRReportBlock, ExtendedReport}; 838 use rtcp::payload_feedbacks::full_intra_request::{FirEntry, FullIntraRequest}; 839 use rtcp::payload_feedbacks::picture_loss_indication::PictureLossIndication; 840 use rtcp::receiver_report::ReceiverReport; 841 use rtcp::reception_report::ReceptionReport; 842 use rtcp::sender_report::SenderReport; 843 use rtcp::transport_feedbacks::transport_layer_nack::{NackPair, TransportLayerNack}; 844 845 use std::sync::Arc; 846 use std::time::{Duration, SystemTime}; 847 848 use crate::error::Result; 849 use crate::mock::mock_stream::MockStream; 850 use crate::stream_info::StreamInfo; 851 852 use super::StatsInterceptor; 853 854 #[tokio::test] 855 async fn test_stats_interceptor_rtp() -> Result<()> { 856 let icpr: Arc<_> = Arc::new(StatsInterceptor::new("Hello".to_owned())); 857 858 let recv_stream = MockStream::new( 859 &StreamInfo { 860 ssrc: 123456, 861 ..Default::default() 862 }, 863 icpr.clone(), 864 ) 865 .await; 866 867 let send_stream = MockStream::new( 868 &StreamInfo { 869 ssrc: 234567, 870 ..Default::default() 871 }, 872 icpr.clone(), 873 ) 874 .await; 875 876 let _ = recv_stream 877 .receive_rtp(rtp::packet::Packet { 878 header: rtp::header::Header { 879 ssrc: 123456, 880 ..Default::default() 881 }, 882 payload: Bytes::from_static(b"\xde\xad\xbe\xef"), 883 }) 884 .await; 885 886 let _ = recv_stream 887 .read_rtp() 888 .await 889 .expect("After calling receive_rtp read_rtp should return Some")?; 890 891 let _ = send_stream 892 .write_rtp(&rtp::packet::Packet { 893 header: rtp::header::Header { 894 ssrc: 234567, 895 ..Default::default() 896 }, 897 payload: Bytes::from_static(b"\xde\xad\xbe\xef\xde\xad\xbe\xef"), 898 }) 899 .await; 900 901 let _ = send_stream 902 .write_rtp(&rtp::packet::Packet { 903 header: rtp::header::Header { 904 ssrc: 234567, 905 ..Default::default() 906 }, 907 payload: Bytes::from_static(&[0x13, 0x37]), 908 }) 909 .await; 910 911 let snapshots = icpr.fetch_inbound_stats(vec![123456]).await; 912 let recv_snapshot = snapshots[0] 913 .as_ref() 914 .expect("Stats should exist for ssrc: 123456"); 915 assert_eq!(recv_snapshot.packets_received(), 1); 916 assert_eq!(recv_snapshot.header_bytes_received(), 12); 917 assert_eq!(recv_snapshot.payload_bytes_received(), 4); 918 919 let snapshots = icpr.fetch_outbound_stats(vec![234567]).await; 920 let send_snapshot = snapshots[0] 921 .as_ref() 922 .expect("Stats should exist for ssrc: 234567"); 923 assert_eq!(send_snapshot.packets_sent(), 2); 924 assert_eq!(send_snapshot.header_bytes_sent(), 24); 925 assert_eq!(send_snapshot.payload_bytes_sent(), 10); 926 927 Ok(()) 928 } 929 930 #[tokio::test] 931 async fn test_stats_interceptor_rtcp() -> Result<()> { 932 let icpr: Arc<_> = Arc::new(StatsInterceptor::with_time_gen("Hello".to_owned(), || { 933 // 10 Nov 1995 11:33:36.5 UTC 934 SystemTime::UNIX_EPOCH + Duration::from_secs_f64(816003216.5) 935 })); 936 937 let recv_stream = MockStream::new( 938 &StreamInfo { 939 ssrc: 123456, 940 ..Default::default() 941 }, 942 icpr.clone(), 943 ) 944 .await; 945 946 let send_stream = MockStream::new( 947 &StreamInfo { 948 ssrc: 234567, 949 ..Default::default() 950 }, 951 icpr.clone(), 952 ) 953 .await; 954 955 send_stream 956 .write_rtcp(&[Box::new(SenderReport { 957 ssrc: 234567, 958 reports: vec![ 959 ReceptionReport { 960 ssrc: 234567, 961 last_sequence_number: (5 << 16) | 10, 962 ..Default::default() 963 }, 964 ReceptionReport { 965 ssrc: 234567, 966 last_sequence_number: (5 << 16) | 85, 967 ..Default::default() 968 }, 969 ], 970 ..Default::default() 971 })]) 972 .await 973 .expect("Failed to write RTCP packets"); 974 975 send_stream 976 .receive_rtcp(vec![ 977 Box::new(ReceiverReport { 978 reports: vec![ 979 ReceptionReport { 980 ssrc: 234567, 981 last_sequence_number: (5 << 16) | 64, 982 total_lost: 5, 983 ..Default::default() 984 }, 985 ReceptionReport { 986 ssrc: 234567, 987 last_sender_report: 0xb705_2000, 988 delay: 0x0005_4000, 989 last_sequence_number: (5 << 16) | 70, 990 total_lost: 8, 991 fraction_lost: 32, 992 jitter: 2250, 993 ..Default::default() 994 }, 995 ], 996 ..Default::default() 997 }), 998 Box::new(TransportLayerNack { 999 sender_ssrc: 0, 1000 media_ssrc: 234567, 1001 nacks: vec![NackPair { 1002 packet_id: 5, 1003 lost_packets: 0b0011_0110, 1004 }], 1005 }), 1006 Box::new(TransportLayerNack { 1007 sender_ssrc: 0, 1008 // NB: Different SSRC 1009 media_ssrc: 999999, 1010 nacks: vec![NackPair { 1011 packet_id: 5, 1012 lost_packets: 0b0011_0110, 1013 }], 1014 }), 1015 Box::new(PictureLossIndication { 1016 sender_ssrc: 0, 1017 media_ssrc: 234567, 1018 }), 1019 Box::new(PictureLossIndication { 1020 sender_ssrc: 0, 1021 media_ssrc: 234567, 1022 }), 1023 Box::new(FullIntraRequest { 1024 sender_ssrc: 0, 1025 media_ssrc: 234567, 1026 fir: vec![ 1027 FirEntry { 1028 ssrc: 234567, 1029 sequence_number: 132, 1030 }, 1031 FirEntry { 1032 ssrc: 234567, 1033 sequence_number: 135, 1034 }, 1035 ], 1036 }), 1037 ]) 1038 .await; 1039 let snapshots = icpr.fetch_outbound_stats(vec![234567]).await; 1040 let send_snapshot = snapshots[0] 1041 .as_ref() 1042 .expect("Outbound Stats should exist for ssrc: 234567"); 1043 1044 assert!( 1045 send_snapshot.remote_round_trip_time().is_none() 1046 && send_snapshot.remote_round_trip_time_measurements() == 0, 1047 "Before receiving the first RR we should not have a remote round trip time" 1048 ); 1049 let _ = send_stream 1050 .read_rtcp() 1051 .await 1052 .expect("After calling `receive_rtcp`, `read_rtcp` should return some packets"); 1053 1054 recv_stream 1055 .write_rtcp(&[ 1056 Box::new(TransportLayerNack { 1057 sender_ssrc: 0, 1058 media_ssrc: 123456, 1059 nacks: vec![NackPair { 1060 packet_id: 5, 1061 lost_packets: 0b0011_0111, 1062 }], 1063 }), 1064 Box::new(TransportLayerNack { 1065 sender_ssrc: 0, 1066 // NB: Different SSRC 1067 media_ssrc: 999999, 1068 nacks: vec![NackPair { 1069 packet_id: 5, 1070 lost_packets: 0b1111_0110, 1071 }], 1072 }), 1073 Box::new(PictureLossIndication { 1074 sender_ssrc: 0, 1075 media_ssrc: 123456, 1076 }), 1077 Box::new(PictureLossIndication { 1078 sender_ssrc: 0, 1079 media_ssrc: 123456, 1080 }), 1081 Box::new(PictureLossIndication { 1082 sender_ssrc: 0, 1083 media_ssrc: 123456, 1084 }), 1085 Box::new(FullIntraRequest { 1086 sender_ssrc: 0, 1087 media_ssrc: 123456, 1088 fir: vec![FirEntry { 1089 ssrc: 123456, 1090 sequence_number: 132, 1091 }], 1092 }), 1093 ]) 1094 .await 1095 .expect("Failed to write RTCP packets for recv_stream"); 1096 1097 recv_stream 1098 .receive_rtcp(vec![ 1099 Box::new(SenderReport { 1100 ssrc: 123456, 1101 ntp_time: 12345, // Used for ordering 1102 packet_count: 52, 1103 octet_count: 8172, 1104 reports: vec![], 1105 ..Default::default() 1106 }), 1107 Box::new(SenderReport { 1108 ssrc: 123456, 1109 ntp_time: 23456, // Used for ordering 1110 packet_count: 82, 1111 octet_count: 10351, 1112 reports: vec![], 1113 ..Default::default() 1114 }), 1115 Box::new(ExtendedReport { 1116 sender_ssrc: 928191, 1117 reports: vec![Box::new(DLRRReportBlock { 1118 reports: vec![DLRRReport { 1119 ssrc: 123456, 1120 last_rr: 0xb705_2000, 1121 dlrr: 0x0005_4000, 1122 }], 1123 })], 1124 }), 1125 Box::new(SenderReport { 1126 /// NB: Different SSRC 1127 ssrc: 9999999, 1128 ntp_time: 99999, // Used for ordering 1129 packet_count: 1231, 1130 octet_count: 193812, 1131 reports: vec![], 1132 ..Default::default() 1133 }), 1134 ]) 1135 .await; 1136 1137 let snapshots = icpr.fetch_inbound_stats(vec![123456]).await; 1138 let recv_snapshot = snapshots[0] 1139 .as_ref() 1140 .expect("Stats should exist for ssrc: 123456"); 1141 assert!( 1142 recv_snapshot.remote_round_trip_time().is_none() 1143 && recv_snapshot.remote_round_trip_time_measurements() == 0, 1144 "Before receiving the first SR/DLRR we should not have a remote round trip time" 1145 ); 1146 1147 let _ = recv_stream.read_rtcp().await.expect("read_rtcp failed"); 1148 1149 let snapshots = icpr.fetch_outbound_stats(vec![234567]).await; 1150 let send_snapshot = snapshots[0] 1151 .as_ref() 1152 .expect("Outbound Stats should exist for ssrc: 234567"); 1153 let rtt_ms = send_snapshot.remote_round_trip_time().expect( 1154 "After receiving an RR with a DSLR block we should have a remote round trip time", 1155 ); 1156 assert_feq!(rtt_ms, 6125.0); 1157 1158 assert_eq!(send_snapshot.nacks_received(), 5); 1159 assert_eq!(send_snapshot.plis_received(), 2); 1160 assert_eq!(send_snapshot.firs_received(), 2); 1161 // Last Seq Num(RR) - total lost(RR) - Initial Seq Num(SR) + 1 1162 // 70 - 8 - 10 + 1 = 53 1163 assert_eq!(send_snapshot.remote_packets_received(), 53); 1164 assert_feq!( 1165 send_snapshot 1166 .remote_fraction_lost() 1167 .expect("Should have a fraction lost values after receiving RR"), 1168 32.0 / 256.0 1169 ); 1170 assert_eq!(send_snapshot.remote_total_lost(), 8); 1171 assert_eq!(send_snapshot.remote_jitter(), 2250); 1172 1173 let snapshots = icpr.fetch_inbound_stats(vec![123456]).await; 1174 let recv_snapshot = snapshots[0] 1175 .as_ref() 1176 .expect("Stats should exist for ssrc: 123456"); 1177 assert_eq!(recv_snapshot.nacks_sent(), 6); 1178 assert_eq!(recv_snapshot.plis_sent(), 3); 1179 assert_eq!(recv_snapshot.firs_sent(), 1); 1180 assert_eq!(recv_snapshot.remote_packets_sent(), 82); 1181 assert_eq!(recv_snapshot.remote_bytes_sent(), 10351); 1182 let rtt_ms = recv_snapshot 1183 .remote_round_trip_time() 1184 .expect("After reciving SR and DLRR we should have a round trip time "); 1185 assert_feq!(rtt_ms, 6125.0); 1186 assert_eq!(recv_snapshot.remote_reports_sent(), 2); 1187 assert_eq!(recv_snapshot.remote_round_trip_time_measurements(), 1); 1188 assert_feq!(recv_snapshot.remote_total_round_trip_time(), 6125.0); 1189 1190 Ok(()) 1191 } 1192 } 1193