1use std::collections::{HashMap, VecDeque};
2use std::sync::Arc;
3use std::time::Duration;
4
5use futures_util::{SinkExt, StreamExt};
6use pondsocket_common::{
7 ChannelEvent, ChannelState, ClientAction, ClientMessage, EventName, JoinParams, PondMessage,
8 PondPresence, PresenceEventType, PresenceMessage, ServerAction, ServerMessage, uuid,
9};
10use serde_json::{Map, Value};
11use thiserror::Error;
12use tokio::sync::{Mutex, broadcast, mpsc, oneshot, watch};
13use tokio::task::JoinHandle;
14use tokio_tungstenite::connect_async;
15use tokio_tungstenite::tungstenite::Message;
16use url::Url;
17
18pub mod typed;
19pub use typed::TypedChannel;
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub enum ConnectionState {
23 Connecting,
24 Connected,
25 Disconnected,
26}
27
28#[derive(Debug, Clone)]
29pub struct ClientOptions {
30 pub connection_timeout: Duration,
31 pub response_timeout: Duration,
32 pub max_queue_size: usize,
33}
34
35impl Default for ClientOptions {
36 fn default() -> Self {
37 Self {
38 connection_timeout: Duration::from_secs(10),
39 response_timeout: Duration::from_secs(5),
40 max_queue_size: 100,
41 }
42 }
43}
44
45#[derive(Debug, Error)]
46pub enum ClientError {
47 #[error("invalid websocket URL: {0}")]
48 Url(#[from] url::ParseError),
49 #[error("unsupported URL scheme: {0}")]
50 UnsupportedScheme(String),
51 #[error("websocket error: {0}")]
52 WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
53 #[error("serialization error: {0}")]
54 Serialization(#[from] serde_json::Error),
55 #[error("connection timed out")]
56 ConnectionTimeout,
57 #[error("client is not connected")]
58 NotConnected,
59 #[error("channel is closed")]
60 ChannelClosed,
61 #[error("response timed out")]
62 ResponseTimeout,
63}
64
65type Result<T> = std::result::Result<T, ClientError>;
66
67#[derive(Clone)]
68pub struct PondClient {
69 inner: Arc<ClientInner>,
70}
71
72struct ClientInner {
73 url: String,
74 options: ClientOptions,
75 state: watch::Sender<ConnectionState>,
76 conn_events: broadcast::Sender<ConnectionState>,
77 channels: Mutex<HashMap<String, Channel>>,
78 outbound: Mutex<Option<mpsc::Sender<ClientMessage>>>,
79 read_task: Mutex<Option<JoinHandle<()>>>,
80 write_task: Mutex<Option<JoinHandle<()>>>,
81}
82
83#[derive(Clone)]
84pub struct Channel {
85 inner: Arc<ChannelInner>,
86}
87
88struct ChannelInner {
89 name: String,
90 params: JoinParams,
91 client: Arc<ClientInner>,
92 state: watch::Sender<ChannelState>,
93 events: broadcast::Sender<ChannelEvent>,
94 presence: Mutex<Vec<PondPresence>>,
95 queue: Mutex<VecDeque<ClientMessage>>,
96 pending: Mutex<HashMap<String, oneshot::Sender<PondMessage>>>,
97 decline_reason: Mutex<Option<PondMessage>>,
98 conn_task: Mutex<Option<JoinHandle<()>>>,
99 closed: Mutex<bool>,
100}
101
102impl PondClient {
103 pub fn new(endpoint: impl AsRef<str>, params: Option<JoinParams>) -> Result<Self> {
104 Self::with_options(endpoint, params, ClientOptions::default())
105 }
106
107 pub fn with_options(
108 endpoint: impl AsRef<str>,
109 params: Option<JoinParams>,
110 options: ClientOptions,
111 ) -> Result<Self> {
112 let url = resolve_url(endpoint.as_ref(), params.as_ref())?;
113 let (state, _) = watch::channel(ConnectionState::Disconnected);
114 let (conn_events, _) = broadcast::channel(16);
115
116 Ok(Self {
117 inner: Arc::new(ClientInner {
118 url,
119 options,
120 state,
121 conn_events,
122 channels: Mutex::new(HashMap::new()),
123 outbound: Mutex::new(None),
124 read_task: Mutex::new(None),
125 write_task: Mutex::new(None),
126 }),
127 })
128 }
129
130 pub fn state(&self) -> ConnectionState {
131 *self.inner.state.borrow()
132 }
133
134 pub fn subscribe_state(&self) -> watch::Receiver<ConnectionState> {
135 self.inner.state.subscribe()
136 }
137
138 pub async fn create_channel(
139 &self,
140 name: impl Into<String>,
141 params: Option<JoinParams>,
142 ) -> Channel {
143 let name = name.into();
144 let mut channels = self.inner.channels.lock().await;
145 if let Some(channel) = channels.get(&name)
146 && channel.state() != ChannelState::Closed
147 && channel.state() != ChannelState::Declined
148 {
149 return channel.clone();
150 }
151
152 let (state, _) = watch::channel(ChannelState::Idle);
153 let (events, _) = broadcast::channel(100);
154 let channel = Channel {
155 inner: Arc::new(ChannelInner {
156 name: name.clone(),
157 params: params.unwrap_or_default(),
158 client: Arc::clone(&self.inner),
159 state,
160 events,
161 presence: Mutex::new(Vec::new()),
162 queue: Mutex::new(VecDeque::new()),
163 pending: Mutex::new(HashMap::new()),
164 decline_reason: Mutex::new(None),
165 conn_task: Mutex::new(None),
166 closed: Mutex::new(false),
167 }),
168 };
169
170 let watcher = channel.clone();
171 let mut conn_rx = self.inner.conn_events.subscribe();
172 let handle = tokio::spawn(async move {
173 loop {
174 match conn_rx.recv().await {
175 Ok(state) => watcher.on_connection_change(state).await,
176 Err(broadcast::error::RecvError::Lagged(_)) => {}
177 Err(broadcast::error::RecvError::Closed) => break,
178 }
179 if *watcher.inner.closed.lock().await {
180 break;
181 }
182 }
183 });
184 *channel.inner.conn_task.lock().await = Some(handle);
185
186 channels.insert(name, channel.clone());
187 channel
188 }
189
190 pub async fn connect(&self) -> Result<()> {
191 if self.state() != ConnectionState::Disconnected {
192 return Ok(());
193 }
194 self.inner.set_connection_state(ConnectionState::Connecting);
195 let connect = connect_async(&self.inner.url);
196 let (socket, _) = tokio::time::timeout(self.inner.options.connection_timeout, connect)
197 .await
198 .map_err(|_| ClientError::ConnectionTimeout)??;
199 let (mut writer, mut reader) = socket.split();
200 let (tx, mut rx) = mpsc::channel::<ClientMessage>(self.inner.options.max_queue_size);
201 *self.inner.outbound.lock().await = Some(tx);
202
203 let write_task = tokio::spawn(async move {
204 while let Some(message) = rx.recv().await {
205 let Ok(text) = serde_json::to_string(&message) else {
206 continue;
207 };
208 if writer.send(Message::Text(text.into())).await.is_err() {
209 break;
210 }
211 }
212 let _ = writer.close().await;
213 });
214
215 let inner = Arc::clone(&self.inner);
216 let read_task = tokio::spawn(async move {
217 while let Some(frame) = reader.next().await {
218 let text = match frame {
219 Ok(Message::Text(text)) => text.to_string(),
220 Ok(Message::Binary(bytes)) => match String::from_utf8(bytes.to_vec()) {
221 Ok(text) => text,
222 Err(_) => continue,
223 },
224 Ok(Message::Close(_)) => break,
225 Ok(_) => continue,
226 Err(_) => break,
227 };
228 let Ok(event) = pondsocket_common::parse_channel_event(&text) else {
229 continue;
230 };
231 inner.route_event(event).await;
232 }
233 inner.set_connection_state(ConnectionState::Disconnected);
234 *inner.outbound.lock().await = None;
235 });
236
237 *self.inner.read_task.lock().await = Some(read_task);
238 *self.inner.write_task.lock().await = Some(write_task);
239 self.inner.set_connection_state(ConnectionState::Connected);
240 Ok(())
241 }
242
243 pub async fn disconnect(&self) {
244 if let Some(task) = self.inner.read_task.lock().await.take() {
245 task.abort();
246 }
247 if let Some(task) = self.inner.write_task.lock().await.take() {
248 task.abort();
249 }
250 *self.inner.outbound.lock().await = None;
251 self.inner
252 .set_connection_state(ConnectionState::Disconnected);
253 let channels: Vec<Channel> = self.inner.channels.lock().await.values().cloned().collect();
254 for channel in channels {
255 channel.force_close().await;
256 }
257 self.inner.channels.lock().await.clear();
258 }
259}
260
261impl ClientInner {
262 async fn publish(&self, message: ClientMessage) -> Result<()> {
263 let tx = self
264 .outbound
265 .lock()
266 .await
267 .clone()
268 .ok_or(ClientError::NotConnected)?;
269 tx.send(message)
270 .await
271 .map_err(|_| ClientError::NotConnected)
272 }
273
274 async fn route_event(&self, event: ChannelEvent) {
275 let channel_name = match &event {
276 ChannelEvent::Message(message) => &message.channel_name,
277 ChannelEvent::Presence(message) => &message.channel_name,
278 };
279 let channel = self.channels.lock().await.get(channel_name).cloned();
280 if let Some(channel) = channel {
281 channel.handle_event(event).await;
282 }
283 }
284
285 fn set_connection_state(&self, state: ConnectionState) {
286 let changed = *self.state.borrow() != state;
287 self.state.send_replace(state);
288 if changed {
289 let _ = self.conn_events.send(state);
290 }
291 }
292}
293
294impl Channel {
295 pub fn name(&self) -> &str {
296 &self.inner.name
297 }
298
299 pub fn state(&self) -> ChannelState {
300 *self.inner.state.borrow()
301 }
302
303 pub fn subscribe_state(&self) -> watch::Receiver<ChannelState> {
304 self.inner.state.subscribe()
305 }
306
307 pub fn subscribe_events(&self) -> broadcast::Receiver<ChannelEvent> {
308 self.inner.events.subscribe()
309 }
310
311 pub async fn presence(&self) -> Vec<PondPresence> {
312 self.inner.presence.lock().await.clone()
313 }
314
315 pub async fn decline_reason(&self) -> Option<PondMessage> {
316 self.inner.decline_reason.lock().await.clone()
317 }
318
319 pub async fn join(&self) {
320 if *self.inner.closed.lock().await {
321 return;
322 }
323 if matches!(
324 self.state(),
325 ChannelState::Joining | ChannelState::Joined | ChannelState::Declined
326 ) {
327 return;
328 }
329 self.inner.state.send_replace(ChannelState::Joining);
330 self.enqueue_or_send(self.join_message()).await;
331 }
332
333 pub async fn leave(&self) {
334 if *self.inner.closed.lock().await {
335 return;
336 }
337 let message = ClientMessage {
338 action: ClientAction::LeaveChannel,
339 event: "LEAVE_CHANNEL".to_owned(),
340 payload: Map::new(),
341 channel_name: self.inner.name.clone(),
342 request_id: uuid(),
343 };
344 let _ = self.inner.client.publish(message).await;
345 self.force_close().await;
346 }
347
348 pub async fn send_message(&self, event: impl Into<String>, payload: Option<PondMessage>) {
349 if *self.inner.closed.lock().await {
350 return;
351 }
352 let message = ClientMessage {
353 action: ClientAction::Broadcast,
354 event: event.into(),
355 payload: payload.unwrap_or_default(),
356 channel_name: self.inner.name.clone(),
357 request_id: uuid(),
358 };
359 self.enqueue_or_send(message).await;
360 }
361
362 pub async fn send_for_response(
363 &self,
364 event: impl Into<String>,
365 payload: Option<PondMessage>,
366 timeout: Option<Duration>,
367 ) -> Result<PondMessage> {
368 if *self.inner.closed.lock().await {
369 return Err(ClientError::ChannelClosed);
370 }
371 let request_id = uuid();
372 let (tx, rx) = oneshot::channel();
373 self.inner
374 .pending
375 .lock()
376 .await
377 .insert(request_id.clone(), tx);
378 let message = ClientMessage {
379 action: ClientAction::Broadcast,
380 event: event.into(),
381 payload: payload.unwrap_or_default(),
382 channel_name: self.inner.name.clone(),
383 request_id: request_id.clone(),
384 };
385 self.enqueue_or_send(message).await;
386 let timeout = timeout.unwrap_or(self.inner.client.options.response_timeout);
387 let result = tokio::time::timeout(timeout, rx).await;
388 self.inner.pending.lock().await.remove(&request_id);
389 match result {
390 Ok(Ok(payload)) => Ok(payload),
391 _ => Err(ClientError::ResponseTimeout),
392 }
393 }
394
395 async fn enqueue_or_send(&self, message: ClientMessage) {
396 let connected = *self.inner.client.state.borrow() == ConnectionState::Connected;
397 let joined = self.state() == ChannelState::Joined;
398 let is_join = message.action == ClientAction::JoinChannel;
399 if connected
400 && (joined || is_join)
401 && self.inner.client.publish(message.clone()).await.is_ok()
402 {
403 return;
404 }
405 let mut queue = self.inner.queue.lock().await;
406 if queue.len() == self.inner.client.options.max_queue_size {
407 queue.pop_front();
408 }
409 queue.push_back(message);
410 }
411
412 async fn on_connection_change(&self, state: ConnectionState) {
413 if *self.inner.closed.lock().await {
414 return;
415 }
416 match state {
417 ConnectionState::Disconnected => {
418 if matches!(self.state(), ChannelState::Joined | ChannelState::Joining) {
419 self.inner.state.send_replace(ChannelState::Stalled);
420 }
421 }
422 ConnectionState::Connected => {
423 if matches!(self.state(), ChannelState::Stalled | ChannelState::Joining) {
424 self.inner
425 .queue
426 .lock()
427 .await
428 .retain(|message| message.action != ClientAction::JoinChannel);
429 self.inner.state.send_replace(ChannelState::Stalled);
430 self.join().await;
431 }
432 }
433 ConnectionState::Connecting => {}
434 }
435 }
436
437 async fn handle_event(&self, event: ChannelEvent) {
438 if *self.inner.closed.lock().await {
439 return;
440 }
441 match event {
442 ChannelEvent::Presence(message) => self.handle_presence(message).await,
443 ChannelEvent::Message(message) => self.handle_message(message).await,
444 }
445 }
446
447 async fn handle_presence(&self, message: PresenceMessage) {
448 *self.inner.presence.lock().await = message.payload.presence.clone();
449 let event = ChannelEvent::Presence(message.clone());
450 let _ = self.inner.events.send(event);
451 }
452
453 async fn handle_message(&self, message: ServerMessage) {
454 if message.action == ServerAction::System && message.event == "EVICTED" {
455 let _ = self.inner.events.send(ChannelEvent::Message(message));
456 self.force_close().await;
457 return;
458 }
459 if message.action == ServerAction::System
460 && message.event == event_name(EventName::Acknowledge)
461 {
462 if matches!(self.state(), ChannelState::Joining | ChannelState::Stalled) {
463 self.acknowledge().await;
464 }
465 return;
466 }
467 if message.action == ServerAction::System
468 && message.event == event_name(EventName::Unauthorized)
469 && matches!(self.state(), ChannelState::Joining | ChannelState::Stalled)
470 {
471 self.decline(message.payload).await;
472 return;
473 }
474 if message.action == ServerAction::System
475 && message.event == event_name(EventName::NotFound)
476 && matches!(self.state(), ChannelState::Joining | ChannelState::Stalled)
477 {
478 self.decline(message.payload).await;
479 return;
480 }
481 if let Some(tx) = self.inner.pending.lock().await.remove(&message.request_id) {
482 let _ = tx.send(message.payload);
483 return;
484 }
485 if self.state() == ChannelState::Joined {
486 let _ = self.inner.events.send(ChannelEvent::Message(message));
487 }
488 }
489
490 async fn acknowledge(&self) {
491 if self.state() != ChannelState::Joined {
492 self.inner.state.send_replace(ChannelState::Joined);
493 }
494 let mut queue = self.inner.queue.lock().await;
495 let pending: Vec<ClientMessage> = queue.drain(..).collect();
496 drop(queue);
497 for message in pending {
498 let _ = self.inner.client.publish(message).await;
499 }
500 }
501
502 async fn decline(&self, reason: PondMessage) {
503 *self.inner.decline_reason.lock().await = Some(reason);
504 self.inner.state.send_replace(ChannelState::Declined);
505 self.inner.queue.lock().await.clear();
506 self.inner.pending.lock().await.clear();
507 }
508
509 async fn force_close(&self) {
510 *self.inner.closed.lock().await = true;
511 self.inner.state.send_replace(ChannelState::Closed);
512 self.inner.queue.lock().await.clear();
513 self.inner.pending.lock().await.clear();
514 if let Some(handle) = self.inner.conn_task.lock().await.take() {
515 handle.abort();
516 }
517 }
518
519 fn join_message(&self) -> ClientMessage {
520 ClientMessage {
521 action: ClientAction::JoinChannel,
522 event: "JOIN_CHANNEL".to_owned(),
523 payload: self.inner.params.clone(),
524 channel_name: self.inner.name.clone(),
525 request_id: uuid(),
526 }
527 }
528}
529
530fn resolve_url(endpoint: &str, params: Option<&JoinParams>) -> Result<String> {
531 let mut url = Url::parse(endpoint)?;
532 match url.scheme() {
533 "http" => url
534 .set_scheme("ws")
535 .map_err(|_| ClientError::UnsupportedScheme("http".to_owned()))?,
536 "https" => url
537 .set_scheme("wss")
538 .map_err(|_| ClientError::UnsupportedScheme("https".to_owned()))?,
539 "ws" | "wss" => {}
540 scheme => return Err(ClientError::UnsupportedScheme(scheme.to_owned())),
541 }
542 if let Some(params) = params {
543 let mut pairs = url.query_pairs_mut();
544 for (key, value) in params {
545 let value = match value {
546 Value::String(value) => value.clone(),
547 other => other.to_string(),
548 };
549 pairs.append_pair(key, &value);
550 }
551 }
552 Ok(url.to_string())
553}
554
555fn event_name(event: EventName) -> String {
556 serde_json::to_string(&event)
557 .unwrap_or_default()
558 .trim_matches('"')
559 .to_owned()
560}
561
562#[allow(dead_code)]
563fn presence_event_name(event: PresenceEventType) -> String {
564 serde_json::to_string(&event)
565 .unwrap_or_default()
566 .trim_matches('"')
567 .to_owned()
568}
569
570#[cfg(test)]
571mod tests {
572 use super::*;
573 use pondsocket_common::{PondEvent, PondSchema, PresencePayload};
574 use serde::{Deserialize, Serialize};
575
576 #[test]
577 fn resolves_http_url_to_ws_with_params() {
578 let mut params = JoinParams::new();
579 params.insert("token".to_owned(), Value::String("abc".to_owned()));
580 let url = resolve_url("https://example.com/socket?room=one", Some(¶ms)).unwrap();
581 assert_eq!(url, "wss://example.com/socket?room=one&token=abc");
582 }
583
584 #[tokio::test]
585 async fn queues_join_message_before_connect() {
586 let client = PondClient::new("ws://example.com/socket", None).unwrap();
587 let channel = client.create_channel("room", None).await;
588 channel.join().await;
589 assert_eq!(channel.state(), ChannelState::Joining);
590 assert_eq!(channel.inner.queue.lock().await.len(), 1);
591 }
592
593 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
594 struct ChatPayload {
595 text: String,
596 }
597
598 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
599 struct AckPayload {
600 ok: bool,
601 }
602
603 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
604 struct Presence {
605 user_id: String,
606 }
607
608 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
609 struct Assigns {
610 role: String,
611 }
612
613 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
614 struct Join {
615 token: String,
616 }
617
618 struct Chat;
619 struct ChatSchema;
620
621 impl PondEvent for Chat {
622 type Payload = ChatPayload;
623 type Response = AckPayload;
624
625 const NAME: &'static str = "chat";
626 }
627
628 impl PondSchema for ChatSchema {
629 type Presence = Presence;
630 type Assigns = Assigns;
631 type JoinParams = Join;
632 }
633
634 #[tokio::test]
635 async fn typed_channel_sends_and_decodes_schema_values() {
636 let client = PondClient::new("ws://example.com/socket", None).unwrap();
637 let params = Join {
638 token: "secret".to_owned(),
639 };
640 let channel = client
641 .create_typed_channel::<ChatSchema>("room", Some(¶ms))
642 .await
643 .unwrap();
644
645 channel.join().await;
646 channel
647 .send::<Chat>(&ChatPayload {
648 text: "hello".to_owned(),
649 })
650 .await
651 .unwrap();
652
653 let queued = channel.raw().inner.queue.lock().await;
654 assert_eq!(queued[0].payload["token"], "secret");
655 assert_eq!(queued[1].event, "chat");
656 assert_eq!(queued[1].payload["text"], "hello");
657 drop(queued);
658
659 let message = ServerMessage {
660 action: ServerAction::Broadcast,
661 event: "chat".to_owned(),
662 channel_name: "room".to_owned(),
663 request_id: "r1".to_owned(),
664 payload: serde_json::from_value(serde_json::json!({ "text": "from server" })).unwrap(),
665 };
666 assert_eq!(
667 channel.decode_message::<Chat>(&message).unwrap(),
668 Some(ChatPayload {
669 text: "from server".to_owned()
670 })
671 );
672
673 let presence = PresenceMessage {
674 action: pondsocket_common::PresenceAction::Presence,
675 event: PresenceEventType::Join,
676 channel_name: "room".to_owned(),
677 request_id: "p1".to_owned(),
678 payload: PresencePayload {
679 changed: serde_json::from_value(serde_json::json!({ "user_id": "u1" })).unwrap(),
680 presence: vec![
681 serde_json::from_value(serde_json::json!({ "user_id": "u1" })).unwrap(),
682 ],
683 },
684 };
685 let (changed, users) = channel.decode_presence(&presence).unwrap();
686 assert_eq!(
687 changed,
688 Presence {
689 user_id: "u1".to_owned()
690 }
691 );
692 assert_eq!(users, vec![changed]);
693 }
694
695 use std::sync::atomic::{AtomicUsize, Ordering};
696 use tokio::net::{TcpListener, TcpStream};
697 use tokio_tungstenite::accept_async;
698
699 async fn serve_session(
700 stream: TcpStream,
701 joins: Arc<AtomicUsize>,
702 conn_id: u64,
703 mut kill: broadcast::Receiver<()>,
704 ) {
705 let Ok(ws) = accept_async(stream).await else {
706 return;
707 };
708 let (mut writer, mut reader) = ws.split();
709 loop {
710 tokio::select! {
711 _ = kill.recv() => {
712 let _ = writer.close().await;
713 return;
714 }
715 frame = reader.next() => {
716 let message = match frame {
717 Some(Ok(Message::Text(text))) => text.to_string(),
718 Some(Ok(_)) => continue,
719 _ => return,
720 };
721 let Ok(request) = serde_json::from_str::<ClientMessage>(&message) else {
722 continue;
723 };
724 if request.action != ClientAction::JoinChannel {
725 continue;
726 }
727 joins.fetch_add(1, Ordering::SeqCst);
728 let ack = ServerMessage {
729 action: ServerAction::System,
730 event: event_name(EventName::Acknowledge),
731 channel_name: request.channel_name.clone(),
732 request_id: request.request_id.clone(),
733 payload: Map::new(),
734 };
735 let _ = writer
736 .send(Message::Text(serde_json::to_string(&ack).unwrap().into()))
737 .await;
738 let mut payload = Map::new();
739 payload.insert("conn".to_owned(), Value::from(conn_id));
740 let greeting = ServerMessage {
741 action: ServerAction::Broadcast,
742 event: "greeting".to_owned(),
743 channel_name: request.channel_name.clone(),
744 request_id: uuid(),
745 payload,
746 };
747 let _ = writer
748 .send(Message::Text(serde_json::to_string(&greeting).unwrap().into()))
749 .await;
750 }
751 }
752 }
753 }
754
755 async fn wait_for_state(channel: &Channel, target: ChannelState) {
756 let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
757 while channel.state() != target {
758 if tokio::time::Instant::now() > deadline {
759 panic!(
760 "timed out waiting for {target:?}, still {:?}",
761 channel.state()
762 );
763 }
764 tokio::time::sleep(Duration::from_millis(5)).await;
765 }
766 }
767
768 async fn recv_greeting(events: &mut broadcast::Receiver<ChannelEvent>) -> u64 {
769 loop {
770 match tokio::time::timeout(Duration::from_secs(2), events.recv()).await {
771 Ok(Ok(ChannelEvent::Message(message))) if message.event == "greeting" => {
772 return message.payload.get("conn").and_then(Value::as_u64).unwrap();
773 }
774 Ok(Ok(_)) => {}
775 other => panic!("expected greeting event, got {other:?}"),
776 }
777 }
778 }
779
780 #[tokio::test]
781 async fn rejoins_channel_after_socket_drop_and_keeps_receiving_events() {
782 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
783 let addr = listener.local_addr().unwrap();
784 let joins = Arc::new(AtomicUsize::new(0));
785 let (kill, _) = broadcast::channel::<()>(4);
786
787 let server_joins = Arc::clone(&joins);
788 let server_kill = kill.clone();
789 tokio::spawn(async move {
790 let mut conn_id = 0;
791 while let Ok((stream, _)) = listener.accept().await {
792 conn_id += 1;
793 tokio::spawn(serve_session(
794 stream,
795 Arc::clone(&server_joins),
796 conn_id,
797 server_kill.subscribe(),
798 ));
799 }
800 });
801
802 let url = format!("ws://{addr}/socket");
803 let client = PondClient::new(&url, None).unwrap();
804 client.connect().await.unwrap();
805 let channel = client.create_channel("room", None).await;
806 let mut events = channel.subscribe_events();
807 channel.join().await;
808
809 wait_for_state(&channel, ChannelState::Joined).await;
810 assert_eq!(recv_greeting(&mut events).await, 1);
811 assert_eq!(joins.load(Ordering::SeqCst), 1);
812
813 kill.send(()).unwrap();
814 wait_for_state(&channel, ChannelState::Stalled).await;
815
816 client.connect().await.unwrap();
817 wait_for_state(&channel, ChannelState::Joined).await;
818 assert_eq!(joins.load(Ordering::SeqCst), 2);
819
820 assert_eq!(recv_greeting(&mut events).await, 2);
821 }
822
823 async fn serve_not_found(stream: TcpStream, mut kill: broadcast::Receiver<()>) {
824 let Ok(ws) = accept_async(stream).await else {
825 return;
826 };
827 let (mut writer, mut reader) = ws.split();
828 loop {
829 tokio::select! {
830 _ = kill.recv() => {
831 let _ = writer.close().await;
832 return;
833 }
834 frame = reader.next() => {
835 let message = match frame {
836 Some(Ok(Message::Text(text))) => text.to_string(),
837 Some(Ok(_)) => continue,
838 _ => return,
839 };
840 let Ok(request) = serde_json::from_str::<ClientMessage>(&message) else {
841 continue;
842 };
843 if request.action != ClientAction::JoinChannel {
844 continue;
845 }
846 let mut payload = Map::new();
847 payload.insert("channel".to_owned(), Value::from(request.channel_name.clone()));
848 let not_found = ServerMessage {
849 action: ServerAction::System,
850 event: event_name(EventName::NotFound),
851 channel_name: request.channel_name.clone(),
852 request_id: request.request_id.clone(),
853 payload,
854 };
855 let _ = writer
856 .send(Message::Text(serde_json::to_string(¬_found).unwrap().into()))
857 .await;
858 }
859 }
860 }
861 }
862
863 async fn serve_join_without_ack(
864 stream: TcpStream,
865 joins: Arc<AtomicUsize>,
866 mut kill: broadcast::Receiver<()>,
867 ) {
868 let Ok(ws) = accept_async(stream).await else {
869 return;
870 };
871 let (mut writer, mut reader) = ws.split();
872 loop {
873 tokio::select! {
874 _ = kill.recv() => {
875 let _ = writer.close().await;
876 return;
877 }
878 frame = reader.next() => {
879 match frame {
880 Some(Ok(Message::Text(text))) => {
881 if let Ok(request) = serde_json::from_str::<ClientMessage>(&text)
882 && request.action == ClientAction::JoinChannel
883 {
884 joins.fetch_add(1, Ordering::SeqCst);
885 }
886 }
887 Some(Ok(_)) => {}
888 _ => return,
889 }
890 }
891 }
892 }
893 }
894
895 #[tokio::test]
896 async fn transitions_to_declined_on_not_found() {
897 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
898 let addr = listener.local_addr().unwrap();
899 let (kill, _) = broadcast::channel::<()>(4);
900
901 let server_kill = kill.clone();
902 tokio::spawn(async move {
903 while let Ok((stream, _)) = listener.accept().await {
904 tokio::spawn(serve_not_found(stream, server_kill.subscribe()));
905 }
906 });
907
908 let url = format!("ws://{addr}/socket");
909 let client = PondClient::new(&url, None).unwrap();
910 client.connect().await.unwrap();
911 let channel = client.create_channel("room", None).await;
912 channel.join().await;
913
914 wait_for_state(&channel, ChannelState::Declined).await;
915 let reason = channel.decline_reason().await.unwrap();
916 assert_eq!(reason.get("channel").and_then(Value::as_str), Some("room"));
917 assert!(channel.inner.queue.lock().await.is_empty());
918 assert!(channel.inner.pending.lock().await.is_empty());
919 }
920
921 #[tokio::test]
922 async fn rejoins_channel_that_was_mid_join_when_socket_dropped() {
923 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
924 let addr = listener.local_addr().unwrap();
925 let joins = Arc::new(AtomicUsize::new(0));
926 let (kill, _) = broadcast::channel::<()>(4);
927
928 let server_joins = Arc::clone(&joins);
929 let server_kill = kill.clone();
930 tokio::spawn(async move {
931 let mut conn_id = 0;
932 while let Ok((stream, _)) = listener.accept().await {
933 conn_id += 1;
934 let joins = Arc::clone(&server_joins);
935 if conn_id == 1 {
936 tokio::spawn(serve_join_without_ack(
937 stream,
938 joins,
939 server_kill.subscribe(),
940 ));
941 } else {
942 tokio::spawn(serve_session(
943 stream,
944 joins,
945 conn_id,
946 server_kill.subscribe(),
947 ));
948 }
949 }
950 });
951
952 let url = format!("ws://{addr}/socket");
953 let client = PondClient::new(&url, None).unwrap();
954 client.connect().await.unwrap();
955 let channel = client.create_channel("room", None).await;
956 let mut events = channel.subscribe_events();
957 channel.join().await;
958
959 assert_eq!(channel.state(), ChannelState::Joining);
960 let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
961 while joins.load(Ordering::SeqCst) < 1 {
962 if tokio::time::Instant::now() > deadline {
963 panic!("server never observed the initial join");
964 }
965 tokio::time::sleep(Duration::from_millis(5)).await;
966 }
967
968 kill.send(()).unwrap();
969 wait_for_state(&channel, ChannelState::Stalled).await;
970
971 client.connect().await.unwrap();
972 wait_for_state(&channel, ChannelState::Joined).await;
973 assert_eq!(joins.load(Ordering::SeqCst), 2);
974 assert_eq!(recv_greeting(&mut events).await, 2);
975 }
976
977 fn system_message(
978 event: EventName,
979 channel: &str,
980 mut payload: Map<String, Value>,
981 ) -> ServerMessage {
982 payload
983 .entry("channel".to_owned())
984 .or_insert_with(|| Value::from(channel.to_owned()));
985 ServerMessage {
986 action: ServerAction::System,
987 event: event_name(event),
988 channel_name: channel.to_owned(),
989 request_id: uuid(),
990 payload,
991 }
992 }
993
994 #[tokio::test]
995 async fn adv_not_found_after_joined_leaves_live_channel_intact() {
996 let client = PondClient::new("ws://example.com/socket", None).unwrap();
997 let channel = client.create_channel("room", None).await;
998 channel.inner.state.send_replace(ChannelState::Joined);
999 let mut events = channel.subscribe_events();
1000
1001 channel
1002 .handle_message(system_message(EventName::NotFound, "room", Map::new()))
1003 .await;
1004
1005 assert_eq!(channel.state(), ChannelState::Joined);
1006 assert!(channel.decline_reason().await.is_none());
1007 match events.try_recv() {
1008 Ok(ChannelEvent::Message(m)) => assert_eq!(m.event, event_name(EventName::NotFound)),
1009 other => panic!("expected NOT_FOUND surfaced as message event, got {other:?}"),
1010 }
1011 }
1012
1013 #[tokio::test]
1014 async fn adv_unauthorized_after_joined_leaves_live_channel_intact() {
1015 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1016 let channel = client.create_channel("room", None).await;
1017 channel.inner.state.send_replace(ChannelState::Joined);
1018
1019 channel
1020 .handle_message(system_message(EventName::Unauthorized, "room", Map::new()))
1021 .await;
1022
1023 assert_eq!(channel.state(), ChannelState::Joined);
1024 assert!(channel.decline_reason().await.is_none());
1025 }
1026
1027 #[tokio::test]
1028 async fn adv_decline_errors_pending_response_promptly() {
1029 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1030 let channel = client.create_channel("room", None).await;
1031
1032 let asker = channel.clone();
1033 let handle = tokio::spawn(async move {
1034 asker
1035 .send_for_response("ask", None, Some(Duration::from_secs(30)))
1036 .await
1037 });
1038
1039 let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
1040 while channel.inner.pending.lock().await.is_empty() {
1041 if tokio::time::Instant::now() > deadline {
1042 panic!("pending request never registered");
1043 }
1044 tokio::time::sleep(Duration::from_millis(5)).await;
1045 }
1046
1047 channel.decline(Map::new()).await;
1048
1049 let joined = tokio::time::timeout(Duration::from_secs(2), handle)
1050 .await
1051 .expect("send_for_response hung after decline instead of returning")
1052 .unwrap();
1053 assert!(joined.is_err());
1054 assert!(channel.inner.pending.lock().await.is_empty());
1055 assert!(channel.inner.queue.lock().await.is_empty());
1056 }
1057
1058 #[tokio::test]
1059 async fn adv_duplicate_decline_keeps_latest_reason_without_panic() {
1060 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1061 let channel = client.create_channel("room", None).await;
1062
1063 let mut first = Map::new();
1064 first.insert("code".to_owned(), Value::from("first"));
1065 channel.decline(first).await;
1066
1067 let mut second = Map::new();
1068 second.insert("code".to_owned(), Value::from("second"));
1069 channel.decline(second).await;
1070
1071 assert_eq!(channel.state(), ChannelState::Declined);
1072 assert_eq!(
1073 channel
1074 .decline_reason()
1075 .await
1076 .unwrap()
1077 .get("code")
1078 .and_then(Value::as_str),
1079 Some("second")
1080 );
1081 }
1082
1083 #[tokio::test]
1084 async fn adv_leave_during_stall_blocks_rejoin_on_reconnect() {
1085 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1086 let channel = client.create_channel("room", None).await;
1087 channel.join().await;
1088 assert_eq!(channel.state(), ChannelState::Joining);
1089
1090 channel
1091 .on_connection_change(ConnectionState::Disconnected)
1092 .await;
1093 assert_eq!(channel.state(), ChannelState::Stalled);
1094
1095 channel.leave().await;
1096 assert_eq!(channel.state(), ChannelState::Closed);
1097
1098 channel
1099 .on_connection_change(ConnectionState::Connected)
1100 .await;
1101 assert_eq!(channel.state(), ChannelState::Closed);
1102 assert!(channel.inner.queue.lock().await.is_empty());
1103 }
1104
1105 #[tokio::test]
1106 async fn adv_join_before_connect_completes_on_connect() {
1107 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1108 let addr = listener.local_addr().unwrap();
1109 let joins = Arc::new(AtomicUsize::new(0));
1110 let (kill, _) = broadcast::channel::<()>(4);
1111
1112 let server_joins = Arc::clone(&joins);
1113 let server_kill = kill.clone();
1114 tokio::spawn(async move {
1115 let mut conn_id = 0;
1116 while let Ok((stream, _)) = listener.accept().await {
1117 conn_id += 1;
1118 tokio::spawn(serve_session(
1119 stream,
1120 Arc::clone(&server_joins),
1121 conn_id,
1122 server_kill.subscribe(),
1123 ));
1124 }
1125 });
1126
1127 let url = format!("ws://{addr}/socket");
1128 let client = PondClient::new(&url, None).unwrap();
1129 let channel = client.create_channel("room", None).await;
1130 channel.join().await;
1131 assert_eq!(channel.state(), ChannelState::Joining);
1132 assert_eq!(channel.inner.queue.lock().await.len(), 1);
1133
1134 client.connect().await.unwrap();
1135 wait_for_state(&channel, ChannelState::Joined).await;
1136 assert_eq!(joins.load(Ordering::SeqCst), 1);
1137 assert!(channel.inner.queue.lock().await.is_empty());
1138 }
1139
1140 async fn wait_for_pending(channel: &Channel) -> String {
1141 let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
1142 loop {
1143 if let Some(id) = channel.inner.pending.lock().await.keys().next().cloned() {
1144 return id;
1145 }
1146 if tokio::time::Instant::now() > deadline {
1147 panic!("pending request never registered");
1148 }
1149 tokio::time::sleep(Duration::from_millis(5)).await;
1150 }
1151 }
1152
1153 #[tokio::test]
1154 async fn adv_event_not_found_on_joined_channel_resolves_pending_fast() {
1155 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1156 let channel = client.create_channel("room", None).await;
1157 channel.inner.state.send_replace(ChannelState::Joined);
1158
1159 let asker = channel.clone();
1160 let handle = tokio::spawn(async move {
1161 let start = tokio::time::Instant::now();
1162 let result = asker
1163 .send_for_response("no_handler", None, Some(Duration::from_millis(300)))
1164 .await;
1165 (start.elapsed(), result)
1166 });
1167 let request_id = wait_for_pending(&channel).await;
1168
1169 let mut payload = Map::new();
1170 payload.insert("channel".to_owned(), Value::from("room"));
1171 payload.insert("event".to_owned(), Value::from("no_handler"));
1172 let not_found = ServerMessage {
1173 action: ServerAction::System,
1174 event: event_name(EventName::NotFound),
1175 channel_name: "room".to_owned(),
1176 request_id,
1177 payload,
1178 };
1179 channel.handle_message(not_found).await;
1180
1181 let (elapsed, result) = tokio::time::timeout(Duration::from_secs(2), handle)
1182 .await
1183 .expect("request hung")
1184 .unwrap();
1185 let resolved = result.expect("pending request should resolve, not time out");
1186 assert_eq!(
1187 resolved.get("event").and_then(Value::as_str),
1188 Some("no_handler")
1189 );
1190 assert!(elapsed < Duration::from_millis(250));
1191 assert_eq!(channel.state(), ChannelState::Joined);
1192 assert!(channel.decline_reason().await.is_none());
1193 }
1194
1195 #[tokio::test]
1196 async fn adv_evicted_closes_client_channel() {
1197 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1198 let channel = client.create_channel("room", None).await;
1199 channel.inner.state.send_replace(ChannelState::Joined);
1200 let mut events = channel.subscribe_events();
1201
1202 let mut payload = Map::new();
1203 payload.insert("reason".to_owned(), Value::from("kicked"));
1204 let evicted = ServerMessage {
1205 action: ServerAction::System,
1206 event: "EVICTED".to_owned(),
1207 channel_name: "room".to_owned(),
1208 request_id: uuid(),
1209 payload,
1210 };
1211 channel.handle_message(evicted).await;
1212
1213 match tokio::time::timeout(Duration::from_secs(1), events.recv()).await {
1214 Ok(Ok(ChannelEvent::Message(m))) => assert_eq!(m.event, "EVICTED"),
1215 other => panic!("expected EVICTED forwarded as message event, got {other:?}"),
1216 }
1217 assert_eq!(channel.state(), ChannelState::Closed);
1218 }
1219
1220 #[tokio::test]
1221 async fn adv_late_response_after_timeout_surfaces_as_message() {
1222 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1223 let channel = client.create_channel("room", None).await;
1224 channel.inner.state.send_replace(ChannelState::Joined);
1225 let mut events = channel.subscribe_events();
1226
1227 let mut payload = Map::new();
1228 payload.insert("ok".to_owned(), Value::from(true));
1229 let late = ServerMessage {
1230 action: ServerAction::Broadcast,
1231 event: "reply".to_owned(),
1232 channel_name: "room".to_owned(),
1233 request_id: "already-timed-out".to_owned(),
1234 payload,
1235 };
1236 channel.handle_message(late).await;
1237
1238 match tokio::time::timeout(Duration::from_secs(1), events.recv()).await {
1239 Ok(Ok(ChannelEvent::Message(m))) => assert_eq!(m.event, "reply"),
1240 other => panic!("expected late response surfaced as message event, got {other:?}"),
1241 }
1242 }
1243
1244 #[tokio::test]
1245 async fn adv_duplicate_response_second_surfaces_as_message() {
1246 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1247 let channel = client.create_channel("room", None).await;
1248 channel.inner.state.send_replace(ChannelState::Joined);
1249 let mut events = channel.subscribe_events();
1250
1251 let asker = channel.clone();
1252 let handle = tokio::spawn(async move {
1253 asker
1254 .send_for_response("ask", None, Some(Duration::from_secs(30)))
1255 .await
1256 });
1257 let request_id = wait_for_pending(&channel).await;
1258
1259 let mut first_payload = Map::new();
1260 first_payload.insert("n".to_owned(), Value::from(1));
1261 let first = ServerMessage {
1262 action: ServerAction::Broadcast,
1263 event: "reply".to_owned(),
1264 channel_name: "room".to_owned(),
1265 request_id: request_id.clone(),
1266 payload: first_payload,
1267 };
1268 channel.handle_message(first).await;
1269
1270 let resolved = tokio::time::timeout(Duration::from_secs(2), handle)
1271 .await
1272 .expect("request hung")
1273 .unwrap()
1274 .unwrap();
1275 assert_eq!(resolved.get("n").and_then(Value::as_i64), Some(1));
1276
1277 let mut second_payload = Map::new();
1278 second_payload.insert("n".to_owned(), Value::from(2));
1279 let second = ServerMessage {
1280 action: ServerAction::Broadcast,
1281 event: "reply".to_owned(),
1282 channel_name: "room".to_owned(),
1283 request_id,
1284 payload: second_payload,
1285 };
1286 channel.handle_message(second).await;
1287
1288 match tokio::time::timeout(Duration::from_secs(1), events.recv()).await {
1289 Ok(Ok(ChannelEvent::Message(m))) => {
1290 assert_eq!(m.payload.get("n").and_then(Value::as_i64), Some(2))
1291 }
1292 other => panic!("expected duplicate response surfaced as message event, got {other:?}"),
1293 }
1294 }
1295
1296 #[tokio::test]
1297 async fn adv_leave_cancels_inflight_request_with_error() {
1298 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1299 let channel = client.create_channel("room", None).await;
1300 channel.inner.state.send_replace(ChannelState::Joined);
1301
1302 let asker = channel.clone();
1303 let handle = tokio::spawn(async move {
1304 asker
1305 .send_for_response("ask", None, Some(Duration::from_secs(30)))
1306 .await
1307 });
1308 wait_for_pending(&channel).await;
1309
1310 channel.leave().await;
1311
1312 let result = tokio::time::timeout(Duration::from_secs(2), handle)
1313 .await
1314 .expect("request hung after leave")
1315 .unwrap();
1316 assert!(result.is_err());
1317 assert_eq!(channel.state(), ChannelState::Closed);
1318 }
1319
1320 #[tokio::test]
1321 async fn adv_double_leave_is_safe() {
1322 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1323 let channel = client.create_channel("room", None).await;
1324 channel.inner.state.send_replace(ChannelState::Joined);
1325 channel.leave().await;
1326 channel.leave().await;
1327 assert_eq!(channel.state(), ChannelState::Closed);
1328 }
1329
1330 #[tokio::test]
1331 async fn adv_not_found_on_joined_with_pending_now_resolves_request() {
1332 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1333 let channel = client.create_channel("room", None).await;
1334 channel.inner.state.send_replace(ChannelState::Joined);
1335
1336 let asker = channel.clone();
1337 let handle = tokio::spawn(async move {
1338 let start = tokio::time::Instant::now();
1339 let result = asker
1340 .send_for_response("no_handler", None, Some(Duration::from_secs(30)))
1341 .await;
1342 (start.elapsed(), result)
1343 });
1344 let request_id = wait_for_pending(&channel).await;
1345
1346 let mut payload = Map::new();
1347 payload.insert("channel".to_owned(), Value::from("room"));
1348 payload.insert("event".to_owned(), Value::from("no_handler"));
1349 let not_found = ServerMessage {
1350 action: ServerAction::System,
1351 event: event_name(EventName::NotFound),
1352 channel_name: "room".to_owned(),
1353 request_id,
1354 payload,
1355 };
1356 channel.handle_message(not_found).await;
1357
1358 let (elapsed, result) = tokio::time::timeout(Duration::from_secs(2), handle)
1359 .await
1360 .expect("request hung")
1361 .unwrap();
1362 let payload = result.expect("expected NOT_FOUND to resolve the pending request");
1363 assert_eq!(
1364 payload.get("event").and_then(Value::as_str),
1365 Some("no_handler")
1366 );
1367 assert!(elapsed < Duration::from_secs(5));
1368 assert_eq!(channel.state(), ChannelState::Joined);
1369 }
1370
1371 #[tokio::test]
1372 async fn adv_stray_acknowledge_on_joined_is_consumed() {
1373 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1374 let channel = client.create_channel("room", None).await;
1375 channel.inner.state.send_replace(ChannelState::Joined);
1376 let mut events = channel.subscribe_events();
1377
1378 channel
1379 .handle_message(system_message(EventName::Acknowledge, "room", Map::new()))
1380 .await;
1381
1382 assert_eq!(channel.state(), ChannelState::Joined);
1383 assert!(events.try_recv().is_err());
1384 }
1385
1386 #[tokio::test]
1387 async fn adv_evicted_cancels_inflight_request_and_closes() {
1388 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1389 let channel = client.create_channel("room", None).await;
1390 channel.inner.state.send_replace(ChannelState::Joined);
1391
1392 let asker = channel.clone();
1393 let handle = tokio::spawn(async move {
1394 asker
1395 .send_for_response("ask", None, Some(Duration::from_secs(30)))
1396 .await
1397 });
1398 wait_for_pending(&channel).await;
1399
1400 let mut payload = Map::new();
1401 payload.insert("reason".to_owned(), Value::from("kicked"));
1402 let evicted = ServerMessage {
1403 action: ServerAction::System,
1404 event: "EVICTED".to_owned(),
1405 channel_name: "room".to_owned(),
1406 request_id: uuid(),
1407 payload,
1408 };
1409 channel.handle_message(evicted).await;
1410
1411 let result = tokio::time::timeout(Duration::from_secs(2), handle)
1412 .await
1413 .expect("request hung after eviction")
1414 .unwrap();
1415 assert!(result.is_err());
1416 assert_eq!(channel.state(), ChannelState::Closed);
1417 }
1418
1419 #[tokio::test]
1420 async fn adv_evicted_for_unknown_channel_is_safe() {
1421 let client = PondClient::new("ws://example.com/socket", None).unwrap();
1422 let _present = client.create_channel("room", None).await;
1423
1424 let mut payload = Map::new();
1425 payload.insert("reason".to_owned(), Value::from("kicked"));
1426 let evicted = ServerMessage {
1427 action: ServerAction::System,
1428 event: "EVICTED".to_owned(),
1429 channel_name: "ghost".to_owned(),
1430 request_id: uuid(),
1431 payload,
1432 };
1433 client
1434 .inner
1435 .route_event(ChannelEvent::Message(evicted))
1436 .await;
1437
1438 assert_eq!(
1439 client.create_channel("ghost", None).await.state(),
1440 ChannelState::Idle
1441 );
1442 }
1443}