active_call/call/
state.rs

1use crate::CallOption;
2use crate::app::AppState;
3use crate::call::active_call::ActiveCallType;
4use crate::callrecord::{CallRecord, CallRecordHangupReason};
5use crate::event::SessionEvent;
6use crate::media::TrackId;
7use arc_swap::ArcSwap;
8use chrono::{DateTime, Utc};
9use rsipstack::dialog::dialog::TerminatedReason;
10use serde_json::Value;
11use std::collections::HashMap;
12use std::sync::Arc;
13use tokio::sync::mpsc;
14
15pub type Extras = Arc<ArcSwap<HashMap<String, Value>>>;
16
17#[derive(Clone, Debug, Default)]
18pub struct CallProgress {
19    pub session_id: String,
20    pub start_time: Option<DateTime<Utc>>,
21    pub ring_time: Option<DateTime<Utc>>,
22    pub answer_time: Option<DateTime<Utc>>,
23    pub answer: Option<String>,
24    pub last_status_code: u16,
25    pub hangup_reason: Option<CallRecordHangupReason>,
26    pub option: Option<CallOption>,
27}
28
29impl CallProgress {
30    pub fn on_early(&mut self, code: u16) {
31        self.ring_time.get_or_insert_with(Utc::now);
32        self.last_status_code = code;
33    }
34
35    pub fn on_confirmed(&mut self, session_id: String) {
36        self.session_id = session_id;
37        self.answer_time.get_or_insert_with(Utc::now);
38        self.last_status_code = 200;
39    }
40
41    pub fn on_answered(&mut self) {
42        self.answer_time.get_or_insert_with(Utc::now);
43        self.last_status_code = 200;
44    }
45
46    pub fn try_set_answer(&mut self, sdp: &str) {
47        if self.answer.is_none() {
48            self.answer = Some(sdp.to_string());
49        }
50    }
51
52    pub fn set_hangup_reason(&mut self, reason: CallRecordHangupReason) {
53        if self.hangup_reason.is_none() {
54            self.hangup_reason = Some(reason);
55        }
56    }
57
58    pub fn merge_option(&self, mut option: CallOption) -> CallOption {
59        if let Some(existing) = &self.option {
60            if option.asr.is_none() {
61                option.asr = existing.asr.clone();
62            }
63            if option.tts.is_none() {
64                option.tts = existing.tts.clone();
65            }
66            if option.vad.is_none() {
67                option.vad = existing.vad.clone();
68            }
69            if option.denoise.is_none() {
70                option.denoise = existing.denoise;
71            }
72            if option.agc.is_none() {
73                option.agc = existing.agc.clone();
74            }
75            if option.recorder.is_none() {
76                option.recorder = existing.recorder.clone();
77            }
78            if option.eou.is_none() {
79                option.eou = existing.eou.clone();
80            }
81            if option.extra.is_none() {
82                option.extra = existing.extra.clone();
83            }
84            if option.ambiance.is_none() {
85                option.ambiance = existing.ambiance.clone();
86            }
87            if option.ringback_detection.is_none() {
88                option.ringback_detection = existing.ringback_detection.clone();
89            }
90        }
91        option
92    }
93
94    pub fn termination(reason: Option<&TerminatedReason>) -> TerminationInfo {
95        match reason {
96            Some(TerminatedReason::UacCancel) => {
97                TerminationInfo::new(487, CallRecordHangupReason::Canceled, "caller")
98            }
99            Some(TerminatedReason::UacBye) => {
100                TerminationInfo::new(200, CallRecordHangupReason::ByCaller, "caller")
101            }
102            Some(TerminatedReason::UacBusy) => {
103                TerminationInfo::new(486, CallRecordHangupReason::ByCaller, "caller")
104            }
105            Some(TerminatedReason::UasBye) => {
106                TerminationInfo::new(200, CallRecordHangupReason::ByCallee, "callee")
107            }
108            Some(TerminatedReason::UasBusy) => {
109                TerminationInfo::new(486, CallRecordHangupReason::ByCallee, "callee")
110            }
111            Some(TerminatedReason::UasDecline) => {
112                TerminationInfo::new(603, CallRecordHangupReason::ByCallee, "callee")
113            }
114            Some(TerminatedReason::UacOther(code)) => {
115                TerminationInfo::new(code.code(), CallRecordHangupReason::ByCaller, "system")
116            }
117            Some(TerminatedReason::UasOther(code)) => {
118                TerminationInfo::new(code.code(), CallRecordHangupReason::ByCallee, "system")
119            }
120            _ => TerminationInfo::new(500, CallRecordHangupReason::BySystem, "system"),
121        }
122    }
123}
124
125pub struct TerminationInfo {
126    pub status_code: u16,
127    pub hangup_reason: CallRecordHangupReason,
128    pub initiator: &'static str,
129}
130
131impl TerminationInfo {
132    fn new(
133        status_code: u16,
134        hangup_reason: CallRecordHangupReason,
135        initiator: &'static str,
136    ) -> Self {
137        Self {
138            status_code,
139            hangup_reason,
140            initiator,
141        }
142    }
143}
144
145#[derive(Clone)]
146pub struct LegShared {
147    pub ssrc: u32,
148    pub is_refer: bool,
149    pub progress: Arc<ArcSwap<CallProgress>>,
150    pub extras: Extras,
151}
152
153impl LegShared {
154    pub fn new(ssrc: u32, is_refer: bool, progress: CallProgress) -> Self {
155        Self {
156            ssrc,
157            is_refer,
158            progress: Arc::new(ArcSwap::from_pointee(progress)),
159            extras: Arc::new(ArcSwap::from_pointee(HashMap::new())),
160        }
161    }
162
163    pub fn update_progress(&self, f: impl Fn(&mut CallProgress)) {
164        self.progress.rcu(|p| {
165            let mut p = CallProgress::clone(p);
166            f(&mut p);
167            p
168        });
169    }
170
171    pub fn set_extra(&self, key: &str, value: Value) {
172        self.extras.rcu(|e| {
173            let mut e = HashMap::clone(e);
174            e.insert(key.to_string(), value.clone());
175            e
176        });
177    }
178
179    pub fn build_hangup_event(&self, track_id: TrackId, initiator: Option<String>) -> SessionEvent {
180        let progress = self.progress.load_full();
181        let extras = self.extras.load_full();
182        build_hangup_event(&progress, &extras, self.is_refer, track_id, initiator)
183    }
184
185    pub fn build_callrecord(&self, app_state: &AppState, call_type: ActiveCallType) -> CallRecord {
186        let session_id = self.progress.load_full().session_id.clone();
187        self.build_callrecord_with_id(app_state, call_type, session_id)
188    }
189
190    pub fn build_callrecord_with_id(
191        &self,
192        app_state: &AppState,
193        call_type: ActiveCallType,
194        session_id: String,
195    ) -> CallRecord {
196        let progress = self.progress.load_full();
197        let extras = self.extras.load_full();
198        build_callrecord(&progress, &extras, None, app_state, session_id, call_type)
199    }
200}
201
202pub fn build_hangup_event(
203    progress: &CallProgress,
204    extras: &HashMap<String, Value>,
205    is_refer: bool,
206    track_id: TrackId,
207    initiator: Option<String>,
208) -> SessionEvent {
209    let from = progress.option.as_ref().and_then(|o| o.caller.as_ref());
210    let to = progress.option.as_ref().and_then(|o| o.callee.as_ref());
211
212    SessionEvent::Hangup {
213        track_id,
214        timestamp: crate::media::get_timestamp(),
215        reason: progress.hangup_reason.as_ref().map(|r| format!("{:?}", r)),
216        initiator,
217        start_time: progress.start_time.unwrap_or_default().to_rfc3339(),
218        answer_time: progress.answer_time.map(|t| t.to_rfc3339()),
219        ringing_time: progress.ring_time.map(|t| t.to_rfc3339()),
220        hangup_time: Utc::now().to_rfc3339(),
221        extra: Some(extras.clone()),
222        from: from.map(|f| f.into()),
223        to: to.map(|f| f.into()),
224        refer: Some(is_refer),
225    }
226}
227
228pub fn build_callrecord(
229    progress: &CallProgress,
230    extras: &HashMap<String, Value>,
231    refer_leg: Option<&LegShared>,
232    app_state: &AppState,
233    session_id: String,
234    call_type: ActiveCallType,
235) -> CallRecord {
236    let option = progress.option.clone().unwrap_or_default();
237    let recorder = if option.recorder.is_some() {
238        let recorder_file = app_state.get_recorder_file(&session_id);
239        if std::path::Path::new(&recorder_file).exists() {
240            let file_size = std::fs::metadata(&recorder_file)
241                .map(|m| m.len())
242                .unwrap_or(0);
243            vec![crate::callrecord::CallRecordMedia {
244                track_id: session_id.clone(),
245                path: recorder_file,
246                size: file_size,
247                extra: None,
248            }]
249        } else {
250            vec![]
251        }
252    } else {
253        vec![]
254    };
255
256    let dump_event_file = app_state.get_dump_events_file(&session_id);
257    let dump_event_file = if std::path::Path::new(&dump_event_file).exists() {
258        Some(dump_event_file)
259    } else {
260        None
261    };
262
263    let refer_callrecord =
264        refer_leg.map(|leg| Box::new(leg.build_callrecord(app_state, ActiveCallType::B2bua)));
265
266    let caller = option.caller.clone().unwrap_or_default();
267    let callee = option.callee.clone().unwrap_or_default();
268
269    CallRecord {
270        option: Some(option),
271        call_id: session_id,
272        call_type,
273        start_time: progress.start_time.unwrap_or_default(),
274        ring_time: progress.ring_time,
275        answer_time: progress.answer_time,
276        end_time: Utc::now(),
277        caller,
278        callee,
279        hangup_reason: progress.hangup_reason.clone(),
280        hangup_messages: Vec::new(),
281        status_code: progress.last_status_code,
282        extras: Some(extras.clone()),
283        dump_event_file,
284        recorder,
285        refer_callrecord,
286    }
287}
288
289pub fn resolve_final_answer(
290    raw: Option<Vec<u8>>,
291    early: Option<&String>,
292) -> Result<(String, bool), &'static str> {
293    match raw {
294        Some(bytes) => {
295            let s = String::from_utf8_lossy(&bytes).to_string();
296            if s.trim().is_empty() {
297                match early {
298                    Some(e) if !e.is_empty() => Ok((e.clone(), true)),
299                    _ => Ok((s, false)),
300                }
301            } else {
302                Ok((s, false))
303            }
304        }
305        None => match early {
306            Some(e) if !e.is_empty() => Ok((e.clone(), true)),
307            _ => Err("no answer received"),
308        },
309    }
310}
311
312pub enum ActorMsg {
313    ReferDone {
314        track_id: TrackId,
315        forward_dtmf: bool,
316        result: Result<String, rsipstack::Error>,
317    },
318}
319
320pub struct CallRuntime {
321    pub input_timeout_expire: (u64, u32),
322    pub actor_tx: mpsc::Sender<ActorMsg>,
323    pub me: Option<crate::call::active_call::ActiveCallRef>,
324}
325
326impl CallRuntime {
327    pub fn new(actor_tx: mpsc::Sender<ActorMsg>) -> Self {
328        Self {
329            input_timeout_expire: (0, 0),
330            actor_tx,
331            me: None,
332        }
333    }
334}
335
336#[cfg(test)]
337mod tests {
338    use super::*;
339
340    const EARLY_SDP: &str = "v=0\r\no=- 1000 1 IN IP4 192.168.1.100\r\n";
341
342    #[test]
343    fn early_then_confirmed_progress() {
344        let mut p = CallProgress::default();
345        p.on_early(183);
346        assert!(p.ring_time.is_some());
347        assert_eq!(p.last_status_code, 183);
348
349        p.on_confirmed("dialog-1".to_string());
350        assert_eq!(p.session_id, "dialog-1");
351        assert_eq!(p.last_status_code, 200);
352        assert!(p.answer_time.is_some());
353
354        // A second confirmed must not move answer_time.
355        let first = p.answer_time;
356        p.on_confirmed("dialog-2".to_string());
357        assert_eq!(p.answer_time, first);
358    }
359
360    #[test]
361    fn try_set_answer_keeps_first() {
362        let mut p = CallProgress::default();
363        p.try_set_answer(EARLY_SDP);
364        p.try_set_answer("late");
365        assert_eq!(p.answer.as_deref(), Some(EARLY_SDP));
366    }
367
368    #[test]
369    fn set_hangup_reason_keeps_first() {
370        let mut p = CallProgress::default();
371        p.set_hangup_reason(CallRecordHangupReason::ByCaller);
372        p.set_hangup_reason(CallRecordHangupReason::ByCallee);
373        assert_eq!(p.hangup_reason, Some(CallRecordHangupReason::ByCaller));
374    }
375
376    #[test]
377    fn resolve_final_answer_uses_early_sdp_on_empty_body() {
378        let early = EARLY_SDP.to_string();
379        let (sdp, applied) = resolve_final_answer(Some(vec![]), Some(&early)).unwrap();
380        assert_eq!(sdp, EARLY_SDP);
381        assert!(applied);
382    }
383
384    #[test]
385    fn resolve_final_answer_uses_200ok_body_when_present() {
386        let early = EARLY_SDP.to_string();
387        let (sdp, applied) = resolve_final_answer(Some(b"final".to_vec()), Some(&early)).unwrap();
388        assert_eq!(sdp, "final");
389        assert!(!applied);
390    }
391
392    #[test]
393    fn resolve_final_answer_no_answer_at_all() {
394        assert!(resolve_final_answer(None, None).is_err());
395        assert!(resolve_final_answer(None, Some(&String::new())).is_err());
396    }
397
398    #[test]
399    fn termination_mapping() {
400        let info = CallProgress::termination(Some(&TerminatedReason::UacCancel));
401        assert_eq!((info.status_code, info.initiator), (487, "caller"));
402        assert_eq!(info.hangup_reason, CallRecordHangupReason::Canceled);
403
404        let info = CallProgress::termination(Some(&TerminatedReason::UasDecline));
405        assert_eq!((info.status_code, info.initiator), (603, "callee"));
406        assert_eq!(info.hangup_reason, CallRecordHangupReason::ByCallee);
407
408        let info = CallProgress::termination(None);
409        assert_eq!((info.status_code, info.initiator), (500, "system"));
410        assert_eq!(info.hangup_reason, CallRecordHangupReason::BySystem);
411    }
412
413    #[test]
414    fn leg_shared_extras_rcu() {
415        let leg = LegShared::new(42, false, CallProgress::default());
416        leg.set_extra("k", Value::String("v".into()));
417        assert_eq!(
418            leg.extras.load_full().get("k"),
419            Some(&Value::String("v".into()))
420        );
421    }
422}
active_call/call/state.rs

active_call/call/
state.rs
