active_call/media/
processor.rs

1use super::INTERNAL_SAMPLERATE;
2use super::track::track_codec::TrackCodec;
3use crate::event::{EventSender, SessionEvent};
4use crate::media::{AudioFrame, Samples, SourcePacket};
5use anyhow::Result;
6use std::any::Any;
7use std::sync::{Arc, Mutex};
8
9pub trait Processor: Send + Sync + Any {
10    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()>;
11}
12
13pub fn convert_to_mono(samples: &mut Vec<i16>, channels: u16) {
14    if channels != 2 {
15        return;
16    }
17    let mut i = 0;
18    let mut j = 0;
19    while i < samples.len() {
20        let l = samples[i] as i32;
21        let r = samples[i + 1] as i32;
22        samples[j] = ((l + r) / 2) as i16;
23        i += 2;
24        j += 1;
25    }
26    samples.truncate(j);
27}
28
29impl Default for AudioFrame {
30    fn default() -> Self {
31        Self {
32            track_id: "".to_string(),
33            samples: Samples::Empty,
34            timestamp: 0,
35            sample_rate: 16000,
36            channels: 1,
37            src_packet: None,
38            speech_probability: None,
39        }
40    }
41}
42
43impl Samples {
44    pub fn is_empty(&self) -> bool {
45        match self {
46            Samples::PCM { samples } => samples.is_empty(),
47            Samples::RTP { payload, .. } => payload.is_empty(),
48            Samples::Empty => true,
49        }
50    }
51}
52
53#[derive(Clone)]
54pub struct ProcessorChain {
55    processors: Arc<Mutex<Vec<Box<dyn Processor>>>>,
56    pub codec: TrackCodec,
57    sample_rate: u32,
58    pub force_decode: bool,
59}
60
61impl ProcessorChain {
62    pub fn new(_sample_rate: u32) -> Self {
63        Self {
64            processors: Arc::new(Mutex::new(Vec::new())),
65            codec: TrackCodec::new(),
66            sample_rate: INTERNAL_SAMPLERATE,
67            force_decode: true,
68        }
69    }
70    pub fn insert_processor(&mut self, processor: Box<dyn Processor>) {
71        self.processors.lock().unwrap().insert(0, processor);
72    }
73    pub fn append_processor(&mut self, processor: Box<dyn Processor>) {
74        self.processors.lock().unwrap().push(processor);
75    }
76
77    pub fn has_processor<T: 'static>(&self) -> bool {
78        let processors = self.processors.lock().unwrap();
79        processors
80            .iter()
81            .any(|processor| (processor.as_ref() as &dyn Any).is::<T>())
82    }
83
84    pub fn remove_processor<T: 'static>(&self) {
85        let mut processors = self.processors.lock().unwrap();
86        processors.retain(|processor| !(processor.as_ref() as &dyn Any).is::<T>());
87    }
88
89    pub fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
90        let mut processors = self.processors.lock().unwrap();
91        if !self.force_decode && processors.is_empty() {
92            return Ok(());
93        }
94        match &mut frame.samples {
95            Samples::RTP {
96                payload_type,
97                payload,
98                sequence_number,
99            } => {
100                if TrackCodec::is_audio(*payload_type) {
101                    let (decoded_sample_rate, channels, samples) =
102                        self.codec.decode(*payload_type, &payload, self.sample_rate);
103                    let src_packet = SourcePacket {
104                        sequence_number: *sequence_number,
105                        payload_type: *payload_type,
106                        payload: std::mem::take(payload),
107                    };
108                    frame.src_packet = Some(src_packet);
109                    frame.channels = channels;
110                    frame.samples = Samples::PCM { samples };
111                    frame.sample_rate = decoded_sample_rate;
112                }
113            }
114            _ => {}
115        }
116
117        if let Samples::PCM { samples } = &mut frame.samples {
118            if frame.sample_rate != self.sample_rate {
119                let new_samples = self.codec.resample(
120                    std::mem::take(samples),
121                    frame.sample_rate,
122                    self.sample_rate,
123                );
124                *samples = new_samples;
125                frame.sample_rate = self.sample_rate;
126            }
127            if frame.channels == 2 {
128                convert_to_mono(samples, 2);
129                frame.channels = 1;
130            }
131        }
132        // Process the frame with all processors
133        for processor in processors.iter_mut() {
134            processor.process_frame(frame)?;
135        }
136        Ok(())
137    }
138}
139
140pub struct SubscribeProcessor {
141    event_sender: EventSender,
142    track_id: String,
143    track_index: u8, // 0 for caller, 1 for callee
144}
145
146impl SubscribeProcessor {
147    pub fn new(event_sender: EventSender, track_id: String, track_index: u8) -> Self {
148        Self {
149            event_sender,
150            track_id,
151            track_index,
152        }
153    }
154}
155
156impl Processor for SubscribeProcessor {
157    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
158        if let Samples::PCM { samples } = &frame.samples {
159            if !samples.is_empty() {
160                let pcm_data = audio_codec::samples_to_bytes(samples);
161                let mut data = Vec::with_capacity(pcm_data.len() + 1);
162                data.push(self.track_index);
163                data.extend_from_slice(&pcm_data);
164
165                let event = SessionEvent::Binary {
166                    track_id: self.track_id.clone(),
167                    timestamp: frame.timestamp,
168                    data,
169                };
170                self.event_sender.send(event).ok();
171            }
172        }
173        Ok(())
174    }
175}
active_call/media/processor.rs

active_call/media/
processor.rs
