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, RwLock};
8use tokio::sync::mpsc;
9
10pub trait Processor: Send + Sync + Any {
11    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()>;
12}
13
14pub fn convert_to_mono(samples: &mut Vec<i16>, channels: u16) {
15    if channels != 2 {
16        return;
17    }
18    let mut i = 0;
19    let mut j = 0;
20    while i < samples.len() {
21        let l = samples[i] as i32;
22        let r = samples[i + 1] as i32;
23        samples[j] = ((l + r) / 2) as i16;
24        i += 2;
25        j += 1;
26    }
27    samples.truncate(j);
28}
29
30impl Default for AudioFrame {
31    fn default() -> Self {
32        Self {
33            track_id: "".to_string(),
34            samples: Samples::Empty,
35            timestamp: 0,
36            sample_rate: 16000,
37            channels: 1,
38            src_packet: None,
39            speech_probability: None,
40        }
41    }
42}
43
44impl Samples {
45    pub fn is_empty(&self) -> bool {
46        match self {
47            Samples::PCM { samples } => samples.is_empty(),
48            Samples::RTP { payload, .. } => payload.is_empty(),
49            Samples::Empty => true,
50        }
51    }
52}
53
54#[derive(Clone)]
55pub struct ProcessorChain {
56    processors: Arc<Mutex<Vec<Box<dyn Processor>>>>,
57    pub codec: TrackCodec,
58    sample_rate: u32,
59    pub force_decode: bool,
60    /// Optional raw tap: when set, frames are mirrored to this channel at
61    /// their native (pre-resample) sample rate right after decoding, before
62    /// the pipeline normalizes them to `INTERNAL_SAMPLERATE`. Used by the
63    /// native-samplerate recorder.
64    ///
65    /// Shared via `Arc<RwLock<..>>`: `RtcTrack::create()` clones the chain
66    /// into its long-lived worker tasks *before* the recorder attaches the
67    /// tap, so a plain field would leave those workers with a stale `None`.
68    pub raw_tap: Arc<RwLock<Option<mpsc::UnboundedSender<AudioFrame>>>>,
69}
70
71impl ProcessorChain {
72    pub fn new(_sample_rate: u32) -> Self {
73        Self {
74            processors: Arc::new(Mutex::new(Vec::new())),
75            codec: TrackCodec::new(),
76            sample_rate: INTERNAL_SAMPLERATE,
77            force_decode: true,
78            raw_tap: Arc::new(RwLock::new(None)),
79        }
80    }
81    pub fn set_raw_tap(&mut self, tap: Option<mpsc::UnboundedSender<AudioFrame>>) {
82        *self.raw_tap.write().unwrap() = tap;
83    }
84    fn raw_tap(&self) -> Option<mpsc::UnboundedSender<AudioFrame>> {
85        self.raw_tap.read().unwrap().clone()
86    }
87    pub fn insert_processor(&mut self, processor: Box<dyn Processor>) {
88        self.processors.lock().unwrap().insert(0, processor);
89    }
90    pub fn append_processor(&mut self, processor: Box<dyn Processor>) {
91        self.processors.lock().unwrap().push(processor);
92    }
93
94    pub fn has_processor<T: 'static>(&self) -> bool {
95        let processors = self.processors.lock().unwrap();
96        processors
97            .iter()
98            .any(|processor| (processor.as_ref() as &dyn Any).is::<T>())
99    }
100
101    pub fn remove_processor<T: 'static>(&self) {
102        let mut processors = self.processors.lock().unwrap();
103        processors.retain(|processor| !(processor.as_ref() as &dyn Any).is::<T>());
104    }
105
106    pub fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
107        let mut processors = self.processors.lock().unwrap();
108        if !self.force_decode && processors.is_empty() && self.raw_tap().is_none() {
109            return Ok(());
110        }
111        match &mut frame.samples {
112            Samples::RTP {
113                payload_type,
114                payload,
115                sequence_number,
116            } => {
117                if TrackCodec::is_audio(*payload_type) {
118                    let (decoded_sample_rate, channels, samples) =
119                        self.codec.decode(*payload_type, &payload);
120                    let src_packet = SourcePacket {
121                        sequence_number: *sequence_number,
122                        payload_type: *payload_type,
123                        payload: std::mem::take(payload),
124                    };
125                    frame.src_packet = Some(src_packet);
126                    frame.channels = channels;
127                    frame.samples = Samples::PCM { samples };
128                    frame.sample_rate = decoded_sample_rate;
129                }
130            }
131            _ => {}
132        }
133
134        // Mirror the frame to the raw tap at its native sample rate, before
135        // the pipeline resamples it to INTERNAL_SAMPLERATE.
136        if let Some(tap) = self.raw_tap()
137            && let Samples::PCM { samples } = &frame.samples
138            && !samples.is_empty()
139            && frame.sample_rate > 0
140        {
141            let mut raw = frame.clone();
142            raw.src_packet = None;
143            let mono = match &mut raw.samples {
144                Samples::PCM { samples } => samples,
145                _ => unreachable!("checked PCM above"),
146            };
147            if raw.channels == 2 {
148                convert_to_mono(mono, 2);
149                raw.channels = 1;
150            }
151            let _ = tap.send(raw);
152        }
153
154        if let Samples::PCM { samples } = &mut frame.samples {
155            if frame.sample_rate != self.sample_rate {
156                let new_samples = self.codec.resample(
157                    std::mem::take(samples),
158                    frame.sample_rate,
159                    self.sample_rate,
160                );
161                *samples = new_samples;
162                frame.sample_rate = self.sample_rate;
163            }
164            if frame.channels == 2 {
165                convert_to_mono(samples, 2);
166                frame.channels = 1;
167            }
168        }
169        // Process the frame with all processors
170        for processor in processors.iter_mut() {
171            processor.process_frame(frame)?;
172        }
173        Ok(())
174    }
175}
176
177pub struct SubscribeProcessor {
178    event_sender: EventSender,
179    track_id: String,
180    track_index: u8, // 0 for caller, 1 for callee
181}
182
183impl SubscribeProcessor {
184    pub fn new(event_sender: EventSender, track_id: String, track_index: u8) -> Self {
185        Self {
186            event_sender,
187            track_id,
188            track_index,
189        }
190    }
191}
192
193impl Processor for SubscribeProcessor {
194    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
195        if let Samples::PCM { samples } = &frame.samples {
196            if !samples.is_empty() {
197                let pcm_data = audio_codec::samples_to_bytes(samples);
198                let mut data = Vec::with_capacity(pcm_data.len() + 1);
199                data.push(self.track_index);
200                data.extend_from_slice(&pcm_data);
201
202                let event = SessionEvent::Binary {
203                    track_id: self.track_id.clone(),
204                    timestamp: frame.timestamp,
205                    data,
206                };
207                self.event_sender.send(event).ok();
208            }
209        }
210        Ok(())
211    }
212}
213
214#[cfg(test)]
215mod tests {
216    use super::*;
217
218    /// Regression: `RtcTrack::create()` clones the chain into its worker
219    /// tasks *before* the recorder attaches the raw tap (native-samplerate
220    /// recording). The tap must be shared state so clones taken before the
221    /// attach observe it; with a plain `Option` field this test fails and
222    /// SIP/RTP calls silently fall back to the 16 kHz recorder.
223    #[test]
224    fn raw_tap_visible_to_clones_taken_before_attach() {
225        let mut chain = ProcessorChain::new(INTERNAL_SAMPLERATE);
226        let mut worker = chain.clone();
227
228        let (tx, mut rx) = mpsc::unbounded_channel();
229        chain.set_raw_tap(Some(tx));
230
231        let mut frame = AudioFrame {
232            track_id: "track".to_string(),
233            samples: Samples::PCM {
234                samples: vec![100i16; 160],
235            },
236            sample_rate: 8000,
237            channels: 1,
238            ..Default::default()
239        };
240        worker.process_frame(&mut frame).unwrap();
241
242        let raw = rx
243            .try_recv()
244            .expect("raw tap should receive a frame from a pre-attach clone");
245        assert_eq!(raw.sample_rate, 8000);
246        assert_eq!(raw.channels, 1);
247        match raw.samples {
248            Samples::PCM { samples } => assert_eq!(samples.len(), 160),
249            _ => panic!("expected PCM samples on the raw tap"),
250        }
251        // The pipeline output is still normalized to the internal rate.
252        assert_eq!(frame.sample_rate, INTERNAL_SAMPLERATE);
253    }
254
255    /// Detaching the tap on the original chain must be observed by clones as
256    /// well (e.g. recorder restart swapping the sender).
257    #[test]
258    fn raw_tap_detach_visible_to_clones() {
259        let mut chain = ProcessorChain::new(INTERNAL_SAMPLERATE);
260        let (tx, _rx) = mpsc::unbounded_channel();
261        chain.set_raw_tap(Some(tx));
262        let mut worker = chain.clone();
263        chain.set_raw_tap(None);
264
265        let mut frame = AudioFrame {
266            track_id: "track".to_string(),
267            samples: Samples::PCM {
268                samples: vec![100i16; 160],
269            },
270            sample_rate: 8000,
271            channels: 1,
272            ..Default::default()
273        };
274        worker.process_frame(&mut frame).unwrap();
275        assert_eq!(frame.sample_rate, INTERNAL_SAMPLERATE);
276    }
277}
active_call/media/processor.rs

active_call/media/
processor.rs
