active_call/media/
stream.rs

1use crate::event::{EventSender, SessionEvent};
2use crate::media::dtmf::DtmfDetector;
3use crate::media::{AudioFrame, Samples, TrackId};
4use crate::media::{
5    processor::Processor,
6    recorder::{Recorder, RecorderOption},
7    track::{Track, TrackPacketReceiver, TrackPacketSender},
8};
9use anyhow::Result;
10use std::collections::{HashMap, HashSet};
11use std::path::Path;
12use std::time::Duration;
13use tokio::task::JoinHandle;
14use tokio::{
15    select,
16    sync::{Mutex, mpsc},
17};
18use tokio_util::sync::CancellationToken;
19use tracing::{debug, info, warn};
20use uuid;
21
22pub struct MediaStream {
23    id: String,
24    pub cancel_token: CancellationToken,
25    recorder_option: Mutex<Option<RecorderOption>>,
26    tracks: Mutex<HashMap<TrackId, (Box<dyn Track>, DtmfDetector)>>,
27    suppressed_sources: Mutex<HashSet<TrackId>>,
28    event_sender: EventSender,
29    pub packet_sender: TrackPacketSender,
30    packet_receiver: Mutex<Option<TrackPacketReceiver>>,
31    recorder_sender: mpsc::UnboundedSender<AudioFrame>,
32    recorder_receiver: Mutex<Option<mpsc::UnboundedReceiver<AudioFrame>>>,
33    recorder_handle: Mutex<Option<JoinHandle<()>>>,
34}
35
36const CALLEE_TRACK_ID: &str = "callee-track";
37const QUEUE_HOLD_TRACK_ID: &str = "queue-hold-track";
38
39pub struct MediaStreamBuilder {
40    cancel_token: Option<CancellationToken>,
41    id: Option<String>,
42    event_sender: EventSender,
43    recorder_config: Option<RecorderOption>,
44}
45
46impl MediaStreamBuilder {
47    pub fn new(event_sender: EventSender) -> Self {
48        Self {
49            id: Some(format!("ms:{}", uuid::Uuid::new_v4())),
50            cancel_token: None,
51            event_sender,
52            recorder_config: None,
53        }
54    }
55    pub fn with_id(mut self, id: String) -> Self {
56        self.id = Some(id);
57        self
58    }
59
60    pub fn with_cancel_token(mut self, cancel_token: CancellationToken) -> Self {
61        self.cancel_token = Some(cancel_token);
62        self
63    }
64
65    pub fn with_recorder_config(mut self, recorder_config: RecorderOption) -> Self {
66        self.recorder_config = Some(recorder_config);
67        self
68    }
69
70    pub fn build(self) -> MediaStream {
71        let cancel_token = self
72            .cancel_token
73            .unwrap_or_else(|| CancellationToken::new());
74        let tracks = Mutex::new(HashMap::new());
75        let (track_packet_sender, track_packet_receiver) = mpsc::unbounded_channel();
76        let (recorder_sender, recorder_receiver) = mpsc::unbounded_channel();
77        MediaStream {
78            id: self.id.unwrap_or_default(),
79            cancel_token,
80            recorder_option: Mutex::new(self.recorder_config),
81            tracks,
82            suppressed_sources: Mutex::new(HashSet::new()),
83            event_sender: self.event_sender,
84            packet_sender: track_packet_sender,
85            packet_receiver: Mutex::new(Some(track_packet_receiver)),
86            recorder_sender,
87            recorder_receiver: Mutex::new(Some(recorder_receiver)),
88            recorder_handle: Mutex::new(None),
89        }
90    }
91}
92
93impl MediaStream {
94    pub async fn serve(&self) -> Result<()> {
95        let packet_receiver = match self.packet_receiver.lock().await.take() {
96            Some(receiver) => receiver,
97            None => {
98                warn!(
99                    session_id = self.id,
100                    "MediaStream::serve() called multiple times, stream already serving"
101                );
102                return Ok(());
103            }
104        };
105        self.start_recorder().await.ok();
106        info!(session_id = self.id, "mediastream serving");
107        select! {
108            _ = self.cancel_token.cancelled() => {}
109            r = self.handle_forward_track(packet_receiver) => {
110                info!(session_id = self.id, "track packet receiver stopped {:?}", r);
111            }
112        }
113        Ok(())
114    }
115
116    pub fn stop(&self, _reason: Option<String>, _initiator: Option<String>) {
117        self.cancel_token.cancel()
118    }
119
120    pub async fn cleanup(&self) -> Result<()> {
121        self.cancel_token.cancel();
122        if let Some(recorder_handle) = self.recorder_handle.lock().await.take() {
123            if let Ok(Ok(_)) = tokio::time::timeout(Duration::from_secs(30), recorder_handle).await
124            {
125                info!(session_id = self.id, "recorder stopped");
126            } else {
127                warn!(session_id = self.id, "recorder timeout");
128            }
129        }
130        Ok(())
131    }
132
133    pub async fn update_recorder_option(&self, recorder_config: RecorderOption) {
134        *self.recorder_option.lock().await = Some(recorder_config);
135        self.start_recorder().await.ok();
136    }
137
138    pub async fn remove_track(&self, id: &TrackId, graceful: bool) {
139        if let Some((track, _)) = self.tracks.lock().await.remove(id) {
140            self.suppressed_sources.lock().await.remove(id);
141            let res = if !graceful {
142                track.stop().await
143            } else {
144                track.stop_graceful().await
145            };
146            match res {
147                Ok(_) => {}
148                Err(e) => {
149                    warn!(session_id = self.id, "failed to stop track: {}", e);
150                }
151            }
152        }
153    }
154    pub async fn update_remote_description(
155        &self,
156        track_id: &TrackId,
157        answer: &String,
158    ) -> Result<()> {
159        if let Some((track, _)) = self.tracks.lock().await.get_mut(track_id) {
160            track.update_remote_description(answer).await?;
161        }
162        Ok(())
163    }
164    pub async fn update_track(&self, mut track: Box<dyn Track>, play_id: Option<String>) {
165        self.remove_track(track.id(), false).await;
166        if self.recorder_option.lock().await.is_some() {
167            track.insert_processor(Box::new(RecorderProcessor::new(
168                self.recorder_sender.clone(),
169            )));
170        }
171        match track
172            .start(self.event_sender.clone(), self.packet_sender.clone())
173            .await
174        {
175            Ok(_) => {
176                info!(session_id = self.id, track_id = track.id(), "track started");
177                let track_id = track.id().clone();
178                self.tracks
179                    .lock()
180                    .await
181                    .insert(track_id.clone(), (track, DtmfDetector::new()));
182                self.event_sender
183                    .send(SessionEvent::TrackStart {
184                        track_id,
185                        timestamp: crate::media::get_timestamp(),
186                        play_id,
187                    })
188                    .ok();
189            }
190            Err(e) => {
191                warn!(
192                    session_id = self.id,
193                    track_id = track.id(),
194                    play_id = play_id.as_deref(),
195                    "Failed to start track: {}",
196                    e
197                );
198            }
199        }
200    }
201
202    pub async fn mute_track(&self, id: Option<TrackId>) {
203        if let Some(id) = id {
204            if let Some((track, _)) = self.tracks.lock().await.get_mut(&id) {
205                MuteProcessor::mute_track(track.as_mut());
206            }
207        } else {
208            for (track, _) in self.tracks.lock().await.values_mut() {
209                MuteProcessor::mute_track(track.as_mut());
210            }
211        }
212    }
213
214    pub async fn unmute_track(&self, id: Option<TrackId>) {
215        if let Some(id) = id {
216            if let Some((track, _)) = self.tracks.lock().await.get_mut(&id) {
217                MuteProcessor::unmute_track(track.as_mut());
218            }
219        } else {
220            for (track, _) in self.tracks.lock().await.values_mut() {
221                MuteProcessor::unmute_track(track.as_mut());
222            }
223        }
224    }
225
226    pub async fn suppress_forwarding(&self, track_id: &TrackId) {
227        self.suppressed_sources
228            .lock()
229            .await
230            .insert(track_id.clone());
231    }
232
233    pub async fn resume_forwarding(&self, track_id: &TrackId) {
234        self.suppressed_sources.lock().await.remove(track_id);
235    }
236}
237
238#[derive(Clone)]
239pub struct RecorderProcessor {
240    sender: mpsc::UnboundedSender<AudioFrame>,
241}
242
243impl RecorderProcessor {
244    pub fn new(sender: mpsc::UnboundedSender<AudioFrame>) -> Self {
245        Self { sender }
246    }
247}
248
249impl Processor for RecorderProcessor {
250    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
251        let frame_clone = frame.clone();
252        let _ = self.sender.send(frame_clone);
253        Ok(())
254    }
255}
256
257impl MediaStream {
258    pub async fn start_recorder(&self) -> Result<()> {
259        let recorder_option = self.recorder_option.lock().await.clone();
260        if let Some(recorder_option) = recorder_option {
261            if recorder_option.recorder_file.is_empty() {
262                warn!(
263                    session_id = self.id,
264                    "recorder file is empty, skipping recorder start"
265                );
266                return Ok(());
267            }
268            let recorder_receiver = match self.recorder_receiver.lock().await.take() {
269                Some(receiver) => receiver,
270                None => {
271                    return Ok(());
272                }
273            };
274            let cancel_token = self.cancel_token.child_token();
275            let session_id_clone = self.id.clone();
276
277            info!(
278                session_id = session_id_clone,
279                sample_rate = recorder_option.samplerate,
280                ptime = recorder_option.ptime,
281                "start recorder",
282            );
283
284            let recorder_handle = crate::spawn(async move {
285                let recorder_file = recorder_option.recorder_file.clone();
286                let recorder =
287                    Recorder::new(cancel_token, session_id_clone.clone(), recorder_option);
288                match recorder
289                    .process_recording(Path::new(&recorder_file), recorder_receiver)
290                    .await
291                {
292                    Ok(_) => {}
293                    Err(e) => {
294                        warn!(
295                            session_id = session_id_clone,
296                            "Failed to process recorder: {}", e
297                        );
298                    }
299                }
300            });
301            *self.recorder_handle.lock().await = Some(recorder_handle);
302        }
303        Ok(())
304    }
305
306    async fn handle_forward_track(&self, mut packet_receiver: TrackPacketReceiver) {
307        let event_sender = self.event_sender.clone();
308        while let Some(packet) = packet_receiver.recv().await {
309            let suppressed = {
310                self.suppressed_sources
311                    .lock()
312                    .await
313                    .contains(&packet.track_id)
314            };
315            // Process the packet with each track
316            for (track, dtmf_detector) in self.tracks.lock().await.values_mut() {
317                if track.id() == &packet.track_id {
318                    match &packet.samples {
319                        Samples::RTP {
320                            payload_type,
321                            payload,
322                            ..
323                        } => {
324                            if let Some(digit) = dtmf_detector.detect_rtp(*payload_type, payload) {
325                                debug!(track_id = track.id(), digit, "DTMF detected");
326                                event_sender
327                                    .send(SessionEvent::Dtmf {
328                                        track_id: packet.track_id.to_string(),
329                                        timestamp: packet.timestamp,
330                                        digit,
331                                    })
332                                    .ok();
333                            }
334                        }
335                        _ => {}
336                    }
337                    continue;
338                }
339                if suppressed {
340                    continue;
341                }
342                if packet.track_id == QUEUE_HOLD_TRACK_ID && track.id() == CALLEE_TRACK_ID {
343                    continue;
344                }
345                if let Err(e) = track.send_packet(&packet).await {
346                    warn!(
347                        id = track.id(),
348                        "media_stream: Failed to send packet to track: {}", e
349                    );
350                }
351            }
352        }
353    }
354}
355
356pub struct MuteProcessor;
357
358impl MuteProcessor {
359    pub fn mute_track(track: &mut dyn Track) {
360        let chain = track.processor_chain();
361        if !chain.has_processor::<MuteProcessor>() {
362            chain.insert_processor(Box::new(MuteProcessor));
363        }
364    }
365
366    pub fn unmute_track(track: &mut dyn Track) {
367        let chain = track.processor_chain();
368        chain.remove_processor::<MuteProcessor>();
369    }
370}
371
372impl Processor for MuteProcessor {
373    fn process_frame(&mut self, frame: &mut AudioFrame) -> Result<()> {
374        match &mut frame.samples {
375            Samples::PCM { samples } => {
376                samples.fill(0);
377            }
378            // discard DTMF frames
379            Samples::RTP { payload_type, .. } if *payload_type >= 96 && *payload_type <= 127 => {
380                frame.samples = Samples::Empty;
381            }
382            _ => {}
383        }
384        Ok(())
385    }
386}
active_call/media/stream.rs

active_call/media/
stream.rs
