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
71/// Lock the processor list, recovering from a poisoned mutex.
72///
73/// A panic inside a processor (or in the decode/resample step) must not
74/// permanently take down the track: the `Vec` itself is never left in an
75/// inconsistent state by unwinding, so reusing the guard is safe and far
76/// better than propagating a `PoisonError` on every subsequent frame.
77///
78/// Takes the mutex directly (rather than `&self`) so the returned guard only
79/// borrows the `processors` field, leaving `self.codec` free to be borrowed
80/// mutably for the decode/resample steps in `process_frame`.
81fn lock_processors(
82    processors: &Mutex<Vec<Box<dyn Processor>>>,
83) -> std::sync::MutexGuard<'_, Vec<Box<dyn Processor>>> {
84    processors.lock().unwrap_or_else(|e| e.into_inner())
85}
86
87impl ProcessorChain {
88    pub fn new(_sample_rate: u32) -> Self {
89        Self {
90            processors: Arc::new(Mutex::new(Vec::new())),
91            codec: TrackCodec::new(),
92            sample_rate: INTERNAL_SAMPLERATE,
93            force_decode: true,
94            raw_tap: Arc::new(RwLock::new(None)),
95        }
96    }
97    pub fn set_raw_tap(&mut self, tap: Option<mpsc::UnboundedSender<AudioFrame>>) {
98        *self.raw_tap.write().unwrap() = tap;
99    }
100    fn raw_tap(&self) -> Option<mpsc::UnboundedSender<AudioFrame>> {
101        self.raw_tap.read().unwrap().clone()
102    }
103    pub fn insert_processor(&mut self, processor: Box<dyn Processor>) {
104        lock_processors(&self.processors).insert(0, processor);
105    }
106    pub fn append_processor(&mut self, processor: Box<dyn Processor>) {
107        lock_processors(&self.processors).push(processor);
108    }
109
110    pub fn has_processor<T: 'static>(&self) -> bool {
111        let processors = lock_processors(&self.processors);
112        processors
113            .iter()
114            .any(|processor| (processor.as_ref() as &dyn Any).is::<T>())
115    }
116
117    pub fn remove_processor<T: 'static>(&self) {
118        let mut processors = lock_processors(&self.processors);
119        processors.retain(|processor| !(processor.as_ref() as &dyn Any).is::<T>());
120    }
121
122    pub fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
123        let mut processors = lock_processors(&self.processors);
124        if !self.force_decode && processors.is_empty() && self.raw_tap().is_none() {
125            return Ok(());
126        }
127        match &mut frame.samples {
128            Samples::RTP {
129                payload_type,
130                payload,
131                sequence_number,
132            } => {
133                if TrackCodec::is_audio(*payload_type) {
134                    let (decoded_sample_rate, channels, samples) =
135                        self.codec.decode(*payload_type, &payload);
136                    let src_packet = SourcePacket {
137                        sequence_number: *sequence_number,
138                        payload_type: *payload_type,
139                        payload: std::mem::take(payload),
140                    };
141                    frame.src_packet = Some(src_packet);
142                    frame.channels = channels;
143                    frame.samples = Samples::PCM { samples };
144                    frame.sample_rate = decoded_sample_rate;
145                }
146            }
147            _ => {}
148        }
149
150        // Mirror the frame to the raw tap at its native sample rate, before
151        // the pipeline resamples it to INTERNAL_SAMPLERATE.
152        if let Some(tap) = self.raw_tap()
153            && let Samples::PCM { samples } = &frame.samples
154            && !samples.is_empty()
155            && frame.sample_rate > 0
156        {
157            let mut raw = frame.clone();
158            raw.src_packet = None;
159            let mono = match &mut raw.samples {
160                Samples::PCM { samples } => samples,
161                _ => unreachable!("checked PCM above"),
162            };
163            if raw.channels == 2 {
164                convert_to_mono(mono, 2);
165                raw.channels = 1;
166            }
167            let _ = tap.send(raw);
168        }
169
170        if let Samples::PCM { samples } = &mut frame.samples {
171            if frame.sample_rate != self.sample_rate {
172                let new_samples = self.codec.resample(
173                    std::mem::take(samples),
174                    frame.sample_rate,
175                    self.sample_rate,
176                );
177                *samples = new_samples;
178                frame.sample_rate = self.sample_rate;
179            }
180            if frame.channels == 2 {
181                convert_to_mono(samples, 2);
182                frame.channels = 1;
183            }
184        }
185        // Process the frame with all processors
186        for processor in processors.iter_mut() {
187            processor.process_frame(frame)?;
188        }
189        Ok(())
190    }
191}
192
193pub struct SubscribeProcessor {
194    event_sender: EventSender,
195    track_id: String,
196    track_index: u8, // 0 for caller, 1 for callee
197}
198
199impl SubscribeProcessor {
200    pub fn new(event_sender: EventSender, track_id: String, track_index: u8) -> Self {
201        Self {
202            event_sender,
203            track_id,
204            track_index,
205        }
206    }
207}
208
209impl Processor for SubscribeProcessor {
210    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
211        if let Samples::PCM { samples } = &frame.samples {
212            if !samples.is_empty() {
213                let pcm_data = audio_codec::samples_to_bytes(samples);
214                let mut data = Vec::with_capacity(pcm_data.len() + 1);
215                data.push(self.track_index);
216                data.extend_from_slice(&pcm_data);
217
218                let event = SessionEvent::Binary {
219                    track_id: self.track_id.clone(),
220                    timestamp: frame.timestamp,
221                    data,
222                };
223                self.event_sender.send(event).ok();
224            }
225        }
226        Ok(())
227    }
228}
229
230#[cfg(test)]
231mod tests {
232    use super::*;
233
234    /// Regression: `RtcTrack::create()` clones the chain into its worker
235    /// tasks *before* the recorder attaches the raw tap (native-samplerate
236    /// recording). The tap must be shared state so clones taken before the
237    /// attach observe it; with a plain `Option` field this test fails and
238    /// SIP/RTP calls silently fall back to the 16 kHz recorder.
239    #[test]
240    fn raw_tap_visible_to_clones_taken_before_attach() {
241        let mut chain = ProcessorChain::new(INTERNAL_SAMPLERATE);
242        let mut worker = chain.clone();
243
244        let (tx, mut rx) = mpsc::unbounded_channel();
245        chain.set_raw_tap(Some(tx));
246
247        let mut frame = AudioFrame {
248            track_id: "track".to_string(),
249            samples: Samples::PCM {
250                samples: vec![100i16; 160],
251            },
252            sample_rate: 8000,
253            channels: 1,
254            ..Default::default()
255        };
256        worker.process_frame(&mut frame).unwrap();
257
258        let raw = rx
259            .try_recv()
260            .expect("raw tap should receive a frame from a pre-attach clone");
261        assert_eq!(raw.sample_rate, 8000);
262        assert_eq!(raw.channels, 1);
263        match raw.samples {
264            Samples::PCM { samples } => assert_eq!(samples.len(), 160),
265            _ => panic!("expected PCM samples on the raw tap"),
266        }
267        // The pipeline output is still normalized to the internal rate.
268        assert_eq!(frame.sample_rate, INTERNAL_SAMPLERATE);
269    }
270
271    /// Detaching the tap on the original chain must be observed by clones as
272    /// well (e.g. recorder restart swapping the sender).
273    #[test]
274    fn raw_tap_detach_visible_to_clones() {
275        let mut chain = ProcessorChain::new(INTERNAL_SAMPLERATE);
276        let (tx, _rx) = mpsc::unbounded_channel();
277        chain.set_raw_tap(Some(tx));
278        let mut worker = chain.clone();
279        chain.set_raw_tap(None);
280
281        let mut frame = AudioFrame {
282            track_id: "track".to_string(),
283            samples: Samples::PCM {
284                samples: vec![100i16; 160],
285            },
286            sample_rate: 8000,
287            channels: 1,
288            ..Default::default()
289        };
290        worker.process_frame(&mut frame).unwrap();
291        assert_eq!(frame.sample_rate, INTERNAL_SAMPLERATE);
292    }
293
294    /// Regression: a track report with no valid source rate (e.g. a media-pass
295    /// track whose `input_sample_rate` is 0) must not panic by trying to build
296    /// a resampler from 0 Hz, which previously poisoned the processor mutex.
297    #[test]
298    fn process_frame_tolerates_zero_sample_rate() {
299        let mut chain = ProcessorChain::new(INTERNAL_SAMPLERATE);
300        let mut frame = AudioFrame {
301            track_id: "track".to_string(),
302            samples: Samples::PCM {
303                samples: vec![100i16; 160],
304            },
305            sample_rate: 0,
306            channels: 1,
307            ..Default::default()
308        };
309
310        chain.process_frame(&mut frame).unwrap();
311        assert_eq!(frame.sample_rate, INTERNAL_SAMPLERATE);
312        match frame.samples {
313            Samples::PCM { samples } => assert_eq!(samples.len(), 160),
314            _ => panic!("expected PCM samples"),
315        }
316    }
317}
active_call/media/processor.rs

active_call/media/
processor.rs
