use std::collections::HashMap;
use std::time::{Duration, Instant};
use crate::packet::rtcp::{
NtpTimestamp, RtcpReceiverReport, RtcpReportBlock, RtcpSdesChunk, RtcpSdesItem,
RtcpSenderReport, RtcpSourceDescription,
};
use crate::stats::loss::PacketLossTracker;
use crate::{RtpSsrc, RtpTimestamp};
pub const RTCP_MIN_INTERVAL: Duration = Duration::from_millis(1000); pub const RTCP_BANDWIDTH_FRACTION: f64 = 0.05; pub const RTCP_SENDER_BANDWIDTH_FRACTION: f64 = 0.25;
#[derive(Debug)]
pub struct RtcpReportGenerator {
local_ssrc: RtpSsrc,
cname: String,
loss_stats: HashMap<RtpSsrc, PacketLossTracker>,
last_sr_time: Option<Instant>,
last_sr_ntp: Option<NtpTimestamp>,
last_rtp_timestamp: Option<RtpTimestamp>,
packets_sent: u32,
octets_sent: u32,
last_interval: Duration,
session_bandwidth: u32,
senders: u32,
receivers: u32,
enabled: bool,
}
impl RtcpReportGenerator {
pub fn new(local_ssrc: RtpSsrc, cname: String) -> Self {
Self {
local_ssrc,
cname,
loss_stats: HashMap::new(),
last_sr_time: None,
last_sr_ntp: None,
last_rtp_timestamp: None,
packets_sent: 0,
octets_sent: 0,
last_interval: RTCP_MIN_INTERVAL,
session_bandwidth: 64000, senders: 1,
receivers: 0,
enabled: true,
}
}
pub fn set_bandwidth(&mut self, bandwidth_bps: u32) {
self.session_bandwidth = bandwidth_bps;
}
pub fn update_sent_stats(&mut self, packets: u32, octets: u32) {
self.packets_sent += packets;
self.octets_sent += octets;
}
pub fn process_received_packet(&mut self, ssrc: RtpSsrc, seq: u16) {
let tracker = self
.loss_stats
.entry(ssrc)
.or_insert_with(|| PacketLossTracker::new());
tracker.process(seq);
}
pub fn calculate_interval(&mut self) -> Duration {
let rtcp_bw = (self.session_bandwidth as f64 * RTCP_BANDWIDTH_FRACTION) as u32;
let members = self.senders + self.receivers;
if members == 0 {
return RTCP_MIN_INTERVAL;
}
let avg_rtcp_size = 100; let interval_seconds = (avg_rtcp_size * 8 * members) as f64 / rtcp_bw as f64;
let randomizer = rand::random::<f64>() + 0.5; let interval = Duration::from_secs_f64(interval_seconds * randomizer);
let interval = if interval < RTCP_MIN_INTERVAL {
RTCP_MIN_INTERVAL
} else {
interval
};
self.last_interval = interval;
interval
}
pub fn generate_sender_report(&mut self, rtp_timestamp: RtpTimestamp) -> RtcpSenderReport {
let ntp = NtpTimestamp::now();
self.last_sr_ntp = Some(ntp);
self.last_sr_time = Some(Instant::now());
self.last_rtp_timestamp = Some(rtp_timestamp);
let mut report_blocks = Vec::new();
for (ssrc, tracker) in &self.loss_stats {
let stats = tracker.get_stats();
let report_block = RtcpReportBlock {
ssrc: *ssrc,
fraction_lost: stats.fraction_lost,
cumulative_lost: tracker.get_cumulative_lost(),
highest_seq: stats.packets_expected as u32,
jitter: 0, last_sr: 0, delay_since_last_sr: 0, };
report_blocks.push(report_block);
if report_blocks.len() >= 31 {
break;
}
}
RtcpSenderReport {
ssrc: self.local_ssrc,
ntp_timestamp: ntp,
rtp_timestamp,
sender_packet_count: self.packets_sent,
sender_octet_count: self.octets_sent,
report_blocks,
}
}
pub fn generate_receiver_report(&self) -> RtcpReceiverReport {
let mut report_blocks = Vec::new();
for (ssrc, tracker) in &self.loss_stats {
let stats = tracker.get_stats();
let report_block = RtcpReportBlock {
ssrc: *ssrc,
fraction_lost: stats.fraction_lost,
cumulative_lost: tracker.get_cumulative_lost(),
highest_seq: stats.packets_expected as u32,
jitter: 0, last_sr: 0, delay_since_last_sr: 0, };
report_blocks.push(report_block);
if report_blocks.len() >= 31 {
break;
}
}
RtcpReceiverReport {
ssrc: self.local_ssrc,
report_blocks,
}
}
pub fn generate_sdes(&self) -> RtcpSourceDescription {
let mut sdes = RtcpSourceDescription::new();
let mut chunk = RtcpSdesChunk::new(self.local_ssrc);
chunk.add_item(RtcpSdesItem::cname(self.cname.clone()));
sdes.add_chunk(chunk);
sdes
}
pub fn should_send_report(&self) -> bool {
if !self.enabled {
return false;
}
if let Some(last_time) = self.last_sr_time {
Instant::now().duration_since(last_time) >= self.last_interval
} else {
true
}
}
pub fn set_enabled(&mut self, enabled: bool) {
self.enabled = enabled;
}
pub fn update_members(&mut self, senders: u32, receivers: u32) {
self.senders = senders;
self.receivers = receivers;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_rtcp_interval() {
let mut generator = RtcpReportGenerator::new(0x12345678, "user@example.com".to_string());
let interval = generator.calculate_interval();
assert!(interval >= RTCP_MIN_INTERVAL);
generator.set_bandwidth(1000000); let high_bw_interval = generator.calculate_interval();
generator.set_bandwidth(10000); let low_bw_interval = generator.calculate_interval();
assert!(high_bw_interval <= low_bw_interval);
}
#[test]
fn test_sender_report_generation() {
let mut generator = RtcpReportGenerator::new(0x12345678, "user@example.com".to_string());
generator.update_sent_stats(100, 10000);
let remote_ssrc = 0xabcdef01;
for seq in 1000..1010 {
generator.process_received_packet(remote_ssrc, seq);
}
let sr = generator.generate_sender_report(12345);
assert_eq!(sr.ssrc, 0x12345678);
assert_eq!(sr.rtp_timestamp, 12345);
assert_eq!(sr.sender_packet_count, 100);
assert_eq!(sr.sender_octet_count, 10000);
assert_eq!(sr.report_blocks.len(), 1);
assert_eq!(sr.report_blocks[0].ssrc, remote_ssrc);
assert_eq!(sr.report_blocks[0].fraction_lost, 0); }
#[test]
fn test_sdes_generation() {
let generator = RtcpReportGenerator::new(0x12345678, "user@example.com".to_string());
let sdes = generator.generate_sdes();
assert_eq!(sdes.chunks.len(), 1);
assert_eq!(sdes.chunks[0].ssrc, 0x12345678);
assert_eq!(sdes.chunks[0].items.len(), 1);
assert_eq!(sdes.chunks[0].items[0].value, "user@example.com");
}
}