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

active_call/media/
stream.rs
