1use crate::error::{Result, VoxRtcError};
2use crate::session::VoxRtcControlSession;
3use crate::socket::RawSocketClient;
4use crate::types::{ConnectionState, EventData, SessionBootstrap};
5use serde_json::Value;
6use std::env;
7use std::time::Duration;
8
9#[derive(Debug, Clone)]
10pub struct VoxRtcServerClientOptions {
11 pub http_base: String,
12 pub api_key: Option<String>,
13 pub socket_base: Option<String>,
14 pub socket_params: EventData,
15 pub connection_timeout: Duration,
16 pub max_reconnect_delay: Duration,
17 pub request_timeout: Duration,
18 pub join_timeout: Duration,
19}
20
21impl VoxRtcServerClientOptions {
22 pub fn new(http_base: impl Into<String>) -> Self {
23 Self {
24 http_base: http_base.into(),
25 api_key: None,
26 socket_base: None,
27 socket_params: EventData::new(),
28 connection_timeout: Duration::from_secs(10),
29 max_reconnect_delay: Duration::from_secs(30),
30 request_timeout: Duration::from_secs(15),
31 join_timeout: Duration::from_secs(10),
32 }
33 }
34}
35
36#[derive(Clone)]
37pub struct VoxRtcServerClient {
38 http_base: String,
39 api_key: Option<String>,
40 socket_base: String,
41 http: reqwest::Client,
42 socket: RawSocketClient,
43 connection_timeout: Duration,
44 join_timeout: Duration,
45}
46
47#[derive(Clone)]
48pub struct ControlledSession {
49 pub bootstrap: SessionBootstrap,
50 pub session: VoxRtcControlSession,
51}
52
53#[derive(Debug, Clone, Copy, Default)]
54pub struct SessionOptions {
55 pub join_timeout: Option<Duration>,
56}
57
58impl VoxRtcServerClient {
59 pub fn new(http_base: impl Into<String>) -> Result<Self> {
60 Self::with_options(VoxRtcServerClientOptions::new(http_base))
61 }
62
63 pub fn with_options(mut options: VoxRtcServerClientOptions) -> Result<Self> {
64 let http_base = normalize_base(&options.http_base);
65 let api_key = options
66 .api_key
67 .take()
68 .or_else(|| env::var("VOX_API_KEY").ok())
69 .map(|value| value.trim().to_owned())
70 .filter(|value| !value.is_empty());
71 let socket_base = options
72 .socket_base
73 .take()
74 .map(|base| normalize_base(&base))
75 .unwrap_or_else(|| default_socket_base(&http_base));
76 let socket_params = socket_params_with_api_key(options.socket_params, api_key.as_deref());
77 let http = reqwest::Client::builder()
78 .timeout(options.request_timeout)
79 .build()?;
80 let socket = RawSocketClient::new(
81 &socket_base,
82 socket_params,
83 options.connection_timeout,
84 options.max_reconnect_delay,
85 )?;
86 Ok(Self {
87 http_base,
88 api_key,
89 socket_base,
90 http,
91 socket,
92 connection_timeout: options.connection_timeout,
93 join_timeout: options.join_timeout,
94 })
95 }
96
97 pub fn http_base(&self) -> &str {
98 &self.http_base
99 }
100
101 pub fn socket_base(&self) -> &str {
102 &self.socket_base
103 }
104
105 pub fn connection_state(&self) -> ConnectionState {
106 self.socket.state()
107 }
108
109 pub async fn connect(&self) -> Result<()> {
110 if self.socket.state() == ConnectionState::Connected {
111 return Ok(());
112 }
113 self.socket.connect().await?;
114 let mut states = self.socket.subscribe_state();
115 tokio::time::timeout(self.connection_timeout, async move {
116 loop {
117 if *states.borrow_and_update() == ConnectionState::Connected {
118 return Ok(());
119 }
120 if states.changed().await.is_err() {
121 return Err(VoxRtcError::Disconnected);
122 }
123 }
124 })
125 .await
126 .map_err(|_| VoxRtcError::Timeout("PondSocket connection"))?
127 }
128
129 pub async fn disconnect(&self) {
130 self.socket.disconnect().await;
131 }
132
133 pub async fn create_session(&self) -> Result<SessionBootstrap> {
134 let mut request = self
135 .http
136 .post(format!("{}/v1/rtc/sessions", self.http_base))
137 .json(&serde_json::json!({}));
138 if let Some(api_key) = &self.api_key {
139 request = request.bearer_auth(api_key);
140 }
141 let response = request.send().await?;
142 let status = response.status();
143 let body = response.text().await?;
144 if !status.is_success() {
145 return Err(VoxRtcError::CreateSessionFailed { status, body });
146 }
147 Ok(serde_json::from_str(&body)?)
148 }
149
150 pub async fn attach_session(
151 &self,
152 session_id: impl Into<String>,
153 ) -> Result<VoxRtcControlSession> {
154 self.attach_session_with_options(session_id, SessionOptions::default())
155 .await
156 }
157
158 pub async fn attach_session_with_options(
159 &self,
160 session_id: impl Into<String>,
161 options: SessionOptions,
162 ) -> Result<VoxRtcControlSession> {
163 let session_id = session_id.into();
164 self.connect().await?;
165 let channel = self
166 .socket
167 .create_channel(format!("/rtc/{session_id}"), EventData::new())
168 .await;
169 let session =
170 VoxRtcControlSession::new(channel, session_id, self.resolve_join_timeout(options));
171 session.join().await?;
172 Ok(session)
173 }
174
175 pub async fn create_controlled_session(&self) -> Result<ControlledSession> {
176 self.create_controlled_session_with_options(SessionOptions::default())
177 .await
178 }
179
180 pub async fn create_controlled_session_with_options(
181 &self,
182 options: SessionOptions,
183 ) -> Result<ControlledSession> {
184 let bootstrap = self.create_session().await?;
185 let session = self
186 .attach_session_with_options(bootstrap.session_id.clone(), options)
187 .await?;
188 Ok(ControlledSession { bootstrap, session })
189 }
190
191 fn resolve_join_timeout(&self, options: SessionOptions) -> Duration {
192 options.join_timeout.unwrap_or(self.join_timeout)
193 }
194}
195
196fn socket_params_with_api_key(mut params: EventData, api_key: Option<&str>) -> EventData {
197 if let Some(api_key) = api_key {
198 params.insert("api_key".to_owned(), Value::String(api_key.to_owned()));
199 }
200 params
201}
202
203fn normalize_base(base: &str) -> String {
204 base.trim_end_matches('/').to_owned()
205}
206
207fn default_socket_base(http_base: &str) -> String {
208 format!("{}/v1/socket", normalize_base(http_base))
209}
210
211#[cfg(test)]
212mod tests {
213 use super::*;
214
215 #[test]
216 fn defaults_socket_base_from_http_base() {
217 let client = VoxRtcServerClient::new("https://vox.example.com/").unwrap();
218 assert_eq!(client.http_base(), "https://vox.example.com");
219 assert_eq!(client.socket_base(), "https://vox.example.com/v1/socket");
220 }
221
222 #[test]
223 fn new_returns_error_on_bad_url_instead_of_panicking() {
224 match VoxRtcServerClient::new("not a url") {
225 Err(VoxRtcError::InvalidUrl(_)) => {}
226 Err(other) => panic!("expected InvalidUrl, got {other:?}"),
227 Ok(_) => panic!("expected an error for a malformed URL"),
228 }
229 }
230
231 #[test]
232 fn forwards_connection_and_reconnect_timeouts() {
233 let mut options = VoxRtcServerClientOptions::new("https://vox.example.com");
234 options.connection_timeout = Duration::from_secs(3);
235 options.max_reconnect_delay = Duration::from_secs(45);
236 let client = VoxRtcServerClient::with_options(options).unwrap();
237 assert_eq!(client.connection_timeout, Duration::from_secs(3));
238 }
239
240 #[test]
241 fn session_options_override_the_client_join_timeout() {
242 let mut options = VoxRtcServerClientOptions::new("https://vox.example.com");
243 options.join_timeout = Duration::from_secs(12);
244 let client = VoxRtcServerClient::with_options(options).unwrap();
245
246 assert_eq!(
247 client.resolve_join_timeout(SessionOptions {
248 join_timeout: Some(Duration::from_secs(2)),
249 }),
250 Duration::from_secs(2)
251 );
252 assert_eq!(
253 client.resolve_join_timeout(SessionOptions::default()),
254 Duration::from_secs(12)
255 );
256 }
257
258 #[test]
259 fn injects_api_key_into_socket_params() {
260 let params = socket_params_with_api_key(EventData::new(), Some("secret"));
261 assert_eq!(params.get("api_key").and_then(Value::as_str), Some("secret"));
262 }
263
264 #[test]
265 fn omits_api_key_from_socket_params_when_absent() {
266 let params = socket_params_with_api_key(EventData::new(), None);
267 assert!(!params.contains_key("api_key"));
268 }
269}