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

active_call/call/
mod.rs
