use crate::rtp::utils; use crate::rtp::RtpPacket; use bytes::BytesMut; use super::{ rtcp_app::RtcpApp, rtcp_bye::RtcpBye, rtcp_header::RtcpHeader, rtcp_rr::{ReportBlock, RtcpReceiverReport}, rtcp_sr::RtcpSenderReport, RTCP_RR, }; //For example: sequence numbers inserted are 65533, 65534, the new coming one is 2, //the new is 2 and old is 65534, the distance between 2 and 65534 is 4 which is //65535 - 65534 + 2 + 1.(65533,65534,65535,0,1,2) pub fn distance(new: u16, old: u16) -> u16 { if new < old { 65535 - old + new + 1 } else { new - old } } const MIN_SEQUENTIAL: u32 = 2; const RTP_SEQ_MOD: u32 = 1 << 16; const MAX_DROPOUT: u32 = 3000; const MAX_MISORDER: u32 = 100; /* * Per-source state information */ #[derive(Debug, Clone, Default)] struct RtcpSource { max_seq: u16, /* highest seq. number seen */ cycles: u32, /* shifted count of seq. number cycles */ base_seq: u32, /* base seq number */ bad_seq: u32, /* last 'bad' seq number + 1 */ probation: u32, /* sequ. packets till source is valid */ received: u32, /* packets received */ expected_prior: u32, /* packet expected at last interval */ received_prior: u32, /* packet received at last interval */ jitter: f64, /* estimated jitter */ } impl RtcpSource { //static int rtp_seq_update(struct rtp_member *sender, uint16_t seq) fn update_sequence(&mut self, seq: u16) -> usize { let delta = distance(seq, self.max_seq); if self.probation > 0 { /* packet is in sequence */ if seq == self.max_seq + 1 { self.probation -= 1; self.max_seq = seq; if self.probation == 0 { self.init_seq(seq); self.received += 1; return 1; } } else { self.probation = MIN_SEQUENTIAL - 1; self.max_seq = seq; } return 0; } else if delta < MAX_DROPOUT as u16 { /* in order, with permissible gap */ if seq < self.max_seq { /* * Sequence number wrapped - count another 64K cycle. */ self.cycles += RTP_SEQ_MOD; } self.max_seq = seq; } else if delta as u32 <= RTP_SEQ_MOD - MAX_MISORDER { if seq == self.bad_seq as u16 { /* * Two sequential packets -- assume that the other side * restarted without telling us so just re-sync * (i.e., pretend this was the first packet). */ self.init_seq(seq); } else { self.bad_seq = (seq as u32 + 1) & (RTP_SEQ_MOD - 1); return 0; } } else { /* duplicate or reordered packet */ } self.received += 1; 1 } fn init_seq(&mut self, seq: u16) { self.base_seq = seq as u32; self.max_seq = seq; self.bad_seq = RTP_SEQ_MOD + 1; /* so seq == bad_seq is false */ self.cycles = 0; self.received = 0; self.received_prior = 0; self.expected_prior = 0; /* other initialization */ self.base_seq = seq as u32; self.max_seq = seq; self.bad_seq = RTP_SEQ_MOD + 1; } } #[derive(Debug, Clone, Default)] pub struct RtcpContext { ssrc: u32, sender_ssrc: u32, sr_ntp_lsr: u64, sr_clock_time: u64, last_rtp_clock: u64, last_rtp_timestamp: u32, sample_rate: u32, send_bytes: u64, send_packets: u64, source: RtcpSource, } impl RtcpContext { pub fn new(ssrc: u32, seq: u16, sample_rate: u32) -> Self { RtcpContext { ssrc, source: RtcpSource { max_seq: seq - 1, probation: MIN_SEQUENTIAL, ..Default::default() }, sample_rate, ..Default::default() } } pub fn generate_app(&self, name: String, data: BytesMut) -> RtcpApp { let mut buf = BytesMut::with_capacity(name.len()); buf.extend_from_slice(name.as_bytes()); RtcpApp { ssrc: self.ssrc, name: buf, app_data: data, ..Default::default() } } pub fn generate_bye(&self) -> RtcpBye { let ssrss = vec![self.ssrc]; RtcpBye { header: RtcpHeader { report_count: 1, ..Default::default() }, ssrss, ..Default::default() } } //int rtcp_report_block(struct rtp_member* sender, uint8_t* ptr, int bytes) fn gen_report_block(&mut self) -> ReportBlock { let extend_max = self.source.cycles + self.source.max_seq as u32; let expected = extend_max - self.source.base_seq + 1; let lost = expected - self.source.received; let expected_interval = expected - self.source.expected_prior; self.source.expected_prior = expected; let received_interval = self.source.received - self.source.received_prior; self.source.received_prior = self.source.received; let lost_interval = expected_interval as i64 - received_interval as i64; let fraction = if expected_interval == 0 || lost_interval < 0 { 0 } else { ((lost_interval as u32) << 8) / expected_interval }; let delay = utils::current_time() - self.sr_clock_time; let lsr = self.sr_ntp_lsr >> 8 & 0xFFFFFFFF; let dlsr = (delay as f64 / 1000000. * 65535.) as u32; ReportBlock { cumutlative_num_of_packets_lost: lost, fraction_lost: fraction as u8, extended_highest_seq_number: extend_max, lsr: lsr as u32, dlsr, ssrc: self.sender_ssrc, jitter: self.source.jitter as u32, } } pub fn generate_rr(&mut self) -> RtcpReceiverReport { let block = self.gen_report_block(); RtcpReceiverReport { header: RtcpHeader { payload_type: RTCP_RR, report_count: 1, version: 2, length: (4 + 24) / 4, ..Default::default() }, report_blocks: vec![block], ssrc: self.ssrc, } } pub fn send_rtp(&mut self, pkt: RtpPacket) { self.send_bytes += pkt.payload.len() as u64; self.send_packets += 1; self.last_rtp_timestamp = pkt.header.timestamp; } pub fn received_sr(&mut self, sr: &RtcpSenderReport) { self.sr_clock_time = utils::current_time(); self.sr_ntp_lsr = sr.ntp; self.sender_ssrc = sr.ssrc; } pub fn received_rtp(&mut self, pkt: RtpPacket) { if 0 == self.source.update_sequence(pkt.header.seq_number) { return; } let rtp_clock = utils::current_time(); if self.last_rtp_clock == 0 { self.source.jitter = 0.; } else { let mut d = ((rtp_clock - self.last_rtp_clock) * self.sample_rate as u64 / 1000000) as i64 - (pkt.header.timestamp - self.last_rtp_timestamp) as i64; if d < 0 { d = -d; } self.source.jitter += (d as f64 - self.source.jitter) / 16.; } self.last_rtp_clock = rtp_clock; self.last_rtp_timestamp = pkt.header.timestamp; } }