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};
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    pub raw_tap: Option<mpsc::UnboundedSender<AudioFrame>>,
65}
66
67impl ProcessorChain {
68    pub fn new(_sample_rate: u32) -> Self {
69        Self {
70            processors: Arc::new(Mutex::new(Vec::new())),
71            codec: TrackCodec::new(),
72            sample_rate: INTERNAL_SAMPLERATE,
73            force_decode: true,
74            raw_tap: None,
75        }
76    }
77    pub fn insert_processor(&mut self, processor: Box<dyn Processor>) {
78        self.processors.lock().unwrap().insert(0, processor);
79    }
80    pub fn append_processor(&mut self, processor: Box<dyn Processor>) {
81        self.processors.lock().unwrap().push(processor);
82    }
83
84    pub fn has_processor<T: 'static>(&self) -> bool {
85        let processors = self.processors.lock().unwrap();
86        processors
87            .iter()
88            .any(|processor| (processor.as_ref() as &dyn Any).is::<T>())
89    }
90
91    pub fn remove_processor<T: 'static>(&self) {
92        let mut processors = self.processors.lock().unwrap();
93        processors.retain(|processor| !(processor.as_ref() as &dyn Any).is::<T>());
94    }
95
96    pub fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
97        let mut processors = self.processors.lock().unwrap();
98        if !self.force_decode && processors.is_empty() && self.raw_tap.is_none() {
99            return Ok(());
100        }
101        match &mut frame.samples {
102            Samples::RTP {
103                payload_type,
104                payload,
105                sequence_number,
106            } => {
107                if TrackCodec::is_audio(*payload_type) {
108                    let (decoded_sample_rate, channels, samples) =
109                        self.codec.decode(*payload_type, &payload);
110                    let src_packet = SourcePacket {
111                        sequence_number: *sequence_number,
112                        payload_type: *payload_type,
113                        payload: std::mem::take(payload),
114                    };
115                    frame.src_packet = Some(src_packet);
116                    frame.channels = channels;
117                    frame.samples = Samples::PCM { samples };
118                    frame.sample_rate = decoded_sample_rate;
119                }
120            }
121            _ => {}
122        }
123
124        // Mirror the frame to the raw tap at its native sample rate, before
125        // the pipeline resamples it to INTERNAL_SAMPLERATE.
126        if let Some(tap) = &self.raw_tap
127            && let Samples::PCM { samples } = &frame.samples
128            && !samples.is_empty()
129            && frame.sample_rate > 0
130        {
131            let mut raw = frame.clone();
132            raw.src_packet = None;
133            let mono = match &mut raw.samples {
134                Samples::PCM { samples } => samples,
135                _ => unreachable!("checked PCM above"),
136            };
137            if raw.channels == 2 {
138                convert_to_mono(mono, 2);
139                raw.channels = 1;
140            }
141            let _ = tap.send(raw);
142        }
143
144        if let Samples::PCM { samples } = &mut frame.samples {
145            if frame.sample_rate != self.sample_rate {
146                let new_samples = self.codec.resample(
147                    std::mem::take(samples),
148                    frame.sample_rate,
149                    self.sample_rate,
150                );
151                *samples = new_samples;
152                frame.sample_rate = self.sample_rate;
153            }
154            if frame.channels == 2 {
155                convert_to_mono(samples, 2);
156                frame.channels = 1;
157            }
158        }
159        // Process the frame with all processors
160        for processor in processors.iter_mut() {
161            processor.process_frame(frame)?;
162        }
163        Ok(())
164    }
165}
166
167pub struct SubscribeProcessor {
168    event_sender: EventSender,
169    track_id: String,
170    track_index: u8, // 0 for caller, 1 for callee
171}
172
173impl SubscribeProcessor {
174    pub fn new(event_sender: EventSender, track_id: String, track_index: u8) -> Self {
175        Self {
176            event_sender,
177            track_id,
178            track_index,
179        }
180    }
181}
182
183impl Processor for SubscribeProcessor {
184    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
185        if let Samples::PCM { samples } = &frame.samples {
186            if !samples.is_empty() {
187                let pcm_data = audio_codec::samples_to_bytes(samples);
188                let mut data = Vec::with_capacity(pcm_data.len() + 1);
189                data.push(self.track_index);
190                data.extend_from_slice(&pcm_data);
191
192                let event = SessionEvent::Binary {
193                    track_id: self.track_id.clone(),
194                    timestamp: frame.timestamp,
195                    data,
196                };
197                self.event_sender.send(event).ok();
198            }
199        }
200        Ok(())
201    }
202}
active_call/media/processor.rs

active_call/media/
processor.rs
