+2 -1

TODO.md

reviewed

··· 25 25 - [ ] Backend : More tests 26 26 - [x] Lobby tests 27 27 - [x] Game end test for actual return from loop 28 28 - - [ ] More transport crates tests 28 28 + - [x] More transport crate tests 29 29 + - [ ] Signaling is wrong, only kick everyone else on host leave if the lobby is open 29 30 - [x] Organize signalling and seperate out more logic 30 31 - [x] Signaling tests 31 32 - [ ] Testing crate for integration testing?

+627 -19

manhunt-transport/src/matchbox.rs

reviewed

··· 1 1 - use std::{collections::HashSet, pin::Pin, sync::Arc}; 1 1 + use std::{collections::HashSet, marker::PhantomData, pin::Pin, sync::Arc}; 2 2 3 3 use anyhow::{Context, anyhow}; 4 4 use futures::{ 5 5 - SinkExt, StreamExt, 5 5 + SinkExt, Stream, StreamExt, 6 6 channel::mpsc::{UnboundedReceiver, UnboundedSender}, 7 7 }; 8 8 use log::{error, info}; ··· 22 22 type MsgPair = (Option<Uuid>, TransportMessage); 23 23 type Queue = QueuePair<MsgPair>; 24 24 25 25 - pub struct MatchboxTransport { 25 25 + pub struct MatchboxTransport<S: SocketImpl = WebRtcSocket> { 26 26 my_id: Uuid, 27 27 incoming: Queue, 28 28 all_peers: Mutex<HashSet<Uuid>>, 29 29 msg_sender: UnboundedSender<MatchboxMsgPair>, 30 30 cancel_token: CancellationToken, 31 31 + phantom: PhantomData<S>, 31 32 } 32 33 33 34 type LoopFutRes = Result<(), SocketError>; 34 35 35 36 type MatchboxMsgPair = (PeerId, Box<[u8]>); 37 37 + type MatchboxSender = UnboundedSender<MatchboxMsgPair>; 38 38 + type MatchboxReceiver = UnboundedReceiver<MatchboxMsgPair>; 39 39 + type MatchboxChannel = (MatchboxSender, MatchboxReceiver); 36 40 37 41 fn map_socket_error(err: SocketError) -> anyhow::Error { 38 42 match err { ··· 41 45 } 42 46 } 43 47 44 44 - impl MatchboxTransport { 45 45 - pub async fn new(join_code: &str, is_host: bool) -> Result<Arc<Self>> { 46 46 - let (itx, irx) = mpsc::channel(15); 48 48 + type FutPin<T> = Pin<Box<dyn Future<Output = T> + Send>>; 49 49 + type MessageLoopFuture = FutPin<LoopFutRes>; 50 50 + type PeerMsg = (PeerId, PeerState); 51 51 + 52 52 + pub trait SocketImpl: Unpin + Send + Sync + Sized + Stream<Item = PeerMsg> { 53 53 + fn new(room_url: &str) -> (Self, MessageLoopFuture); 54 54 + fn get_id(&mut self) -> Option<PeerId>; 55 55 + fn take_channel(&mut self) -> MatchboxChannel; 56 56 + } 57 57 + 58 58 + impl SocketImpl for WebRtcSocket { 59 59 + fn new(room_url: &str) -> (Self, MessageLoopFuture) { 60 60 + Self::new_reliable(room_url) 61 61 + } 62 62 + 63 63 + fn get_id(&mut self) -> Option<PeerId> { 64 64 + self.id() 65 65 + } 47 66 67 67 + fn take_channel(&mut self) -> MatchboxChannel { 68 68 + self.take_channel(0).expect("Failed to get channel").split() 69 69 + } 70 70 + } 71 71 + 72 72 + impl<S: SocketImpl + 'static> MatchboxTransport<S> { 73 73 + pub async fn new(join_code: &str, is_host: bool) -> Result<Arc<Self>> { 48 74 let ws_url = server::room_url(join_code, is_host); 75 75 + let (socket, loop_fut) = S::new(&ws_url); 76 76 + Self::from_socket_and_loop_fut(socket, loop_fut).await 77 77 + } 49 78 50 50 - let (mut socket, mut loop_fut) = WebRtcSocket::new_reliable(&ws_url); 51 51 - 52 52 - let (mtx, mrx) = socket 53 53 - .take_channel(0) 54 54 - .expect("Failed to get channel") 55 55 - .split(); 79 79 + async fn from_socket_and_loop_fut( 80 80 + mut socket: S, 81 81 + mut loop_fut: MessageLoopFuture, 82 82 + ) -> Result<Arc<Self>> { 83 83 + let (itx, irx) = mpsc::channel(15); 84 84 + let (mtx, mrx) = socket.take_channel(); 56 85 57 86 let res = loop { 58 87 tokio::select! { ··· 79 108 all_peers: Mutex::new(HashSet::with_capacity(5)), 80 109 msg_sender: mtx.clone(), 81 110 cancel_token: CancellationToken::new(), 111 111 + phantom: PhantomData, 82 112 }); 83 113 84 114 tokio::spawn({ ··· 92 122 } 93 123 Err(why) => { 94 124 drop(mrx); 95 95 - mtx.close_channel(); 125 125 + drop(mtx); 96 126 drop(socket); 97 127 Err(why) 98 128 } 99 129 } 100 130 } 101 131 102 102 - async fn wait_for_id(socket: &mut WebRtcSocket) -> Option<Uuid> { 103 103 - if let Some(id) = socket.id() { 132 132 + async fn wait_for_id(socket: &mut S) -> Option<Uuid> { 133 133 + if let Some(id) = socket.get_id() { 104 134 Some(id.0) 105 135 } else { 106 136 yield_now().await; ··· 118 148 119 149 async fn main_loop( 120 150 &self, 121 121 - mut socket: WebRtcSocket, 122 122 - loop_fut: Pin<Box<dyn Future<Output = LoopFutRes> + Send + 'static>>, 151 151 + mut socket: S, 152 152 + loop_fut: MessageLoopFuture, 123 153 mut mrx: UnboundedReceiver<MatchboxMsgPair>, 124 154 ) { 125 155 tokio::pin!(loop_fut); ··· 164 194 self.push_incoming(Some(self.my_id), msg).await; 165 195 166 196 self.msg_sender.close_channel(); 197 197 + self.incoming.1.lock().await.close(); 167 198 drop(mrx); 168 168 - socket.try_update_peers().ok(); 169 199 drop(socket); 170 200 if should_await { 171 201 if let Err(why) = loop_fut.await { ··· 251 281 buffer 252 282 } 253 283 284 284 + #[cfg(test)] 285 285 + pub async fn force_recv_msg(&self) -> MsgPair { 286 286 + self.incoming 287 287 + .1 288 288 + .lock() 289 289 + .await 290 290 + .recv() 291 291 + .await 292 292 + .expect("No messages") 293 293 + } 294 294 + 295 295 + #[cfg(test)] 296 296 + pub async fn assert_no_incoming(&self) { 297 297 + assert!(self.incoming.1.lock().await.is_empty()); 298 298 + } 299 299 + 254 300 pub fn cancel(&self) { 255 301 self.cancel_token.cancel(); 256 302 } 257 303 } 258 304 259 259 - impl Transport for MatchboxTransport { 305 305 + impl<S: SocketImpl + 'static> Transport for MatchboxTransport<S> { 260 306 fn self_id(&self) -> Uuid { 261 307 self.my_id 262 308 } ··· 295 341 Self::new(code, host).await 296 342 } 297 343 } 344 344 + 345 345 + #[cfg(test)] 346 346 + mod tests { 347 347 + 348 348 + use futures::{ 349 349 + channel::{mpsc, oneshot}, 350 350 + lock::Mutex as FutMutex, 351 351 + }; 352 352 + use manhunt_logic::{GameEvent, LobbyMessage, PlayerProfile}; 353 353 + use matchbox_socket::SignalingError; 354 354 + 355 355 + use super::*; 356 356 + use tokio::test; 357 357 + 358 358 + use std::{collections::HashMap, sync::Mutex as StdMutex, time::Duration}; 359 359 + 360 360 + type PeerRx = UnboundedReceiver<PeerMsg>; 361 361 + type PeerTx = UnboundedSender<PeerMsg>; 362 362 + type IdHandle = Arc<StdMutex<Option<PeerId>>>; 363 363 + 364 364 + struct MockSocket { 365 365 + peer_recv: PeerRx, 366 366 + id: IdHandle, 367 367 + channel: Option<MatchboxChannel>, 368 368 + cancel: CancellationToken, 369 369 + } 370 370 + 371 371 + impl MockSocket { 372 372 + pub fn new( 373 373 + peer_recv: PeerRx, 374 374 + channel: MatchboxChannel, 375 375 + id: IdHandle, 376 376 + ) -> ( 377 377 + Self, 378 378 + MessageLoopFuture, 379 379 + oneshot::Sender<LoopFutRes>, 380 380 + CancellationToken, 381 381 + ) { 382 382 + let (stop_tx, stop_rx) = oneshot::channel(); 383 383 + let cancel = CancellationToken::new(); 384 384 + let sock = Self { 385 385 + peer_recv, 386 386 + channel: Some(channel), 387 387 + id, 388 388 + cancel: cancel.clone(), 389 389 + }; 390 390 + 391 391 + let fut = Box::pin(async move { stop_rx.await.expect("Failed to recv") }); 392 392 + 393 393 + (sock, fut, stop_tx, cancel) 394 394 + } 395 395 + } 396 396 + 397 397 + impl Drop for MockSocket { 398 398 + fn drop(&mut self) { 399 399 + self.cancel.cancel(); 400 400 + } 401 401 + } 402 402 + 403 403 + impl Stream for MockSocket { 404 404 + type Item = (PeerId, PeerState); 405 405 + 406 406 + fn poll_next( 407 407 + self: Pin<&mut Self>, 408 408 + cx: &mut std::task::Context<'_>, 409 409 + ) -> std::task::Poll<Option<Self::Item>> { 410 410 + let mut peer_state_rx = Pin::new(&mut self.get_mut().peer_recv); 411 411 + peer_state_rx.as_mut().poll_next(cx) 412 412 + } 413 413 + } 414 414 + 415 415 + impl SocketImpl for MockSocket { 416 416 + fn new(_room_url: &str) -> (Self, MessageLoopFuture) { 417 417 + unreachable!("Tests should use [MatchboxTransport::from_socket_and_loop_fut]") 418 418 + } 419 419 + 420 420 + fn get_id(&mut self) -> Option<PeerId> { 421 421 + *self.id.lock().unwrap() 422 422 + } 423 423 + 424 424 + fn take_channel(&mut self) -> (MatchboxSender, MatchboxReceiver) { 425 425 + self.channel.take().expect("Channel already taken") 426 426 + } 427 427 + } 428 428 + 429 429 + type MatchboxTransport = super::MatchboxTransport<MockSocket>; 430 430 + 431 431 + struct WaitingPeer { 432 432 + incoming: MatchboxSender, 433 433 + outgoing: MatchboxReceiver, 434 434 + peer_tx: PeerTx, 435 435 + intended_id: PeerId, 436 436 + id_handle: IdHandle, 437 437 + disconnect: oneshot::Sender<LoopFutRes>, 438 438 + client_cancel: CancellationToken, 439 439 + } 440 440 + 441 441 + #[derive(Default, Debug)] 442 442 + struct MockSignaling { 443 443 + peers: HashMap< 444 444 + PeerId, 445 445 + ( 446 446 + PeerTx, 447 447 + oneshot::Sender<LoopFutRes>, 448 448 + CancellationToken, 449 449 + CancellationToken, 450 450 + ), 451 451 + >, 452 452 + senders: Arc<FutMutex<HashMap<PeerId, MatchboxSender>>>, 453 453 + } 454 454 + 455 455 + impl MockSignaling { 456 456 + fn new() -> Self { 457 457 + tokio::time::pause(); 458 458 + Self::default() 459 459 + } 460 460 + 461 461 + fn client_connect( 462 462 + &self, 463 463 + id: Uuid, 464 464 + ) -> ( 465 465 + WaitingPeer, 466 466 + FutPin<Result<Arc<MatchboxTransport>, anyhow::Error>>, 467 467 + ) { 468 468 + let (itx, irx) = mpsc::unbounded(); 469 469 + let (otx, orx) = mpsc::unbounded(); 470 470 + let (peer_tx, peer_rx) = mpsc::unbounded(); 471 471 + let id_handle = Arc::new(StdMutex::new(None)); 472 472 + 473 473 + let (sock, fut, disconnect, cancel) = 474 474 + MockSocket::new(peer_rx, (otx, irx), id_handle.clone()); 475 475 + 476 476 + let transport_fut = Box::pin(MatchboxTransport::from_socket_and_loop_fut(sock, fut)); 477 477 + 478 478 + let peer = WaitingPeer { 479 479 + incoming: itx, 480 480 + outgoing: orx, 481 481 + peer_tx, 482 482 + intended_id: PeerId(id), 483 483 + id_handle, 484 484 + disconnect, 485 485 + client_cancel: cancel, 486 486 + }; 487 487 + 488 488 + (peer, transport_fut) 489 489 + } 490 490 + 491 491 + async fn broadcast_peer_join(&mut self, source: PeerId) { 492 492 + let (source_sender, _, _, _) = self.peers.get(&source).expect("Source not in peers"); 493 493 + let mut source_sender = source_sender.clone(); 494 494 + let peers = self.peers.iter_mut().filter(|(k, _)| **k != source); 495 495 + for (id, (peer_tx, _, _, _)) in peers { 496 496 + peer_tx 497 497 + .send((source, PeerState::Connected)) 498 498 + .await 499 499 + .expect("Failed to send"); 500 500 + source_sender 501 501 + .send((*id, PeerState::Connected)) 502 502 + .await 503 503 + .expect("Failed to send"); 504 504 + } 505 505 + } 506 506 + 507 507 + async fn broadcast_peer_leave(&mut self, source: PeerId) { 508 508 + let peers = self.peers.iter_mut().filter(|(k, _)| **k != source); 509 509 + for (_, (peer_tx, _, _, _)) in peers { 510 510 + peer_tx 511 511 + .send((source, PeerState::Disconnected)) 512 512 + .await 513 513 + .expect("Failed to send"); 514 514 + } 515 515 + } 516 516 + 517 517 + /// Assign an ID to a MockSocket and set it so the future resolves 518 518 + async fn assign_id(&mut self, waiting: WaitingPeer) { 519 519 + let WaitingPeer { 520 520 + id_handle, 521 521 + intended_id, 522 522 + incoming, 523 523 + mut outgoing, 524 524 + peer_tx, 525 525 + disconnect, 526 526 + client_cancel, 527 527 + } = waiting; 528 528 + 529 529 + let cancel = CancellationToken::new(); 530 530 + 531 531 + *id_handle.lock().unwrap() = Some(intended_id); 532 532 + self.peers.insert( 533 533 + intended_id, 534 534 + (peer_tx, disconnect, cancel.clone(), client_cancel), 535 535 + ); 536 536 + self.senders.lock().await.insert(intended_id, incoming); 537 537 + self.broadcast_peer_join(intended_id).await; 538 538 + 539 539 + let senders = self.senders.clone(); 540 540 + 541 541 + tokio::spawn(async move { 542 542 + let id = intended_id; 543 543 + loop { 544 544 + tokio::select! { 545 545 + biased; 546 546 + 547 547 + _ = cancel.cancelled() => { break; } 548 548 + 549 549 + Some((peer, packet)) = outgoing.next() => { 550 550 + let mut senders = senders.lock().await; 551 551 + let sender = senders.get_mut(&peer).expect("Failed to find peer"); 552 552 + sender.send((id, packet)).await.expect("Failed to send"); 553 553 + } 554 554 + } 555 555 + } 556 556 + }); 557 557 + } 558 558 + 559 559 + async fn disconnect_peer(&mut self, id: Uuid, res: LoopFutRes) { 560 560 + let (_, dc, cancel, _) = self.peers.remove(&PeerId(id)).expect("Peer not connected"); 561 561 + cancel.cancel(); 562 562 + dc.send(res).expect("Failed to send dc"); 563 563 + self.broadcast_peer_leave(PeerId(id)).await; 564 564 + } 565 565 + 566 566 + async fn wait_for_socket_drop(&self, id: Uuid) { 567 567 + let cancel = self.peers.get(&PeerId(id)).unwrap().3.clone(); 568 568 + cancel.cancelled().await; 569 569 + } 570 570 + 571 571 + async fn wait_for_client_disconnected(&mut self, id: Uuid) { 572 572 + self.wait_for_socket_drop(id).await; 573 573 + self.disconnect_peer(id, Ok(())).await; 574 574 + } 575 575 + 576 576 + async fn wait(&self) { 577 577 + tokio::time::sleep(Duration::from_millis(1)).await; 578 578 + } 579 579 + 580 580 + async fn quick_join(&mut self, id: Uuid) -> Arc<MatchboxTransport> { 581 581 + let (wait, fut) = self.client_connect(id); 582 582 + self.assign_id(wait).await; 583 583 + fut.await.expect("Transport init failed") 584 584 + } 585 585 + } 586 586 + 587 587 + const fn id(x: u128) -> Uuid { 588 588 + Uuid::from_u128(x) 589 589 + } 590 590 + 591 591 + #[test] 592 592 + async fn test_full_loop() { 593 593 + let mut sig = MockSignaling::new(); 594 594 + 595 595 + let (wait, fut) = sig.client_connect(id(1)); 596 596 + 597 597 + sig.assign_id(wait).await; 598 598 + 599 599 + let transport = fut.await.expect("Tansport failed to initialize"); 600 600 + 601 601 + assert_eq!(transport.my_id, id(1)); 602 602 + 603 603 + transport.disconnect().await; 604 604 + 605 605 + sig.wait_for_client_disconnected(id(1)).await; 606 606 + } 607 607 + 608 608 + #[test] 609 609 + async fn test_dc_pre_assign() { 610 610 + let sig = MockSignaling::new(); 611 611 + 612 612 + let (wait, fut) = sig.client_connect(id(1)); 613 613 + 614 614 + wait.disconnect.send(Ok(())).expect("Failed to send"); 615 615 + 616 616 + let res = fut.await; 617 617 + 618 618 + assert!(res.is_err()); 619 619 + assert!(wait.incoming.is_closed()); 620 620 + assert!(wait.peer_tx.is_closed()); 621 621 + assert!(wait.client_cancel.is_cancelled()); 622 622 + } 623 623 + 624 624 + #[test] 625 625 + async fn test_err_pre_assign() { 626 626 + let sig = MockSignaling::new(); 627 627 + 628 628 + let (wait, fut) = sig.client_connect(id(1)); 629 629 + 630 630 + wait.disconnect 631 631 + .send(Err(SocketError::Disconnected( 632 632 + SignalingError::UnknownFormat, 633 633 + ))) 634 634 + .expect("Failed to send"); 635 635 + 636 636 + let res = fut.await; 637 637 + 638 638 + assert!(res.is_err()); 639 639 + assert!(wait.incoming.is_closed()); 640 640 + assert!(wait.peer_tx.is_closed()); 641 641 + assert!(wait.client_cancel.is_cancelled()); 642 642 + } 643 643 + 644 644 + #[test] 645 645 + async fn test_graceful_disconnect() { 646 646 + let mut sig = MockSignaling::new(); 647 647 + 648 648 + let (wait, fut) = sig.client_connect(id(1)); 649 649 + 650 650 + let can = wait.client_cancel.clone(); 651 651 + 652 652 + sig.assign_id(wait).await; 653 653 + 654 654 + let transport = fut.await.expect("Transport init failed"); 655 655 + 656 656 + sig.disconnect_peer(id(1), Ok(())).await; 657 657 + 658 658 + let (_, disconnected) = transport 659 659 + .incoming 660 660 + .1 661 661 + .lock() 662 662 + .await 663 663 + .recv() 664 664 + .await 665 665 + .expect("Transport didnt send error"); 666 666 + 667 667 + assert!(matches!(disconnected, TransportMessage::Disconnected)); 668 668 + 669 669 + can.cancelled().await; 670 670 + 671 671 + assert!(transport.incoming.0.is_closed()); 672 672 + assert!(transport.msg_sender.is_closed()); 673 673 + } 674 674 + 675 675 + #[test] 676 676 + async fn test_error_handle() { 677 677 + let mut sig = MockSignaling::new(); 678 678 + 679 679 + let (wait, fut) = sig.client_connect(id(1)); 680 680 + 681 681 + let can = wait.client_cancel.clone(); 682 682 + 683 683 + sig.assign_id(wait).await; 684 684 + 685 685 + let transport = fut.await.expect("Transport init failed"); 686 686 + 687 687 + sig.disconnect_peer( 688 688 + id(1), 689 689 + Err(SocketError::Disconnected(SignalingError::UnknownFormat)), 690 690 + ) 691 691 + .await; 692 692 + 693 693 + let (_, disconnected) = transport 694 694 + .incoming 695 695 + .1 696 696 + .lock() 697 697 + .await 698 698 + .recv() 699 699 + .await 700 700 + .expect("Transport didnt send error"); 701 701 + 702 702 + assert!(matches!(disconnected, TransportMessage::Error(_))); 703 703 + 704 704 + // Wait for the transport to drop the socket 705 705 + can.cancelled().await; 706 706 + 707 707 + assert!(transport.incoming.0.is_closed()); 708 708 + assert!(transport.msg_sender.is_closed()); 709 709 + } 710 710 + 711 711 + #[test] 712 712 + async fn test_message_passing() { 713 713 + let mut sig = MockSignaling::new(); 714 714 + 715 715 + let t1 = sig.quick_join(id(1)).await; 716 716 + let t2 = sig.quick_join(id(2)).await; 717 717 + 718 718 + sig.wait().await; 719 719 + 720 720 + let (_, msg) = t1.force_recv_msg().await; 721 721 + let (_, msg2) = t2.force_recv_msg().await; 722 722 + 723 723 + assert_eq!(t1.all_peers.lock().await.len(), 1); 724 724 + assert_eq!(t2.all_peers.lock().await.len(), 1); 725 725 + assert!(matches!(msg, TransportMessage::PeerConnect(pid) if pid == id(2))); 726 726 + assert!(matches!(msg2, TransportMessage::PeerConnect(pid) if pid == id(1))); 727 727 + 728 728 + t1.send_transport_message(Some(id(2)), GameEvent::PlayerCaught(id(1)).into()) 729 729 + .await; 730 730 + 731 731 + sig.wait().await; 732 732 + 733 733 + let (_, msg) = t2.force_recv_msg().await; 734 734 + 735 735 + assert!( 736 736 + matches!(msg, TransportMessage::Game(ge) if matches!(*ge, GameEvent::PlayerCaught(i) if i == id(1))) 737 737 + ); 738 738 + 739 739 + t2.send_transport_message(None, LobbyMessage::PlayerSwitch(id(2), true).into()) 740 740 + .await; 741 741 + 742 742 + sig.wait().await; 743 743 + 744 744 + let (_, msg) = t1.force_recv_msg().await; 745 745 + 746 746 + assert!( 747 747 + matches!(msg, TransportMessage::Lobby(lm) if matches!(*lm, LobbyMessage::PlayerSwitch(i, b) if i == id(2) && b)) 748 748 + ); 749 749 + } 750 750 + 751 751 + #[test] 752 752 + async fn test_msg_broadcast() { 753 753 + let mut sig = MockSignaling::new(); 754 754 + 755 755 + let t1 = sig.quick_join(id(1)).await; 756 756 + let t2 = sig.quick_join(id(2)).await; 757 757 + let t3 = sig.quick_join(id(3)).await; 758 758 + let t4 = sig.quick_join(id(4)).await; 759 759 + 760 760 + sig.wait().await; 761 761 + 762 762 + let ts = [t1, t2, t3, t4]; 763 763 + 764 764 + for t in ts.iter() { 765 765 + assert_eq!(t.all_peers.lock().await.len(), ts.len() - 1); 766 766 + // Eat the PeerConnected messages 767 767 + for _ in 0..(ts.len() - 1) { 768 768 + t.force_recv_msg().await; 769 769 + } 770 770 + } 771 771 + 772 772 + ts[0] 773 773 + .send_transport_message(None, GameEvent::PlayerCaught(id(1)).into()) 774 774 + .await; 775 775 + 776 776 + sig.wait().await; 777 777 + 778 778 + ts[0].assert_no_incoming().await; 779 779 + 780 780 + for t in ts.iter().skip(1) { 781 781 + let (pid, msg) = t.force_recv_msg().await; 782 782 + assert_eq!(pid, Some(id(1))); 783 783 + assert!( 784 784 + matches!(msg, TransportMessage::Game(ge) if matches!(*ge, GameEvent::PlayerCaught(i) if i == id(1))) 785 785 + ); 786 786 + } 787 787 + } 788 788 + 789 789 + #[test] 790 790 + async fn test_direct_msg() { 791 791 + let mut sig = MockSignaling::new(); 792 792 + 793 793 + let t1 = sig.quick_join(id(1)).await; 794 794 + let t2 = sig.quick_join(id(2)).await; 795 795 + let t3 = sig.quick_join(id(3)).await; 796 796 + 797 797 + sig.wait().await; 798 798 + 799 799 + let ts = [t1, t2, t3]; 800 800 + 801 801 + for t in ts.iter() { 802 802 + assert_eq!(t.all_peers.lock().await.len(), ts.len() - 1); 803 803 + // Eat the PeerConnected messages 804 804 + for _ in 0..(ts.len() - 1) { 805 805 + t.force_recv_msg().await; 806 806 + } 807 807 + } 808 808 + 809 809 + ts[0] 810 810 + .send_transport_message(Some(id(2)), GameEvent::PlayerCaught(id(1)).into()) 811 811 + .await; 812 812 + 813 813 + sig.wait().await; 814 814 + 815 815 + ts[0].assert_no_incoming().await; 816 816 + ts[2].assert_no_incoming().await; 817 817 + 818 818 + let (pid, msg) = ts[1].force_recv_msg().await; 819 819 + assert_eq!(pid, Some(id(1))); 820 820 + assert!( 821 821 + matches!(msg, TransportMessage::Game(ge) if matches!(*ge, GameEvent::PlayerCaught(i) if i == id(1))) 822 822 + ); 823 823 + } 824 824 + 825 825 + #[test] 826 826 + async fn test_multiple_disconnect() { 827 827 + let mut sig = MockSignaling::new(); 828 828 + 829 829 + let t1 = sig.quick_join(id(1)).await; 830 830 + let t2 = sig.quick_join(id(2)).await; 831 831 + let t3 = sig.quick_join(id(3)).await; 832 832 + 833 833 + sig.wait().await; 834 834 + 835 835 + let ts = [t1, t2, t3]; 836 836 + 837 837 + for t in ts.iter() { 838 838 + assert_eq!(t.all_peers.lock().await.len(), ts.len() - 1); 839 839 + // Eat the PeerConnected messages 840 840 + for _ in 0..(ts.len() - 1) { 841 841 + t.force_recv_msg().await; 842 842 + } 843 843 + } 844 844 + 845 845 + ts[0].disconnect().await; 846 846 + 847 847 + sig.wait_for_client_disconnected(id(1)).await; 848 848 + 849 849 + sig.wait().await; 850 850 + 851 851 + for t in ts.iter().skip(1) { 852 852 + let (_, msg) = t.force_recv_msg().await; 853 853 + 854 854 + let all = t.all_peers.lock().await; 855 855 + assert!(!all.contains(&id(1))); 856 856 + assert!(matches!(msg, TransportMessage::PeerDisconnect(i) if i == id(1))); 857 857 + } 858 858 + } 859 859 + 860 860 + #[test] 861 861 + async fn test_big_message() { 862 862 + // Just a random string that's bigger than the max packet size 863 863 + let pfp = "vbnsj".repeat(65560); 864 864 + let pfp2 = pfp.clone(); 865 865 + 866 866 + let mut sig = MockSignaling::new(); 867 867 + 868 868 + let t1 = sig.quick_join(id(1)).await; 869 869 + let t2 = sig.quick_join(id(2)).await; 870 870 + 871 871 + sig.wait().await; 872 872 + 873 873 + t1.force_recv_msg().await; 874 874 + t2.force_recv_msg().await; 875 875 + 876 876 + t1.send_transport_message( 877 877 + Some(id(2)), 878 878 + LobbyMessage::PlayerSync( 879 879 + id(1), 880 880 + PlayerProfile { 881 881 + display_name: "asdf".to_string(), 882 882 + pfp_base64: Some(pfp2), 883 883 + }, 884 884 + ) 885 885 + .into(), 886 886 + ) 887 887 + .await; 888 888 + 889 889 + sig.wait().await; 890 890 + 891 891 + let (_, msg) = t2.force_recv_msg().await; 892 892 + 893 893 + if let TransportMessage::Lobby(le) = msg { 894 894 + if let LobbyMessage::PlayerSync(i, p) = *le { 895 895 + assert_eq!(i, id(1)); 896 896 + assert_eq!(p.display_name, "asdf".to_string()); 897 897 + assert_eq!(p.pfp_base64, Some(pfp)); 898 898 + } else { 899 899 + panic!("Incorrect lobby message"); 900 900 + } 901 901 + } else { 902 902 + panic!("Incorrect message"); 903 903 + } 904 904 + } 905 905 + }