1use crate::error::{Result, VoxRtcError};
2use crate::types::{ChannelState, ConnectionState, EventData};
3use pondsocket_client::{
4 Channel as PondChannel, ClientError, ClientOptions, ConnectionState as PondConnectionState,
5 PondClient,
6};
7use pondsocket_common::{ChannelEvent, ChannelState as PondChannelState};
8use std::sync::atomic::{AtomicBool, Ordering};
9use std::sync::{Arc, Mutex};
10use std::time::Duration;
11use tokio::sync::{broadcast, watch};
12use tokio::task::JoinHandle;
13
14const INITIAL_RECONNECT_DELAY: Duration = Duration::from_millis(200);
15
16#[derive(Clone)]
17pub(crate) struct RawSocketClient {
18 client: PondClient,
19 state_tx: watch::Sender<ConnectionState>,
20 active: Arc<AtomicBool>,
21 supervisor: Arc<ReconnectSupervisor>,
22 max_reconnect_delay: Duration,
23}
24
25struct ReconnectSupervisor {
26 handle: Mutex<Option<JoinHandle<()>>>,
27}
28
29impl ReconnectSupervisor {
30 fn new() -> Self {
31 Self {
32 handle: Mutex::new(None),
33 }
34 }
35
36 fn abort(&self) {
37 if let Some(handle) = self
38 .handle
39 .lock()
40 .expect("reconnect supervisor mutex poisoned")
41 .take()
42 {
43 handle.abort();
44 }
45 }
46}
47
48impl Drop for ReconnectSupervisor {
49 fn drop(&mut self) {
50 self.abort();
51 }
52}
53
54#[derive(Clone)]
55pub(crate) struct RawSocketChannel {
56 channel: PondChannel,
57 state_tx: watch::Sender<ChannelState>,
58 message_tx: broadcast::Sender<(String, EventData)>,
59}
60
61impl RawSocketClient {
62 pub(crate) fn new(
63 endpoint: &str,
64 params: EventData,
65 connection_timeout: Duration,
66 max_reconnect_delay: Duration,
67 ) -> Result<Self> {
68 let options = ClientOptions {
69 connection_timeout,
70 ..ClientOptions::default()
71 };
72 let client = PondClient::with_options(endpoint, Some(params), options)?;
73 let (state_tx, _) = watch::channel(map_connection_state(client.state()));
74
75 Ok(Self {
76 client,
77 state_tx,
78 active: Arc::new(AtomicBool::new(false)),
79 supervisor: Arc::new(ReconnectSupervisor::new()),
80 max_reconnect_delay,
81 })
82 }
83
84 fn ensure_supervisor(&self) {
85 let mut slot = self
86 .supervisor
87 .handle
88 .lock()
89 .expect("reconnect supervisor mutex poisoned");
90 if slot.as_ref().is_some_and(|handle| !handle.is_finished()) {
91 return;
92 }
93 *slot = Some(spawn_reconnect_supervisor(
94 self.client.clone(),
95 self.state_tx.clone(),
96 self.active.clone(),
97 self.max_reconnect_delay,
98 ));
99 }
100
101 pub(crate) fn state(&self) -> ConnectionState {
102 map_connection_state(self.client.state())
103 }
104
105 pub(crate) fn subscribe_state(&self) -> watch::Receiver<ConnectionState> {
106 self.state_tx.subscribe()
107 }
108
109 pub(crate) async fn connect(&self) -> Result<()> {
110 self.active.store(true, Ordering::SeqCst);
111 self.ensure_supervisor();
112 self.state_tx
113 .send_replace(map_connection_state(self.client.state()));
114 self.client.connect().await?;
115 self.state_tx
116 .send_replace(map_connection_state(self.client.state()));
117 Ok(())
118 }
119
120 pub(crate) async fn disconnect(&self) {
121 self.active.store(false, Ordering::SeqCst);
122 self.supervisor.abort();
123 self.client.disconnect().await;
124 self.state_tx
125 .send_replace(map_connection_state(self.client.state()));
126 }
127
128 pub(crate) async fn create_channel(
129 &self,
130 name: impl Into<String>,
131 params: EventData,
132 ) -> RawSocketChannel {
133 let channel = self.client.create_channel(name, Some(params)).await;
134 RawSocketChannel::new(channel)
135 }
136
137 #[cfg(test)]
138 pub(crate) fn supervisor_present(&self) -> bool {
139 self.supervisor
140 .handle
141 .lock()
142 .expect("reconnect supervisor mutex poisoned")
143 .is_some()
144 }
145
146 #[cfg(test)]
147 pub(crate) fn supervisor_finished(&self) -> bool {
148 self.supervisor
149 .handle
150 .lock()
151 .expect("reconnect supervisor mutex poisoned")
152 .as_ref()
153 .map(JoinHandle::is_finished)
154 .unwrap_or(true)
155 }
156}
157
158fn spawn_reconnect_supervisor(
159 client: PondClient,
160 state_tx: watch::Sender<ConnectionState>,
161 active: Arc<AtomicBool>,
162 max_reconnect_delay: Duration,
163) -> JoinHandle<()> {
164 let mut states = client.subscribe_state();
165 tokio::spawn(async move {
166 loop {
167 if states.changed().await.is_err() {
168 break;
169 }
170 if !active.load(Ordering::SeqCst) {
171 break;
172 }
173 let current = *states.borrow_and_update();
174 state_tx.send_replace(map_connection_state(current));
175 if current != PondConnectionState::Disconnected {
176 continue;
177 }
178 let mut delay = INITIAL_RECONNECT_DELAY;
179 while active.load(Ordering::SeqCst)
180 && client.state() == PondConnectionState::Disconnected
181 {
182 tokio::time::sleep(delay).await;
183 if !active.load(Ordering::SeqCst) {
184 break;
185 }
186 if client.connect().await.is_ok() {
187 state_tx.send_replace(map_connection_state(client.state()));
188 break;
189 }
190 delay = next_reconnect_delay(delay, max_reconnect_delay);
191 }
192 }
193 })
194}
195
196fn next_reconnect_delay(current: Duration, max: Duration) -> Duration {
197 let doubled = current.saturating_mul(2);
198 if doubled > max { max } else { doubled }
199}
200
201impl RawSocketChannel {
202 fn new(channel: PondChannel) -> Self {
203 let (state_tx, _) = watch::channel(map_channel_state(channel.state()));
204 let (message_tx, _) = broadcast::channel(1024);
205
206 let mut pond_states = channel.subscribe_state();
207 let mirror_state_tx = state_tx.clone();
208 tokio::spawn(async move {
209 loop {
210 mirror_state_tx.send_replace(map_channel_state(*pond_states.borrow_and_update()));
211 if pond_states.changed().await.is_err() {
212 break;
213 }
214 }
215 });
216
217 let mut pond_events = channel.subscribe_events();
218 let mirror_message_tx = message_tx.clone();
219 tokio::spawn(async move {
220 while let Ok(event) = pond_events.recv().await {
221 if let Some((event, payload)) = map_channel_event(event) {
222 let _ = mirror_message_tx.send((event, payload));
223 }
224 }
225 });
226
227 Self {
228 channel,
229 state_tx,
230 message_tx,
231 }
232 }
233
234 pub(crate) fn name(&self) -> &str {
235 self.channel.name()
236 }
237
238 pub(crate) fn subscribe_state(&self) -> watch::Receiver<ChannelState> {
239 self.state_tx.subscribe()
240 }
241
242 pub(crate) fn subscribe_messages(&self) -> broadcast::Receiver<(String, EventData)> {
243 self.message_tx.subscribe()
244 }
245
246 pub(crate) async fn decline_reason(&self) -> Option<EventData> {
247 self.channel.decline_reason().await
248 }
249
250 fn closed_error(&self) -> Option<VoxRtcError> {
251 match self.channel.state() {
252 PondChannelState::Closed | PondChannelState::Declined => {
253 Some(VoxRtcError::ChannelClosed)
254 }
255 _ => None,
256 }
257 }
258
259 pub(crate) async fn join(&self) -> Result<()> {
260 if let Some(error) = self.closed_error() {
261 return Err(error);
262 }
263 self.channel.join().await;
264 Ok(())
265 }
266
267 pub(crate) async fn leave(&self) -> Result<()> {
268 if let Some(error) = self.closed_error() {
269 return Err(error);
270 }
271 self.channel.leave().await;
272 Ok(())
273 }
274
275 pub(crate) async fn send_message(&self, event: &str, payload: EventData) -> Result<()> {
276 if let Some(error) = self.closed_error() {
277 return Err(error);
278 }
279 self.channel.send_message(event, Some(payload)).await;
280 Ok(())
281 }
282}
283
284fn map_connection_state(state: PondConnectionState) -> ConnectionState {
285 match state {
286 PondConnectionState::Connecting => ConnectionState::Connecting,
287 PondConnectionState::Connected => ConnectionState::Connected,
288 PondConnectionState::Disconnected => ConnectionState::Disconnected,
289 }
290}
291
292fn map_channel_state(state: PondChannelState) -> ChannelState {
293 match state {
294 PondChannelState::Idle => ChannelState::Idle,
295 PondChannelState::Joining => ChannelState::Joining,
296 PondChannelState::Joined => ChannelState::Joined,
297 PondChannelState::Closed => ChannelState::Closed,
298 PondChannelState::Declined => ChannelState::Declined,
299 PondChannelState::Stalled => ChannelState::Joining,
300 }
301}
302
303fn map_channel_event(event: ChannelEvent) -> Option<(String, EventData)> {
304 match event {
305 ChannelEvent::Message(message) => Some((message.event, message.payload)),
306 ChannelEvent::Presence(_) => None,
307 }
308}
309
310impl From<ClientError> for VoxRtcError {
311 fn from(value: ClientError) -> Self {
312 match value {
313 ClientError::Url(err) => Self::InvalidUrl(err),
314 ClientError::Serialization(err) => Self::Json(err),
315 ClientError::WebSocket(err) => Self::PondSocketClient(err.to_string()),
316 ClientError::NotConnected => Self::NotConnected,
317 ClientError::ChannelClosed => Self::ChannelClosed,
318 other => Self::PondSocketClient(other.to_string()),
319 }
320 }
321}
322
323#[cfg(test)]
324pub(crate) async fn test_channel() -> (RawSocketChannel, broadcast::Sender<(String, EventData)>) {
325 let client = PondClient::new("ws://localhost/socket", None).expect("valid test url");
326 let channel = client.create_channel("/rtc/test", None).await;
327 let raw = RawSocketChannel::new(channel);
328 let sender = raw.message_tx.clone();
329 (raw, sender)
330}
331
332#[cfg(test)]
333mod tests {
334 use super::*;
335
336 #[test]
337 fn distinguishes_not_connected_from_channel_closed() {
338 assert!(matches!(
339 VoxRtcError::from(ClientError::NotConnected),
340 VoxRtcError::NotConnected
341 ));
342 assert!(matches!(
343 VoxRtcError::from(ClientError::ChannelClosed),
344 VoxRtcError::ChannelClosed
345 ));
346 }
347
348 #[test]
349 fn reconnect_delay_doubles_then_caps() {
350 let max = Duration::from_secs(5);
351 assert_eq!(
352 next_reconnect_delay(Duration::from_millis(200), max),
353 Duration::from_millis(400)
354 );
355 assert_eq!(
356 next_reconnect_delay(Duration::from_secs(4), max),
357 Duration::from_secs(5)
358 );
359 assert_eq!(next_reconnect_delay(max, max), max);
360 }
361
362 #[tokio::test]
363 async fn reconnect_supervisor_terminates_after_disconnect() {
364 let client = RawSocketClient::new(
365 "ws://localhost/socket",
366 EventData::new(),
367 Duration::from_millis(50),
368 Duration::from_secs(1),
369 )
370 .expect("valid socket client");
371
372 client.active.store(true, Ordering::SeqCst);
373 client.ensure_supervisor();
374 assert!(
375 !client.supervisor_finished(),
376 "supervisor task must be running once started"
377 );
378
379 client.disconnect().await;
380
381 let terminated = tokio::time::timeout(Duration::from_secs(2), async {
382 while !client.supervisor_finished() {
383 tokio::time::sleep(Duration::from_millis(5)).await;
384 }
385 })
386 .await;
387 assert!(
388 terminated.is_ok(),
389 "supervisor task must finish after disconnect flips active false"
390 );
391 }
392
393 #[tokio::test]
394 async fn ensure_supervisor_respawns_after_disconnect_reconnect() {
395 let client = RawSocketClient::new(
396 "ws://localhost/socket",
397 EventData::new(),
398 Duration::from_millis(50),
399 Duration::from_secs(1),
400 )
401 .expect("valid socket client");
402
403 client.active.store(true, Ordering::SeqCst);
404 client.ensure_supervisor();
405 assert!(
406 !client.supervisor_finished(),
407 "supervisor task must be running once started"
408 );
409
410 client.disconnect().await;
411
412 let terminated = tokio::time::timeout(Duration::from_secs(2), async {
413 while !client.supervisor_finished() {
414 tokio::time::sleep(Duration::from_millis(5)).await;
415 }
416 })
417 .await;
418 assert!(
419 terminated.is_ok(),
420 "supervisor task must finish after disconnect flips active false"
421 );
422
423 client.active.store(true, Ordering::SeqCst);
424 client.ensure_supervisor();
425 assert!(
426 !client.supervisor_finished(),
427 "reconnect must replace the finished supervisor handle with a live one"
428 );
429 }
430
431 #[tokio::test]
432 async fn disconnect_clears_supervisor_slot_then_connect_respawns() {
433 let client = RawSocketClient::new(
434 "ws://localhost/socket",
435 EventData::new(),
436 Duration::from_millis(50),
437 Duration::from_secs(1),
438 )
439 .expect("valid socket client");
440
441 client.active.store(true, Ordering::SeqCst);
442 client.ensure_supervisor();
443 assert!(
444 client.supervisor_present(),
445 "supervisor handle must be present once started"
446 );
447
448 client.disconnect().await;
449 assert!(
450 !client.supervisor_present(),
451 "disconnect must take the supervisor handle out of the slot"
452 );
453
454 client.active.store(true, Ordering::SeqCst);
455 client.ensure_supervisor();
456 assert!(
457 client.supervisor_present(),
458 "an immediate reconnect must repopulate the empty slot"
459 );
460 assert!(
461 !client.supervisor_finished(),
462 "the respawned supervisor must be live"
463 );
464 }
465
466 #[tokio::test]
467 async fn send_message_errors_when_channel_closed() {
468 let (channel, _sender) = test_channel().await;
469 channel.leave().await.expect("first leave closes channel");
470 let error = channel
471 .send_message("response.start", EventData::new())
472 .await
473 .expect_err("closed channel must reject sends");
474 assert!(matches!(error, VoxRtcError::ChannelClosed));
475 }
476
477 #[tokio::test]
478 async fn join_and_leave_error_when_channel_closed() {
479 let (channel, _sender) = test_channel().await;
480 channel.leave().await.expect("first leave closes channel");
481 assert!(matches!(
482 channel
483 .join()
484 .await
485 .expect_err("cannot join a closed channel"),
486 VoxRtcError::ChannelClosed
487 ));
488 assert!(matches!(
489 channel
490 .leave()
491 .await
492 .expect_err("cannot leave an already-closed channel"),
493 VoxRtcError::ChannelClosed
494 ));
495 }
496
497 #[tokio::test]
498 async fn lagged_broadcast_does_not_stop_consumption() {
499 let (tx, mut rx) = broadcast::channel::<(String, EventData)>(2);
500 for index in 0..5u32 {
501 let _ = tx.send((format!("event-{index}"), EventData::new()));
502 }
503
504 let mut lagged = false;
505 let mut delivered = Vec::new();
506 loop {
507 match rx.try_recv() {
508 Ok(message) => delivered.push(message.0),
509 Err(broadcast::error::TryRecvError::Lagged(_)) => lagged = true,
510 Err(_) => break,
511 }
512 }
513
514 assert!(lagged, "small buffer overflow must surface a lag");
515 assert!(
516 delivered.contains(&"event-4".to_owned()),
517 "consumer must keep reading past the lag: {delivered:?}"
518 );
519 }
520}