active_call/media/
stream.rs

1use crate::event::{EventSender, SessionEvent};
2use crate::media::ambiance::{AmbianceOption, AmbianceProcessor};
3use crate::media::dtmf::DtmfDetector;
4use crate::media::volume_control::HoldProcessor;
5use crate::media::{AudioFrame, INTERNAL_SAMPLERATE, Samples, TrackId};
6use crate::media::{
7    processor::Processor,
8    recorder::{Recorder, RecorderOption},
9    track::{Track, TrackPacketReceiver, TrackPacketSender},
10};
11use anyhow::Result;
12use std::collections::{HashMap, HashSet};
13use std::path::Path;
14use std::sync::{
15    Arc, Mutex as StdMutex,
16    atomic::{AtomicBool, AtomicU64, Ordering},
17};
18use std::time::Duration;
19use tokio::task::JoinHandle;
20use tokio::{
21    select,
22    sync::{Mutex, mpsc},
23};
24use tokio_util::sync::CancellationToken;
25use tracing::{debug, info, warn};
26use uuid;
27
28pub struct MediaStream {
29    id: String,
30    pub cancel_token: CancellationToken,
31    recorder_option: Mutex<Option<RecorderOption>>,
32    tracks: Mutex<HashMap<TrackId, (Box<dyn Track>, DtmfDetector)>>,
33    /// Trickle ICE candidates that arrived before any track existed to feed
34    /// them to. Drained into the next track started.
35    pending_ice_candidates: Mutex<Vec<(String, Option<String>, Option<u32>)>>,
36    suppressed_sources: Mutex<HashSet<TrackId>>,
37    event_sender: EventSender,
38    pub packet_sender: TrackPacketSender,
39    packet_receiver: Mutex<Option<TrackPacketReceiver>>,
40    recorder_sender: mpsc::UnboundedSender<AudioFrame>,
41    recorder_receiver: Mutex<Option<mpsc::UnboundedReceiver<AudioFrame>>>,
42    recorder_handle: Mutex<Option<JoinHandle<()>>>,
43    /// True once a recorder task is actually consuming `recorder_sender`.
44    /// The ambiance idle loop mirrors its frames to the recorder only when
45    /// this is set, avoiding unbounded buffering when recording starts late.
46    recording_active: Arc<AtomicBool>,
47    ambiance: Mutex<Option<Arc<StdMutex<AmbianceProcessor>>>>,
48    ambiance_source_id: StdMutex<Option<TrackId>>,
49    last_server_packet_ts: Arc<AtomicU64>,
50    ambiance_idle_started: AtomicBool,
51}
52
53const CALLEE_TRACK_ID: &str = "callee-track";
54const QUEUE_HOLD_TRACK_ID: &str = "queue-hold-track";
55pub const SERVER_SIDE_TRACK_ID: &str = "server-side-track";
56const AMBIANCE_IDLE_TRACK_ID: &str = "ambiance-track";
57const AMBIANCE_IDLE_PTIME: Duration = Duration::from_millis(20);
58// Skip idle fill if server-side audio arrived within this window (TTS ptime is 20ms).
59const AMBIANCE_IDLE_GAP_MS: u64 = 25;
60
61pub struct MediaStreamBuilder {
62    cancel_token: Option<CancellationToken>,
63    id: Option<String>,
64    event_sender: EventSender,
65    recorder_config: Option<RecorderOption>,
66}
67
68impl MediaStreamBuilder {
69    pub fn new(event_sender: EventSender) -> Self {
70        Self {
71            id: Some(format!("ms:{}", uuid::Uuid::new_v4())),
72            cancel_token: None,
73            event_sender,
74            recorder_config: None,
75        }
76    }
77    pub fn with_id(mut self, id: String) -> Self {
78        self.id = Some(id);
79        self
80    }
81
82    pub fn with_cancel_token(mut self, cancel_token: CancellationToken) -> Self {
83        self.cancel_token = Some(cancel_token);
84        self
85    }
86
87    pub fn with_recorder_config(mut self, recorder_config: RecorderOption) -> Self {
88        self.recorder_config = Some(recorder_config);
89        self
90    }
91
92    pub fn build(self) -> MediaStream {
93        let cancel_token = self
94            .cancel_token
95            .unwrap_or_else(|| CancellationToken::new());
96        let tracks = Mutex::new(HashMap::new());
97        let (track_packet_sender, track_packet_receiver) = mpsc::unbounded_channel();
98        let (recorder_sender, recorder_receiver) = mpsc::unbounded_channel();
99        MediaStream {
100            id: self.id.unwrap_or_default(),
101            cancel_token,
102            recorder_option: Mutex::new(self.recorder_config),
103            tracks,
104            pending_ice_candidates: Mutex::new(Vec::new()),
105            suppressed_sources: Mutex::new(HashSet::new()),
106            event_sender: self.event_sender,
107            packet_sender: track_packet_sender,
108            packet_receiver: Mutex::new(Some(track_packet_receiver)),
109            recorder_sender,
110            recorder_receiver: Mutex::new(Some(recorder_receiver)),
111            recorder_handle: Mutex::new(None),
112            recording_active: Arc::new(AtomicBool::new(false)),
113            ambiance: Mutex::new(None),
114            ambiance_source_id: StdMutex::new(None),
115            last_server_packet_ts: Arc::new(AtomicU64::new(0)),
116            ambiance_idle_started: AtomicBool::new(false),
117        }
118    }
119}
120
121impl MediaStream {
122    pub async fn serve(&self) -> Result<()> {
123        let packet_receiver = match self.packet_receiver.lock().await.take() {
124            Some(receiver) => receiver,
125            None => {
126                warn!(
127                    session_id = self.id,
128                    "MediaStream::serve() called multiple times, stream already serving"
129                );
130                return Ok(());
131            }
132        };
133        self.start_recorder().await.ok();
134        info!(session_id = self.id, "mediastream serving");
135        select! {
136            _ = self.cancel_token.cancelled() => {}
137            r = self.handle_forward_track(packet_receiver) => {
138                info!(session_id = self.id, "track packet receiver stopped {:?}", r);
139            }
140        }
141        Ok(())
142    }
143
144    /// Load ambiance once per call and keep mixing it while TTS/file playback is silent.
145    pub async fn ensure_ambiance(
146        &self,
147        option: AmbianceOption,
148        source_track_id: TrackId,
149    ) -> Result<Option<Arc<StdMutex<AmbianceProcessor>>>> {
150        let mut slot = self.ambiance.lock().await;
151        if let Some(existing) = slot.as_ref() {
152            return Ok(Some(existing.clone()));
153        }
154
155        if option.path.is_none() || option.enabled == Some(false) {
156            return Ok(None);
157        }
158
159        let processor = AmbianceProcessor::new(option).await?;
160        let shared = Arc::new(StdMutex::new(processor));
161        *slot = Some(shared.clone());
162        drop(slot);
163
164        *self.ambiance_source_id.lock().unwrap() = Some(source_track_id);
165        self.start_ambiance_idle_loop(shared.clone());
166        info!(session_id = self.id, "ambiance idle mixer started");
167        Ok(Some(shared))
168    }
169
170    fn start_ambiance_idle_loop(&self, processor: Arc<StdMutex<AmbianceProcessor>>) {
171        if self.ambiance_idle_started.swap(true, Ordering::SeqCst) {
172            return;
173        }
174
175        let cancel_token = self.cancel_token.clone();
176        let packet_sender = self.packet_sender.clone();
177        let last_server_packet_ts = self.last_server_packet_ts.clone();
178        let session_id = self.id.clone();
179        let recorder_sender = self.recorder_sender.clone();
180        let recording_active = self.recording_active.clone();
181
182        crate::spawn(async move {
183            let mut ticker = tokio::time::interval(AMBIANCE_IDLE_PTIME);
184            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
185            loop {
186                tokio::select! {
187                    _ = cancel_token.cancelled() => break,
188                    _ = ticker.tick() => {
189                        let now = crate::media::get_timestamp();
190                        let last = last_server_packet_ts.load(Ordering::Relaxed);
191                        if last != 0 && now.saturating_sub(last) < AMBIANCE_IDLE_GAP_MS {
192                            continue;
193                        }
194
195                        let mut frame = AudioFrame {
196                            track_id: AMBIANCE_IDLE_TRACK_ID.to_string(),
197                            samples: Samples::Empty,
198                            timestamp: now,
199                            sample_rate: INTERNAL_SAMPLERATE,
200                            channels: 1,
201                            ..Default::default()
202                        };
203                        {
204                            let mut ambiance = match processor.lock() {
205                                Ok(guard) => guard,
206                                Err(_) => break,
207                            };
208                            if let Err(e) = ambiance.process_frame(&mut frame) {
209                                warn!(session_id, "ambiance idle mix failed: {}", e);
210                                continue;
211                            }
212                        }
213                        if matches!(frame.samples, Samples::Empty) {
214                            continue;
215                        }
216                        // Mirror the mixed idle ambiance into the recording so the
217                        // stereo WAV's server-side channel reflects what the user
218                        // actually heard, including silent-period background audio.
219                        if recording_active.load(Ordering::SeqCst) {
220                            let mut recorded = frame.clone();
221                            recorded.track_id = SERVER_SIDE_TRACK_ID.to_string();
222                            let _ = recorder_sender.send(recorded);
223                        }
224                        if packet_sender.send(frame).is_err() {
225                            debug!(session_id, "ambiance idle sender closed");
226                            break;
227                        }
228                    }
229                }
230            }
231        });
232    }
233
234    pub fn stop(&self, _reason: Option<String>, _initiator: Option<String>) {
235        self.cancel_token.cancel()
236    }
237
238    pub async fn cleanup(&self) -> Result<()> {
239        self.cancel_token.cancel();
240        {
241            let mut tracks = self.tracks.lock().await;
242            for (id, (track, _)) in tracks.drain() {
243                if let Err(e) = track.stop().await {
244                    warn!(session_id = self.id, track_id = %id, "failed to stop track during cleanup: {}", e);
245                }
246            }
247        }
248        self.suppressed_sources.lock().await.clear();
249
250        if let Some(recorder_handle) = self.recorder_handle.lock().await.take() {
251            if let Ok(Ok(_)) = tokio::time::timeout(Duration::from_secs(30), recorder_handle).await
252            {
253                info!(session_id = self.id, "recorder stopped");
254            } else {
255                warn!(session_id = self.id, "recorder timeout");
256            }
257        }
258        Ok(())
259    }
260    pub async fn track_count(&self) -> usize {
261        self.tracks.lock().await.len()
262    }
263
264    pub async fn update_recorder_option(&self, recorder_config: RecorderOption) {
265        *self.recorder_option.lock().await = Some(recorder_config);
266        self.start_recorder().await.ok();
267    }
268
269    pub async fn remove_track(&self, id: &TrackId, graceful: bool) {
270        let track_entry = { self.tracks.lock().await.remove(id) };
271        if let Some((track, _)) = track_entry {
272            self.suppressed_sources.lock().await.remove(id);
273            let res = if !graceful {
274                track.stop().await
275            } else {
276                track.stop_graceful().await
277            };
278            match res {
279                Ok(_) => {}
280                Err(e) => {
281                    warn!(session_id = self.id, "failed to stop track: {}", e);
282                }
283            }
284        }
285    }
286    pub async fn update_remote_description(
287        &self,
288        track_id: &TrackId,
289        answer: &String,
290    ) -> Result<()> {
291        let track_entry = { self.tracks.lock().await.remove(track_id) };
292        if let Some((mut track, dtmf)) = track_entry {
293            let res = track.update_remote_description(answer).await;
294            self.tracks
295                .lock()
296                .await
297                .insert(track_id.clone(), (track, dtmf));
298            res?;
299        }
300        Ok(())
301    }
302
303    pub async fn update_remote_description_force(
304        &self,
305        track_id: &TrackId,
306        answer: &String,
307    ) -> Result<()> {
308        let track_entry = { self.tracks.lock().await.remove(track_id) };
309        if let Some((mut track, dtmf)) = track_entry {
310            let res = track.update_remote_description_force(answer).await;
311            self.tracks
312                .lock()
313                .await
314                .insert(track_id.clone(), (track, dtmf));
315            res?;
316        }
317        Ok(())
318    }
319
320    pub async fn handshake(
321        &self,
322        track_id: &TrackId,
323        offer: String,
324        timeout: Option<Duration>,
325    ) -> Result<String> {
326        let track_entry = { self.tracks.lock().await.remove(track_id) };
327        if let Some((mut track, dtmf)) = track_entry {
328            let res = track.handshake(offer, timeout).await;
329            self.tracks
330                .lock()
331                .await
332                .insert(track_id.clone(), (track, dtmf));
333            res
334        } else {
335            anyhow::bail!("track not found: {}", track_id)
336        }
337    }
338
339    pub async fn update_track(&self, mut track: Box<dyn Track>, play_id: Option<String>) {
340        self.remove_track(track.id(), false).await;
341        if self.recorder_option.lock().await.is_some() {
342            track.append_processor(Box::new(RecorderProcessor::new(
343                self.recorder_sender.clone(),
344            )));
345        }
346        match track
347            .start(self.event_sender.clone(), self.packet_sender.clone())
348            .await
349        {
350            Ok(_) => {
351                info!(session_id = self.id, track_id = track.id(), "track started");
352                let track_id = track.id().clone();
353                if track_id.as_str() == self.id.as_str() {
354                    let pending = std::mem::take(&mut *self.pending_ice_candidates.lock().await);
355                    for (candidate, sdp_mid, sdp_mline_index) in pending {
356                        if let Err(e) =
357                            track.add_ice_candidate(&candidate, sdp_mid.as_deref(), sdp_mline_index)
358                        {
359                            warn!(
360                                session_id = self.id,
361                                track_id = track.id(),
362                                "failed to apply buffered ICE candidate: {}",
363                                e
364                            );
365                        }
366                    }
367                }
368                self.tracks
369                    .lock()
370                    .await
371                    .insert(track_id.clone(), (track, DtmfDetector::new()));
372                self.event_sender
373                    .send(SessionEvent::TrackStart {
374                        track_id,
375                        timestamp: crate::media::get_timestamp(),
376                        play_id,
377                    })
378                    .ok();
379            }
380            Err(e) => {
381                warn!(
382                    session_id = self.id,
383                    track_id = track.id(),
384                    play_id = play_id.as_deref(),
385                    "Failed to start track: {}",
386                    e
387                );
388            }
389        }
390    }
391
392    pub async fn mute_track(&self, id: Option<TrackId>) {
393        if let Some(id) = id {
394            if let Some((track, _)) = self.tracks.lock().await.get_mut(&id) {
395                MuteProcessor::mute_track(track.as_mut());
396            }
397        } else {
398            for (track, _) in self.tracks.lock().await.values_mut() {
399                MuteProcessor::mute_track(track.as_mut());
400            }
401        }
402    }
403
404    pub async fn unmute_track(&self, id: Option<TrackId>) {
405        if let Some(id) = id {
406            if let Some((track, _)) = self.tracks.lock().await.get_mut(&id) {
407                MuteProcessor::unmute_track(track.as_mut());
408            }
409        } else {
410            for (track, _) in self.tracks.lock().await.values_mut() {
411                MuteProcessor::unmute_track(track.as_mut());
412            }
413        }
414    }
415
416    /// Trickle ICE: feed a remote candidate into the ICE-backed (WebRTC)
417    /// track, if it's up yet, or buffer it for `update_track` to replay
418    /// otherwise.
419    pub async fn add_ice_candidate(
420        &self,
421        candidate: &str,
422        sdp_mid: Option<&str>,
423        sdp_mline_index: Option<u32>,
424    ) -> Result<()> {
425        let tracks = self.tracks.lock().await;
426        if let Some((track, _)) = tracks.get(self.id.as_str()) {
427            track.add_ice_candidate(candidate, sdp_mid, sdp_mline_index)?;
428            return Ok(());
429        }
430        drop(tracks);
431        self.pending_ice_candidates.lock().await.push((
432            candidate.to_string(),
433            sdp_mid.map(|s| s.to_string()),
434            sdp_mline_index,
435        ));
436        Ok(())
437    }
438
439    pub async fn pause_playback(&self, id: TrackId) -> Result<()> {
440        self.set_playback_paused(id, true).await
441    }
442
443    pub async fn resume_playback(&self, id: TrackId) -> Result<()> {
444        self.set_playback_paused(id, false).await
445    }
446
447    async fn set_playback_paused(&self, id: TrackId, paused: bool) -> Result<()> {
448        if let Some((track, _)) = self.tracks.lock().await.get_mut(&id) {
449            if track.set_paused(paused) {
450                Ok(())
451            } else {
452                warn!(
453                    session_id = self.id,
454                    track_id = %id,
455                    paused,
456                    "pause state requested for track that does not support pausing"
457                );
458                Err(anyhow::anyhow!("track does not support pausing: {}", id))
459            }
460        } else {
461            warn!(
462                session_id = self.id,
463                track_id = %id,
464                paused,
465                "pause state requested for unknown track"
466            );
467            Err(anyhow::anyhow!("track not found: {}", id))
468        }
469    }
470
471    pub async fn hold_track(&self, id: Option<TrackId>) {
472        if let Some(id) = id {
473            if let Some((track, _)) = self.tracks.lock().await.get_mut(&id) {
474                HoldTrack::hold_track(track.as_mut());
475            }
476        } else {
477            for (track, _) in self.tracks.lock().await.values_mut() {
478                HoldTrack::hold_track(track.as_mut());
479            }
480        }
481    }
482
483    pub async fn resume_track(&self, id: Option<TrackId>) {
484        if let Some(id) = id {
485            if let Some((track, _)) = self.tracks.lock().await.get_mut(&id) {
486                HoldTrack::resume_track(track.as_mut());
487            }
488        } else {
489            for (track, _) in self.tracks.lock().await.values_mut() {
490                HoldTrack::resume_track(track.as_mut());
491            }
492        }
493    }
494
495    pub async fn suppress_forwarding(&self, track_id: &TrackId) {
496        self.suppressed_sources
497            .lock()
498            .await
499            .insert(track_id.clone());
500    }
501
502    pub async fn resume_forwarding(&self, track_id: &TrackId) {
503        self.suppressed_sources.lock().await.remove(track_id);
504    }
505
506    pub async fn remove_processor<T: 'static>(&self, track_id: &TrackId) -> Result<()> {
507        if let Some((track, _)) = self.tracks.lock().await.get_mut(track_id) {
508            track.as_mut().processor_chain().remove_processor::<T>();
509            Ok(())
510        } else {
511            Err(anyhow::anyhow!("Track {} not found", track_id))
512        }
513    }
514
515    pub async fn append_processor(
516        &self,
517        track_id: &TrackId,
518        processor: Box<dyn crate::media::processor::Processor>,
519    ) -> Result<()> {
520        if let Some((track, _)) = self.tracks.lock().await.get_mut(track_id) {
521            track.as_mut().processor_chain().append_processor(processor);
522            Ok(())
523        } else {
524            Err(anyhow::anyhow!("Track {} not found", track_id))
525        }
526    }
527}
528
529#[derive(Clone)]
530pub struct RecorderProcessor {
531    sender: mpsc::UnboundedSender<AudioFrame>,
532}
533
534impl RecorderProcessor {
535    pub fn new(sender: mpsc::UnboundedSender<AudioFrame>) -> Self {
536        Self { sender }
537    }
538}
539
540impl Processor for RecorderProcessor {
541    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
542        let frame_clone = frame.clone();
543        let _ = self.sender.send(frame_clone);
544        Ok(())
545    }
546}
547
548impl MediaStream {
549    pub async fn start_recorder(&self) -> Result<()> {
550        let recorder_option = self.recorder_option.lock().await.clone();
551        if let Some(recorder_option) = recorder_option {
552            if recorder_option.recorder_file.is_empty() {
553                warn!(
554                    session_id = self.id,
555                    "recorder file is empty, skipping recorder start"
556                );
557                return Ok(());
558            }
559            let recorder_receiver = match self.recorder_receiver.lock().await.take() {
560                Some(receiver) => receiver,
561                None => {
562                    return Ok(());
563                }
564            };
565            let cancel_token = self.cancel_token.child_token();
566            let session_id_clone = self.id.clone();
567
568            info!(
569                session_id = session_id_clone,
570                sample_rate = recorder_option.samplerate,
571                ptime = recorder_option.ptime,
572                "start recorder",
573            );
574
575            let recorder_handle = crate::spawn(async move {
576                let recorder_file = recorder_option.recorder_file.clone();
577                let recorder =
578                    Recorder::new(cancel_token, session_id_clone.clone(), recorder_option);
579                match recorder
580                    .process_recording(Path::new(&recorder_file), recorder_receiver)
581                    .await
582                {
583                    Ok(_) => {}
584                    Err(e) => {
585                        warn!(
586                            session_id = session_id_clone,
587                            "Failed to process recorder: {}", e
588                        );
589                    }
590                }
591            });
592            *self.recorder_handle.lock().await = Some(recorder_handle);
593            self.recording_active.store(true, Ordering::SeqCst);
594
595            // Inject RecorderProcessor into tracks that were added before the recorder started
596            for (track, _) in self.tracks.lock().await.values_mut() {
597                track.insert_processor(Box::new(RecorderProcessor::new(
598                    self.recorder_sender.clone(),
599                )));
600            }
601        }
602        Ok(())
603    }
604
605    pub async fn set_track_refer(&self, track_id: &TrackId, refer: Option<bool>) {
606        if let Some((_, dtmf)) = self.tracks.lock().await.get_mut(track_id) {
607            dtmf.refer = refer;
608        }
609    }
610
611    pub async fn set_track_dtmf_forward(&self, track_id: &TrackId, forward: bool) {
612        if let Some((_, dtmf)) = self.tracks.lock().await.get_mut(track_id) {
613            dtmf.suppress_dtmf_forward = !forward;
614        }
615    }
616
617    async fn handle_forward_track(&self, mut packet_receiver: TrackPacketReceiver) {
618        let event_sender = self.event_sender.clone();
619        while let Some(packet) = packet_receiver.recv().await {
620            if self
621                .ambiance_source_id
622                .lock()
623                .ok()
624                .and_then(|id| id.clone())
625                .as_ref()
626                == Some(&packet.track_id)
627            {
628                self.last_server_packet_ts
629                    .store(crate::media::get_timestamp(), Ordering::Relaxed);
630            }
631
632            let suppressed = {
633                self.suppressed_sources
634                    .lock()
635                    .await
636                    .contains(&packet.track_id)
637            };
638
639            let is_dtmf = matches!(&packet.samples,
640                Samples::RTP { payload_type, .. } if *payload_type >= 96 && *payload_type <= 127);
641
642            let mut tracks = self.tracks.lock().await;
643
644            // Check once whether the source track suppresses DTMF forwarding.
645            let source_suppresses_dtmf = is_dtmf
646                && tracks
647                    .get(&packet.track_id)
648                    .map(|(_, d)| d.suppress_dtmf_forward)
649                    .unwrap_or(false);
650
651            for (track, dtmf_detector) in tracks.values_mut() {
652                if track.id() == &packet.track_id {
653                    if let Samples::RTP {
654                        payload_type,
655                        payload,
656                        ..
657                    } = &packet.samples
658                    {
659                        if let Some(digit) = dtmf_detector.detect_rtp(*payload_type, payload) {
660                            debug!(track_id = track.id(), digit, "DTMF detected");
661                            event_sender
662                                .send(SessionEvent::Dtmf {
663                                    track_id: packet.track_id.to_string(),
664                                    timestamp: packet.timestamp,
665                                    digit,
666                                    refer: dtmf_detector.refer,
667                                })
668                                .ok();
669                        }
670                    }
671                    continue;
672                }
673                if suppressed {
674                    continue;
675                }
676                // Skip DTMF forwarding if source or destination has it suppressed.
677                if source_suppresses_dtmf || (is_dtmf && dtmf_detector.suppress_dtmf_forward) {
678                    continue;
679                }
680                if packet.track_id == QUEUE_HOLD_TRACK_ID && track.id() == CALLEE_TRACK_ID {
681                    continue;
682                }
683                if let Err(e) = track.send_packet(&packet).await {
684                    warn!(
685                        id = track.id(),
686                        "media_stream: Failed to send packet to track: {}", e
687                    );
688                }
689            }
690        }
691    }
692}
693
694pub struct MuteProcessor;
695
696impl MuteProcessor {
697    pub fn mute_track(track: &mut dyn Track) {
698        let chain = track.processor_chain();
699        if !chain.has_processor::<MuteProcessor>() {
700            chain.insert_processor(Box::new(MuteProcessor));
701        }
702    }
703
704    pub fn unmute_track(track: &mut dyn Track) {
705        let chain = track.processor_chain();
706        chain.remove_processor::<MuteProcessor>();
707    }
708}
709
710impl Processor for MuteProcessor {
711    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
712        match &mut frame.samples {
713            Samples::PCM { samples } => {
714                samples.fill(0);
715            }
716            // discard DTMF frames
717            Samples::RTP { payload_type, .. } if *payload_type >= 96 && *payload_type <= 127 => {
718                frame.samples = Samples::Empty;
719            }
720            _ => {}
721        }
722        Ok(())
723    }
724}
725
726pub struct HoldTrack;
727
728impl HoldTrack {
729    pub fn hold_track(track: &mut dyn Track) {
730        let chain = track.processor_chain();
731        // Remove existing processor if present
732        chain.remove_processor::<HoldProcessor>();
733        // Add a new processor with hold state set to true
734        let processor = HoldProcessor::new();
735        processor.set_hold(true);
736        chain.insert_processor(Box::new(processor));
737    }
738
739    pub fn resume_track(track: &mut dyn Track) {
740        let chain = track.processor_chain();
741        // Simply remove the hold processor to resume normal operation
742        chain.remove_processor::<HoldProcessor>();
743    }
744}
active_call/media/stream.rs

active_call/media/
stream.rs
