Detect when a waiting player disconnects

src/bin/client.rs

@@ -1,6 +1,5 @@
 extern crate ctrlc;
-use clichess::RecvPositionError;
+use clichess::{RecvPositionError, UserRole, EXIT_MSG};
-use clichess::UserRole;
 use serde_json::json;
 use shakmaty::fen::Fen;
 use shakmaty::{Chess, Color, Outcome, Position, Setup};
@@ -83,16 +82,17 @@ fn main() {
             //it's the user turn, taking user input
             let input = read_user_input(&client);
             clichess::write_to_stream(&mut stream, String::from(input.trim())).unwrap();
-            if input.trim() == "exit" {
+            if input.trim() == EXIT_MSG {
                 break;
             }
         }
         //update position after playing.
         match get_current_position(&client) {
             Ok(position) => current_position = position,
-            Err(RecvPositionError::UserCanceledError) => break,
+            Err(_) => {
-            Err(RecvPositionError::CommunicationError) => break,
+                clichess::write_to_stream(&mut stream, String::from(EXIT_MSG)).unwrap();
-            Err(RecvPositionError::ParsePositionError) => break,
+                break;
+            }
         };
     }
     match current_position.outcome() {
@@ -141,7 +141,6 @@ fn setupctrlc() -> Arc<AtomicBool> {
 }
 
 fn setup_server_message_recv(stream: &UnixStream) -> io::Result<Receiver<String>> {
-    let buf = Arc::new(Mutex::new(String::new()));
     let (sender, receiver) = channel();
     let thread_stream = stream.try_clone()?;
 
@@ -178,7 +177,7 @@ fn read_user_input(client: &Client) -> String {
     if client.running.load(Ordering::SeqCst) {
         input
     } else {
-        String::from("exit")
+        String::from(EXIT_MSG)
     }
 }
 
@@ -238,8 +237,8 @@ fn prompt_user_for_role(client: &Client, stream: &mut UnixStream) -> Option<User
         println!("{}", prompt);
         //wait for user to give a correct answer.
         let mut input = String::from(read_user_input(client).trim());
-        if input.trim() == "exit" {
+        if input.trim() == EXIT_MSG {
-            clichess::write_to_stream(stream, String::from("exit")).unwrap();
+            clichess::write_to_stream(stream, String::from(EXIT_MSG)).unwrap();
             break;
         }
 

src/bin/server.rs

@@ -1,22 +1,23 @@
 use clichess;
-use clichess::Player;
+use clichess::{Player, RecvPositionError, UserRole, EXIT_MSG};
-use clichess::RecvPositionError;
-use clichess::UserRole;
 use serde_json::Value;
 use shakmaty::{fen, Chess, Color, Setup};
 use std::collections::HashMap;
 use std::fs;
+use std::io;
 use std::os::unix::net::{UnixListener, UnixStream};
-use std::sync::{Arc, Condvar, Mutex};
+use std::sync::mpsc::{channel, Receiver, TryRecvError};
+use std::sync::{Arc, Mutex};
 
 use std::thread;
+use std::time;
 
-#[derive(Clone)]
 struct Server {
     id: usize,
     chess_position: Arc<Mutex<Chess>>,
-    players: Arc<Mutex<HashMap<usize, (Player, Arc<(Mutex<String>, Condvar)>)>>>,
+    players: Arc<Mutex<HashMap<usize, (Player, Arc<Mutex<String>>)>>>,
-    receiving_buffer: Arc<(Mutex<String>, Condvar)>,
+    others_serv_msg_buf: Arc<Mutex<String>>,
+    client_message_recv: Receiver<String>,
 }
 
 fn main() {
@@ -31,11 +32,13 @@ fn main() {
     for stream in listener.incoming() {
         match stream {
             Ok(stream) => {
+                let client_message_recv = setup_client_message_recv(&stream).unwrap();
                 let server = Server {
                     id: counter,
                     chess_position: chess.clone(),
                     players: players.clone(),
-                    receiving_buffer: Arc::new((Mutex::new(String::new()), Condvar::new())),
+                    others_serv_msg_buf: Arc::new(Mutex::new(String::new())),
+                    client_message_recv,
                 };
                 /* connection succeeded */
                 thread::spawn(move || handle_player(stream, server));
@@ -84,14 +87,14 @@ fn main_loop(stream: &mut UnixStream, server: &Server, player: Player, mut playe
             //let go of the lock while waiting for user input.
             println!("server {}, waiting for player move..", server.id);
             let input;
-            match clichess::read_line_from_stream(stream) {
+            match server.client_message_recv.recv() {
                 Ok(i) => input = i,
                 Err(e) => {
                     println!("Error while getting user input: {}", e);
                     break;
                 }
             };
-            if input == "exit" {
+            if input == EXIT_MSG {
                 break;
             }
             {
@@ -103,43 +106,71 @@ fn main_loop(stream: &mut UnixStream, server: &Server, player: Player, mut playe
                     Err(e) => println!("Error: {}", e),
                 };
                 let chessfen = fen::fen(&*chess);
-                for (id, (_, cvar_buffer)) in players.iter() {
+                for (id, (_, others_serv_msg_buf)) in players.iter() {
                     if server.id != *id {
-                        cvar_buffer.0.lock().unwrap().push_str(&chessfen);
+                        others_serv_msg_buf.lock().unwrap().push_str(&chessfen);
-                        cvar_buffer.1.notify_one();
                     }
                 }
                 clichess::write_to_stream(stream, chessfen).unwrap();
                 player_turn = chess.turn();
             }
         } else {
-            {
+            let position_as_fen_result = wait_for_opponent_move(server);
-                let (buffer_mutex, cvar) = &*server.receiving_buffer;
+            if position_as_fen_result.is_err() {
-                let mut buffer = buffer_mutex.lock().unwrap();
+                break;
-                println!("server id: {}, waiting for move to send...", server.id);
-                while buffer.is_empty() {
-                    buffer = cvar.wait(buffer).unwrap();
-                }
-                println!("server id: {}, sending {}", server.id, buffer);
-                match clichess::write_to_stream(stream, buffer.clone()) {
-                    Ok(()) => buffer.clear(),
-                    Err(e) => {
-                        println!("{}", e);
-                        break;
-                    }
-                }
-                let chess = server.chess_position.lock().unwrap();
-                player_turn = chess.turn();
             }
+            let position_as_fen = position_as_fen_result.unwrap();
+            println!("server id: {}, sending {}", server.id, position_as_fen);
+            clichess::write_to_stream(stream, position_as_fen.clone()).unwrap();
+            let chess = server.chess_position.lock().unwrap();
+            player_turn = chess.turn();
         }
     }
 }
 
+fn wait_for_opponent_move(server: &Server) -> Result<String, RecvPositionError> {
+    println!("server id: {}, waiting for move to send...", server.id);
+    //wait: either we receive next position from other server threads, or we receive
+    //"exit" from the client.
+    let returned_result: Result<String, RecvPositionError>;
+    loop {
+        {
+            let mut others_serv_msg_buf = server.others_serv_msg_buf.lock().unwrap();
+            if !others_serv_msg_buf.is_empty() {
+                returned_result = Ok(others_serv_msg_buf.clone());
+                others_serv_msg_buf.clear();
+                break;
+            } else {
+                match server.client_message_recv.try_recv() {
+                    Ok(msg) => {
+                        if msg == EXIT_MSG {
+                            returned_result = Err(RecvPositionError::UserCanceledError);
+                            break;
+                        } else {
+                            println!(
+                                "Client sent message while it's not its turn, this is an error."
+                            );
+                            println!("Here is the message: {}", msg);
+                            continue;
+                        }
+                    }
+                    Err(TryRecvError::Disconnected) => println!("Error: client disconnected."),
+                    Err(TryRecvError::Empty) => continue,
+                }
+            }
+        }
+        thread::sleep(time::Duration::from_millis(10));
+    }
+    returned_result
+}
+
 fn create_player(server: &Server, stream: &mut UnixStream) -> Result<Player, RecvPositionError> {
     println!("Creating player {}...", server.id);
     //get player name and pubkey
-    let username_pubkey_json =
+    let username_pubkey_json = server
-        clichess::read_line_from_stream(stream).expect("Player closed connection.");
+        .client_message_recv
+        .recv()
+        .expect("Player closed connection.");
     let username_pubkey_value: Value = serde_json::from_str(&username_pubkey_json).unwrap();
     let username = username_pubkey_value["username"].to_string();
     let public_key = username_pubkey_value["pubkey"].to_string();
@@ -152,7 +183,10 @@ fn create_player(server: &Server, stream: &mut UnixStream) -> Result<Player, Rec
         username,
         public_key,
     };
-    players.insert(server.id, (player.clone(), server.receiving_buffer.clone()));
+    players.insert(
+        server.id,
+        (player.clone(), server.others_serv_msg_buf.clone()),
+    );
     println!("Created player {}", server.id);
     Ok(player)
 }
@@ -168,9 +202,11 @@ fn receive_user_role(
         println!("Computed available_roles as str: {}", available_roles);
         clichess::write_to_stream(stream, available_roles.clone()).unwrap();
         //receive chosen role
-        let chosen_role_str =
+        let chosen_role_str = server
-            clichess::read_line_from_stream(stream).expect("Player closed connection.");
+            .client_message_recv
-        if chosen_role_str == "exit" {
+            .recv()
+            .expect("Player closed connection.");
+        if chosen_role_str.trim() == EXIT_MSG {
             return Err(RecvPositionError::UserCanceledError);
         }
         println!("Client id {} has chosen {}", server.id, chosen_role_str);
@@ -197,12 +233,18 @@ fn receive_user_role(
             if available_roles_after_choice.contains(&chosen_role) {
                 println!("OK");
                 clichess::write_to_stream(stream, String::from("OK")).unwrap();
-                clichess::read_line_from_stream(stream).expect("Player closed connection.");
+                server
+                    .client_message_recv
+                    .recv()
+                    .expect("Player closed connection.");
                 break;
             } else {
                 println!("KO");
                 clichess::write_to_stream(stream, String::from("KO")).unwrap();
-                clichess::read_line_from_stream(stream).expect("Player closed connection.");
+                server
+                    .client_message_recv
+                    .recv()
+                    .expect("Player closed connection.");
                 continue;
             }
         }
@@ -228,6 +270,29 @@ fn compute_available_roles_to_str(server: &Server) -> String {
 
 fn player_disconnected(server: &Server) {
     let mut players = server.players.lock().unwrap();
+    let player_and_receiving_buf = players.get(&server.id);
+    if player_and_receiving_buf.is_some() {
+        let (player, _) = player_and_receiving_buf.expect("is some");
+        println!("Player {} disconnected.", player.username);
+    }
     players.remove(&server.id);
-    println!("Player disconnected.");
+}
+
+fn setup_client_message_recv(stream: &UnixStream) -> io::Result<Receiver<String>> {
+    let (sender, receiver) = channel();
+    let thread_stream = stream.try_clone()?;
+
+    thread::spawn(move || {
+        loop {
+            //wait for client message
+            let buffer =
+                clichess::read_line_from_stream(&thread_stream).expect("Error message from server");
+            let send_result = sender.send(buffer);
+            // stop the thread during a disconnection
+            if send_result.is_err() {
+                break;
+            }
+        }
+    });
+    Ok(receiver)
 }

src/lib.rs

@@ -10,6 +10,9 @@ use std::io::prelude::*;
 use std::io::{BufReader, Write};
 use std::os::unix::net::UnixStream;
 
+//message to send to the server to signal we disconnected.
+pub const EXIT_MSG: &str = "exit";
+
 #[derive(Clone)]
 pub struct Player {
     pub role: UserRole,
@@ -38,6 +41,7 @@ impl fmt::Display for UserRole {
     }
 }
 
+#[derive(Debug)]
 pub enum RecvPositionError {
     UserCanceledError,
     CommunicationError,