use crate::{Result, RtpSsrc};
use crate::feedback::{
FeedbackGenerator, FeedbackContext, FeedbackConfig, FeedbackDecision,
FeedbackPriority, QualityDegradation, CongestionState
};
use crate::api::common::stats::StreamStats;
use std::time::{Instant, Duration};
use std::collections::VecDeque;
pub struct LossFeedbackGenerator {
loss_history: VecDeque<LossEvent>,
last_pli: Option<Instant>,
last_fir: Option<Instant>,
fir_sequence: u8,
consecutive_loss: u16,
total_packets: u32,
total_lost: u32,
}
#[derive(Debug, Clone)]
struct LossEvent {
timestamp: Instant,
loss_rate: f32,
consecutive_count: u16,
}
impl LossFeedbackGenerator {
pub fn new() -> Self {
Self {
loss_history: VecDeque::new(),
last_pli: None,
last_fir: None,
fir_sequence: 0,
consecutive_loss: 0,
total_packets: 0,
total_lost: 0,
}
}
fn current_loss_rate(&self) -> f32 {
if self.total_packets == 0 {
return 0.0;
}
self.total_lost as f32 / self.total_packets as f32
}
fn should_generate_pli(&self, context: &FeedbackContext, config: &FeedbackConfig) -> bool {
if !config.enable_pli {
return false;
}
if let Some(last) = self.last_pli {
if last.elapsed().as_millis() < config.pli_interval_ms as u128 {
return false;
}
}
let loss_rate = self.current_loss_rate();
loss_rate > 0.01 ||
self.consecutive_loss > 3 ||
matches!(context.congestion_state, CongestionState::Moderate | CongestionState::Severe | CongestionState::Critical)
}
fn should_generate_fir(&self, context: &FeedbackContext, config: &FeedbackConfig) -> bool {
if !config.enable_fir {
return false;
}
if let Some(last) = self.last_fir {
if last.elapsed().as_millis() < config.fir_interval_ms as u128 {
return false;
}
}
let loss_rate = self.current_loss_rate();
loss_rate > 0.05 ||
self.consecutive_loss > 10 ||
matches!(context.congestion_state, CongestionState::Critical)
}
fn calculate_priority(&self, loss_rate: f32) -> FeedbackPriority {
match loss_rate {
rate if rate > 0.10 => FeedbackPriority::Critical, rate if rate > 0.05 => FeedbackPriority::High, rate if rate > 0.02 => FeedbackPriority::Normal, _ => FeedbackPriority::Low,
}
}
}
impl FeedbackGenerator for LossFeedbackGenerator {
fn generate_feedback(&self, context: &FeedbackContext, config: &FeedbackConfig) -> Result<FeedbackDecision> {
let loss_rate = self.current_loss_rate();
if self.should_generate_fir(context, config) {
return Ok(FeedbackDecision::Fir {
priority: FeedbackPriority::Critical,
sequence_number: self.fir_sequence,
});
}
if self.should_generate_pli(context, config) {
let priority = self.calculate_priority(loss_rate);
let reason = QualityDegradation::PacketLoss {
rate: (loss_rate * 100.0) as u8,
consecutive: self.consecutive_loss,
};
return Ok(FeedbackDecision::Pli {
priority,
reason,
});
}
Ok(FeedbackDecision::None)
}
fn update_statistics(&mut self, stats: &StreamStats) {
self.total_packets = stats.packet_count as u32;
self.total_lost = stats.packets_lost as u32;
if stats.packets_lost > 0 {
self.consecutive_loss += 1;
} else {
self.consecutive_loss = 0;
}
let loss_rate = self.current_loss_rate();
if loss_rate > 0.0 {
self.loss_history.push_back(LossEvent {
timestamp: Instant::now(),
loss_rate,
consecutive_count: self.consecutive_loss,
});
let cutoff = Instant::now() - Duration::from_secs(10);
while let Some(event) = self.loss_history.front() {
if event.timestamp < cutoff {
self.loss_history.pop_front();
} else {
break;
}
}
}
}
fn name(&self) -> &'static str {
"LossFeedbackGenerator"
}
}
pub struct CongestionFeedbackGenerator {
bandwidth_history: VecDeque<BandwidthSample>,
last_remb: Option<Instant>,
estimated_bandwidth: u32,
bandwidth_trend: BandwidthTrend,
confidence: f32,
}
#[derive(Debug, Clone)]
struct BandwidthSample {
timestamp: Instant,
bandwidth_bps: u32,
rtt_ms: u32,
loss_rate: f32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BandwidthTrend {
Increasing,
Stable,
Decreasing,
Unknown,
}
impl CongestionFeedbackGenerator {
pub fn new() -> Self {
Self {
bandwidth_history: VecDeque::new(),
last_remb: None,
estimated_bandwidth: 1_000_000, bandwidth_trend: BandwidthTrend::Unknown,
confidence: 0.5, }
}
fn update_bandwidth_estimate(&mut self, rtt_ms: u32, loss_rate: f32, jitter: u32) {
let congestion_factor = self.calculate_congestion_factor(rtt_ms, loss_rate, jitter);
if congestion_factor > 0.8 {
self.estimated_bandwidth = (self.estimated_bandwidth as f32 * 0.85) as u32;
self.bandwidth_trend = BandwidthTrend::Decreasing;
self.confidence = 0.8; } else if congestion_factor < 0.3 {
self.estimated_bandwidth = (self.estimated_bandwidth as f32 * 1.05) as u32;
self.bandwidth_trend = BandwidthTrend::Increasing;
self.confidence = 0.6; } else {
self.bandwidth_trend = BandwidthTrend::Stable;
self.confidence = 0.9; }
self.estimated_bandwidth = self.estimated_bandwidth.clamp(64_000, 50_000_000);
self.bandwidth_history.push_back(BandwidthSample {
timestamp: Instant::now(),
bandwidth_bps: self.estimated_bandwidth,
rtt_ms,
loss_rate,
});
let cutoff = Instant::now() - Duration::from_secs(30);
while let Some(sample) = self.bandwidth_history.front() {
if sample.timestamp < cutoff {
self.bandwidth_history.pop_front();
} else {
break;
}
}
}
fn calculate_congestion_factor(&self, rtt_ms: u32, loss_rate: f32, jitter: u32) -> f32 {
let rtt_weight = 0.4;
let loss_weight = 0.4;
let jitter_weight = 0.2;
let normalized_rtt = (rtt_ms as f32 / 500.0).clamp(0.0, 1.0); let normalized_loss = (loss_rate * 20.0).clamp(0.0, 1.0); let normalized_jitter = (jitter as f32 / 50.0).clamp(0.0, 1.0);
normalized_rtt * rtt_weight +
normalized_loss * loss_weight +
normalized_jitter * jitter_weight
}
fn should_generate_remb(&self, config: &FeedbackConfig) -> bool {
if !config.enable_remb {
return false;
}
if let Some(last) = self.last_remb {
let elapsed = last.elapsed().as_millis();
if elapsed > 2000 {
return true;
}
if self.confidence > 0.7 && elapsed > 500 {
match self.bandwidth_trend {
BandwidthTrend::Increasing | BandwidthTrend::Decreasing => true,
_ => false,
}
} else {
false
}
} else {
true }
}
}
impl FeedbackGenerator for CongestionFeedbackGenerator {
fn generate_feedback(&self, _context: &FeedbackContext, config: &FeedbackConfig) -> Result<FeedbackDecision> {
if self.should_generate_remb(config) {
Ok(FeedbackDecision::Remb {
bitrate_bps: self.estimated_bandwidth,
confidence: self.confidence,
})
} else {
Ok(FeedbackDecision::None)
}
}
fn update_statistics(&mut self, stats: &StreamStats) {
let rtt_ms = stats.rtt_ms.unwrap_or(100.0) as u32; let loss_rate = if stats.packet_count > 0 {
stats.packets_lost as f32 / stats.packet_count as f32
} else {
0.0
};
let jitter = stats.jitter_ms as u32;
self.update_bandwidth_estimate(rtt_ms, loss_rate, jitter);
}
fn name(&self) -> &'static str {
"CongestionFeedbackGenerator"
}
}
pub struct QualityFeedbackGenerator {
quality_history: VecDeque<QualityScore>,
last_feedback: Option<Instant>,
quality_trend: QualityTrend,
quality_thresholds: QualityThresholds,
}
#[derive(Debug, Clone)]
struct QualityScore {
timestamp: Instant,
overall_score: f32, loss_score: f32,
jitter_score: f32,
bandwidth_score: f32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum QualityTrend {
Improving,
Stable,
Degrading,
Unknown,
}
#[derive(Debug, Clone)]
struct QualityThresholds {
excellent: f32, good: f32, fair: f32, poor: f32, }
impl Default for QualityThresholds {
fn default() -> Self {
Self {
excellent: 0.9,
good: 0.7,
fair: 0.5,
poor: 0.3,
}
}
}
impl QualityFeedbackGenerator {
pub fn new() -> Self {
Self {
quality_history: VecDeque::new(),
last_feedback: None,
quality_trend: QualityTrend::Unknown,
quality_thresholds: QualityThresholds::default(),
}
}
fn calculate_quality_score(&self, loss_rate: f32, jitter: u32, bandwidth_ratio: f32) -> QualityScore {
let loss_score = (1.0 - (loss_rate * 10.0)).clamp(0.0, 1.0); let jitter_score = (1.0 - (jitter as f32 / 100.0)).clamp(0.0, 1.0); let bandwidth_score = bandwidth_ratio.clamp(0.0, 1.0);
let overall_score = (loss_score * 0.4 + jitter_score * 0.3 + bandwidth_score * 0.3).clamp(0.0, 1.0);
QualityScore {
timestamp: Instant::now(),
overall_score,
loss_score,
jitter_score,
bandwidth_score,
}
}
fn update_quality_trend(&mut self) {
if self.quality_history.len() < 3 {
self.quality_trend = QualityTrend::Unknown;
return;
}
let recent: Vec<f32> = self.quality_history
.iter()
.rev()
.take(3)
.map(|q| q.overall_score)
.collect();
if recent[0] > recent[1] && recent[1] > recent[2] {
self.quality_trend = QualityTrend::Improving;
} else if recent[0] < recent[1] && recent[1] < recent[2] {
self.quality_trend = QualityTrend::Degrading;
} else {
self.quality_trend = QualityTrend::Stable;
}
}
fn determine_feedback(&self, current_quality: &QualityScore) -> FeedbackDecision {
let score = current_quality.overall_score;
match score {
s if s < self.quality_thresholds.poor => {
if current_quality.loss_score < 0.3 {
FeedbackDecision::Fir {
priority: FeedbackPriority::Critical,
sequence_number: 1, }
} else {
FeedbackDecision::Pli {
priority: FeedbackPriority::Critical,
reason: QualityDegradation::PacketLoss {
rate: ((1.0 - current_quality.loss_score) * 100.0) as u8,
consecutive: 0, },
}
}
}
s if s < self.quality_thresholds.fair => {
FeedbackDecision::Pli {
priority: FeedbackPriority::High,
reason: QualityDegradation::PacketLoss {
rate: ((1.0 - current_quality.loss_score) * 100.0) as u8,
consecutive: 0,
},
}
}
s if s < self.quality_thresholds.good => {
let estimated_bandwidth = 1_000_000; FeedbackDecision::Remb {
bitrate_bps: estimated_bandwidth,
confidence: 0.7,
}
}
_ => {
FeedbackDecision::None
}
}
}
}
impl FeedbackGenerator for QualityFeedbackGenerator {
fn generate_feedback(&self, _context: &FeedbackContext, _config: &FeedbackConfig) -> Result<FeedbackDecision> {
if let Some(latest_quality) = self.quality_history.back() {
match self.quality_trend {
QualityTrend::Degrading | QualityTrend::Unknown => {
Ok(self.determine_feedback(latest_quality))
}
_ => {
if latest_quality.overall_score < self.quality_thresholds.fair {
Ok(self.determine_feedback(latest_quality))
} else {
Ok(FeedbackDecision::None)
}
}
}
} else {
Ok(FeedbackDecision::None)
}
}
fn update_statistics(&mut self, stats: &StreamStats) {
let loss_rate = if stats.packet_count > 0 {
stats.packets_lost as f32 / stats.packet_count as f32
} else {
0.0
};
let bandwidth_ratio = 1.0;
let quality_score = self.calculate_quality_score(loss_rate, stats.jitter_ms as u32, bandwidth_ratio);
self.quality_history.push_back(quality_score);
let cutoff = Instant::now() - Duration::from_secs(60);
while let Some(score) = self.quality_history.front() {
if score.timestamp < cutoff {
self.quality_history.pop_front();
} else {
break;
}
}
self.update_quality_trend();
}
fn name(&self) -> &'static str {
"QualityFeedbackGenerator"
}
}
pub struct ComprehensiveFeedbackGenerator {
loss_generator: LossFeedbackGenerator,
congestion_generator: CongestionFeedbackGenerator,
quality_generator: QualityFeedbackGenerator,
last_feedback: Option<Instant>,
}
impl ComprehensiveFeedbackGenerator {
pub fn new() -> Self {
Self {
loss_generator: LossFeedbackGenerator::new(),
congestion_generator: CongestionFeedbackGenerator::new(),
quality_generator: QualityFeedbackGenerator::new(),
last_feedback: None,
}
}
fn combine_feedback_decisions(&self, decisions: Vec<FeedbackDecision>) -> FeedbackDecision {
let mut high_priority = Vec::new();
let mut normal_priority = Vec::new();
for decision in decisions {
match decision {
FeedbackDecision::None => continue,
FeedbackDecision::Pli { priority, .. } |
FeedbackDecision::Fir { priority, .. } => {
if priority >= FeedbackPriority::High {
high_priority.push(decision);
} else {
normal_priority.push(decision);
}
}
_ => normal_priority.push(decision),
}
}
if !high_priority.is_empty() {
if high_priority.len() == 1 {
high_priority.into_iter().next().unwrap()
} else {
FeedbackDecision::Multiple(high_priority)
}
} else if !normal_priority.is_empty() {
if normal_priority.len() == 1 {
normal_priority.into_iter().next().unwrap()
} else {
FeedbackDecision::Multiple(normal_priority)
}
} else {
FeedbackDecision::None
}
}
}
impl FeedbackGenerator for ComprehensiveFeedbackGenerator {
fn generate_feedback(&self, context: &FeedbackContext, config: &FeedbackConfig) -> Result<FeedbackDecision> {
if let Some(last) = self.last_feedback {
let interval_ms = 1000 / config.max_feedback_rate; if last.elapsed().as_millis() < interval_ms as u128 {
return Ok(FeedbackDecision::None);
}
}
let mut decisions = Vec::new();
decisions.push(self.loss_generator.generate_feedback(context, config)?);
decisions.push(self.congestion_generator.generate_feedback(context, config)?);
decisions.push(self.quality_generator.generate_feedback(context, config)?);
Ok(self.combine_feedback_decisions(decisions))
}
fn update_statistics(&mut self, stats: &StreamStats) {
self.loss_generator.update_statistics(stats);
self.congestion_generator.update_statistics(stats);
self.quality_generator.update_statistics(stats);
}
fn name(&self) -> &'static str {
"ComprehensiveFeedbackGenerator"
}
}