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