active_call/call/
mod.rs

1use crate::{CallOption, ReferOption, media::recorder::RecorderOption, synthesis::SynthesisOption};
2use serde::{Deserialize, Serialize};
3use serde_with::skip_serializing_none;
4use std::{
5    collections::HashMap,
6    sync::{Arc, Mutex},
7};
8
9pub mod active_call;
10pub mod sip;
11pub mod state;
12pub mod tracks;
13pub use active_call::ActiveCall;
14pub use active_call::ActiveCallRef;
15pub use active_call::ActiveCallType;
16
17pub type CommandSender = tokio::sync::broadcast::Sender<Command>;
18pub type CommandReceiver = tokio::sync::broadcast::Receiver<Command>;
19
20// WebSocket Commands
21#[skip_serializing_none]
22#[derive(Debug, Deserialize, Serialize, Clone)]
23#[serde(
24    tag = "command",
25    rename_all = "camelCase",
26    rename_all_fields = "camelCase"
27)]
28pub enum Command {
29    Invite {
30        option: CallOption,
31    },
32    Accept {
33        option: CallOption,
34    },
35    Reject {
36        reason: String,
37        code: Option<u32>,
38    },
39    Ringing {
40        recorder: Option<RecorderOption>,
41        early_media: Option<bool>,
42        ringtone: Option<String>,
43    },
44    Tts {
45        text: String,
46        speaker: Option<String>,
47        /// If the play_id is the same, it will not interrupt the previous playback
48        play_id: Option<String>,
49        /// If auto_hangup is true, it means the call will be hung up automatically after the TTS playback is finished
50        auto_hangup: Option<bool>,
51        /// If streaming is true, it means the input text is streaming text,
52        /// and end_of_stream needs to be used to determine if it's finished,
53        /// equivalent to LLM's streaming output to TTS synthesis
54        streaming: Option<bool>,
55        /// If end_of_stream is true, it means the input text is finished
56        end_of_stream: Option<bool>,
57        option: Option<SynthesisOption>,
58        wait_input_timeout: Option<u32>,
59        /// if true, the text is base64 encoded pcm samples
60        base64: Option<bool>,
61        /// Customizing cache key for TTS Result
62        cache_key: Option<String>,
63    },
64    Play {
65        url: String,
66        play_id: Option<String>,
67        auto_hangup: Option<bool>,
68        wait_input_timeout: Option<u32>,
69        offset_ms: Option<u32>,
70    },
71    Interrupt {
72        graceful: Option<bool>,
73        fade_out_ms: Option<u32>,
74    },
75    Pause {},
76    Resume {},
77    Hangup {
78        reason: Option<String>,
79        initiator: Option<String>,
80        headers: Option<HashMap<String, String>>,
81        refer: Option<bool>,
82    },
83    Refer {
84        caller: String,
85        /// aor of the calee, e.g., sip:bob@restsend.com
86        callee: String,
87        options: Option<ReferOption>,
88    },
89    Message {
90        /// MIME body to send in a SIP MESSAGE request.
91        body: String,
92        /// Defaults to text/plain;charset=utf-8.
93        content_type: Option<String>,
94        /// Additional SIP headers for the MESSAGE request.
95        headers: Option<HashMap<String, String>>,
96        /// If true, send on the active refer dialog instead of the main call dialog.
97        refer: Option<bool>,
98    },
99    /// Bridge audio with another established call.
100    /// This creates separate bridge tracks for the two sessions and patches
101    /// audio bidirectionally. It does not replace the server-side track and
102    /// does not control hangup; each call keeps its own session/event flow.
103    Bridge {
104        /// session_id of the other call to bridge audio with
105        target_session_id: String,
106    },
107    /// Remove audio bridge tracks with another established call.
108    Unbridge {
109        /// session_id of the other call to unbridge from
110        target_session_id: String,
111    },
112    Mute {
113        track_id: Option<String>,
114    },
115    Unmute {
116        track_id: Option<String>,
117    },
118    History {
119        speaker: String,
120        text: String,
121    },
122    Custom {
123        sender: Option<String>,
124        data: serde_json::Value,
125    },
126    /// Trickle ICE: feed a remote candidate into an already-established session.
127    AddIceCandidate {
128        candidate: String,
129        sdp_mid: Option<String>,
130        sdp_mline_index: Option<u32>,
131    },
132}
133
134/// Routing state for managing stateful load balancing
135#[derive(Debug)]
136pub struct RoutingState {
137    /// Round-robin counters for each destination group
138    round_robin_counters: Arc<Mutex<HashMap<String, usize>>>,
139}
140
141impl Default for RoutingState {
142    fn default() -> Self {
143        Self::new()
144    }
145}
146
147impl RoutingState {
148    pub fn new() -> Self {
149        Self {
150            round_robin_counters: Arc::new(Mutex::new(HashMap::new())),
151        }
152    }
153
154    /// Get the next trunk index for round-robin selection
155    pub fn next_round_robin_index(&self, destination_key: &str, trunk_count: usize) -> usize {
156        if trunk_count == 0 {
157            return 0;
158        }
159
160        let mut counters = self.round_robin_counters.lock().unwrap();
161        let counter = counters
162            .entry(destination_key.to_string())
163            .or_insert_with(|| 0);
164        let r = *counter % trunk_count;
165        *counter += 1;
166        return r;
167    }
168}
169
170#[cfg(test)]
171mod tests {
172    use super::Command;
173
174    #[test]
175    fn message_command_deserializes_body() {
176        let command: Command = serde_json::from_value(serde_json::json!({
177            "command": "message",
178            "body": "customer_id=12345",
179            "contentType": "text/plain"
180        }))
181        .unwrap();
182
183        assert!(matches!(
184            command,
185            Command::Message {
186                body,
187                content_type: Some(content_type),
188                ..
189            } if body == "customer_id=12345" && content_type == "text/plain"
190        ));
191    }
192
193    #[test]
194    fn message_command_deserializes_legacy_text() {
195        let command: Command = serde_json::from_value(serde_json::json!({
196            "command": "message",
197            "body": "customer_id=12345"
198        }))
199        .unwrap();
200
201        assert!(matches!(
202            command,
203            Command::Message { body, .. } if body == "customer_id=12345"
204        ));
205    }
206
207    #[test]
208    fn message_command_serializes_body() {
209        let command = Command::Message {
210            body: "customer_id=12345".to_string(),
211            content_type: None,
212            headers: None,
213            refer: None,
214        };
215        let value = serde_json::to_value(command).unwrap();
216
217        assert_eq!(value["body"], "customer_id=12345");
218        assert!(value.get("text").is_none());
219    }
220}
active_call/call/mod.rs

active_call/call/
mod.rs
