fix: Make join work on UDP package loss

- Client retries the handshake operation until receives response from
  server or times out.
- Make server and client work on all network interfaces instead of just localhost

Author: balintkissdev

src/app.rs

@@ -76,9 +76,6 @@ impl<'a> App<'a> {
     }
 
     fn run(&mut self, event_loop: &mut EventLoop<()>) {
-        const MAX_LOGIC_UPDATE_PER_SECOND: f32 = 60.0;
-        const FIXED_UPDATE_TIMESTEP: f32 = 1.0 / MAX_LOGIC_UPDATE_PER_SECOND;
-
         let mut previous_time = std::time::Instant::now();
         let mut lag: f32 = 0.0;
         loop {
@@ -95,9 +92,9 @@ impl<'a> App<'a> {
                 self.process_server_response();
             }
 
-            while lag >= FIXED_UPDATE_TIMESTEP {
+            while lag >= globals::FIXED_UPDATE_TIMESTEP_SEC {
                 self.update();
-                lag -= FIXED_UPDATE_TIMESTEP;
+                lag -= globals::FIXED_UPDATE_TIMESTEP_SEC;
             }
 
             self.window.as_ref().unwrap().request_redraw();
@@ -116,20 +113,22 @@ impl<'a> App<'a> {
             .client_session
             .as_mut()
             .unwrap()
-            .receive_server_resposne()
+            .receive_server_response()
         {
             message::trace(format!("Received: {}", msg));
             match Message::deserialize(&msg) {
                 Ok(Message::Replicate(new_player)) => {
-                    self.remote_players.insert(new_player.id, new_player);
-                    self.gui
-                        .as_mut()
-                        .unwrap()
-                        .log(format!("Player {} has joined the server", new_player.id));
-                }
-                Ok(Message::Position(id, pos)) => {
-                    if let Some(player) = self.remote_players.get_mut(&id) {
-                        player.pos = pos;
+                    if let Some(player) = self.remote_players.get_mut(&new_player.id) {
+                        // Update existing player based on server's simulation
+                        player.pos = new_player.pos;
+                    } else {
+                        // On-demand remote player creation because replication does not
+                        // fit into the handshake ACK message.
+                        self.remote_players.insert(new_player.id, new_player);
+                        self.gui
+                            .as_mut()
+                            .unwrap()
+                            .log(format!("Player {} has joined the server", new_player.id));
                     }
                 }
                 Ok(Message::Leave(id)) => {
@@ -188,7 +187,7 @@ impl<'a> App<'a> {
                             }
                         }
                     }
-                    Some(_) => (), // Task is still running
+                    Some(_) => (), // Task is still running, nothing to do
                     None => {
                         // Fire task if not exists
                         let server_address = server_address.clone();

src/client.rs

@@ -28,8 +28,8 @@ pub type ClientSessionResult = Result<ClientSession, Box<dyn Error + Send + Sync
 
 impl ClientSession {
     pub async fn new(server_address: String) -> ClientSessionResult {
-        match tokio::time::timeout(globals::CONNECTION_TIMEOUT_DURATION, async {
-            let client_socket = UdpSocket::bind("127.0.0.1:0").await?;
+        match tokio::time::timeout(globals::CONNECTION_TIMEOUT_SEC, async {
+            let client_socket = UdpSocket::bind("0.0.0.0:0").await?;
             let client_socket = Arc::new(client_socket);
 
             let session_player = join_server(&client_socket, &server_address).await?;
@@ -58,7 +58,7 @@ impl ClientSession {
             Ok(client_session) => client_session,
             Err(_) => Err(format!(
                 "Connection timed out after {} seconds.",
-                globals::CONNECTION_TIMEOUT_DURATION.as_secs()
+                globals::CONNECTION_TIMEOUT_SEC.as_secs()
             )
             .into()),
         }
@@ -68,7 +68,7 @@ impl ClientSession {
         self.session_player
     }
 
-    pub fn receive_server_resposne(&mut self) -> Result<String, TryRecvError> {
+    pub fn receive_server_response(&mut self) -> Result<String, TryRecvError> {
         match self.listen_rx.try_recv() {
             Ok(response) => {
                 if let Ok(Message::Ping) = Message::deserialize(&response) {
@@ -87,7 +87,7 @@ impl ClientSession {
     }
 
     pub fn is_server_alive(&self) -> bool {
-        self.last_ping.elapsed() < globals::CONNECTION_TIMEOUT_DURATION
+        self.last_ping.elapsed() < globals::CONNECTION_TIMEOUT_SEC
     }
 
     pub fn leave_server(&self, player_id: PlayerID) {
@@ -108,33 +108,53 @@ async fn join_server(
     client_socket: &UdpSocket,
     server_address: &String,
 ) -> Result<Player, Box<dyn Error + Send + Sync>> {
-    let msg = Message::Handshake.serialize();
-    client_socket
-        .send_to(msg.as_bytes(), server_address)
-        .await?;
-    message::trace(format!("Sent: {msg}"));
-
-    let mut ack_buf = [0u8; 32];
-    let (len, _) = client_socket.recv_from(&mut ack_buf).await?;
-    let response = String::from_utf8_lossy(&ack_buf[..len]).to_string();
-    message::trace(format!("Handshake result: {response}"));
-
-    let result = match Message::deserialize(&response) {
-        Ok(Message::Ack(new_id, new_color)) => Ok(Player::new(new_id, new_color)),
-        Ok(_) => Err("Invalid handshake received".into()),
-        Err(e) => Err(format!("Handshake failed: {e}").into()),
-    };
-    result
+    let handshake_msg = Message::Handshake.serialize();
+    // Loop abort happens on timeout in ClientSession::new()
+    loop {
+        client_socket
+            .send_to(handshake_msg.as_bytes(), server_address)
+            .await?;
+        message::trace(format!("Sent: {handshake_msg}"));
+
+        match receive_with_retry_timeout(client_socket).await {
+            Ok(response) => {
+                if let Ok(Message::Ack(new_id, new_color)) = Message::deserialize(&response) {
+                    message::trace(format!("Handshake result: {response}"));
+                    return Ok(Player::new(new_id, new_color));
+                }
+
+                message::trace(format!("Invalid handshake response: {response}"));
+            }
+            _ => continue, // Keep trying, I know you can do it!
+        }
+    }
+}
+
+async fn receive_with_retry_timeout(
+    socket: &UdpSocket,
+) -> Result<String, Box<dyn Error + Send + Sync>> {
+    let retry_timeout = std::time::Duration::from_millis(300);
+    let mut buf = [0u8; 32];
+    match tokio::time::timeout(retry_timeout, socket.recv_from(&mut buf)).await {
+        Ok(result) => {
+            let (len, _) = result?;
+            Ok(String::from_utf8_lossy(&buf[..len]).to_string())
+        }
+        Err(_) => {
+            message::trace("No response (sender or receiver package lost)".to_string());
+            Err("Receive operation timed out".into())
+        }
+    }
 }
 
-async fn listen_handler(socket: Arc<UdpSocket>, tx: ChannelSender) {
+async fn listen_handler(socket: Arc<UdpSocket>, listen_tx: ChannelSender) {
     let mut buf = [0u8; 1024];
     loop {
         match socket.recv_from(&mut buf).await {
             Ok((len, _)) => {
                 if let Ok(msg) = std::str::from_utf8(&buf[..len]) {
                     // Pass message to main thread
-                    if tx.send(msg.to_string()).is_err() {
+                    if listen_tx.send(msg.to_string()).is_err() {
                         break;
                     }
                 }

src/lib.rs

@@ -53,12 +53,15 @@ pub mod globals {
 
     pub const LOCALHOST: &str = "127.0.0.1";
     pub const DEFAULT_PORT: u16 = 8080;
-    pub const PING_INTERVAL: std::time::Duration = std::time::Duration::from_secs(2);
-    pub const CONNECTION_TIMEOUT_DURATION: std::time::Duration = std::time::Duration::from_secs(5);
+    pub const PING_INTERVAL_MS: std::time::Duration = std::time::Duration::from_millis(20);
+    pub const CONNECTION_TIMEOUT_SEC: std::time::Duration = std::time::Duration::from_secs(5);
 
     pub const WINDOW_SIZE: (u16, u16) = (800, 600);
     pub const WINDOW_TITLE: &str = "Multiplayer game demo by Bálint Kiss";
 
+    pub const MAX_LOGIC_UPDATE_PER_SEC: f32 = 60.0;
+    pub const FIXED_UPDATE_TIMESTEP_SEC: f32 = 1.0 / MAX_LOGIC_UPDATE_PER_SEC;
+
     pub const PLAYER_QUAD_SIZE: f32 = 24.0;
     pub const WORLD_BOUNDS: WorldBounds = WorldBounds {
         min_x: -1200.0,

src/message.rs

@@ -11,14 +11,16 @@ pub enum Message {
     Ack(PlayerID, Vector3<f32>),
     Leave(PlayerID),
     Replicate(Player),
+    // TODO: Avoid clients self-reporting their exact own position and opt for sending input state
+    // instead
     Position(PlayerID, Vector2<f32>),
 }
 
 const PING: &str = "PING";
 const HANDSHAKE: &str = "HANDSHAKE";
 const ACK: &str = "ACK";
 const LEAVE: &str = "LEAVE";
-const PUSH: &str = "PUSH";
+const REPL: &str = "REPL";
 const POS: &str = "POS";
 
 impl Message {
@@ -68,7 +70,7 @@ impl Message {
                 })?;
                 Ok(Message::Leave(player_id))
             }
-            Some(PUSH) if parts.len() == 3 => {
+            Some(REPL) if parts.len() == 3 => {
                 let player_id = parts[1].parse().map_err(|_| {
                     std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid PlayerID")
                 })?;
@@ -126,7 +128,7 @@ impl Message {
             Message::Handshake => HANDSHAKE,
             Message::Ack(_, _) => ACK,
             Message::Leave(_) => LEAVE,
-            Message::Replicate(_) => PUSH,
+            Message::Replicate(_) => REPL,
             Message::Position(_, _) => POS,
         }
     }