Manage client disconnect case where he is waiting

src/bin/client.rs

@@ -9,11 +9,18 @@ use std::sync::{Arc, Mutex};
 use std::thread;
 use std::time;
 
+enum RecvPositionError {
+    UserCanceledError,
+    CommunicationError,
+    ParsePositionError,
+}
+
 fn main() {
     let running = setupctrlc();
     let username = std::env::args().nth(1).expect("no name given");
     let public_key = std::env::args().nth(2).expect("no public key given");
     println!("Name: {}, Public key: {}", username, public_key);
+    //send username and public key to server
     let mut stream = match UnixStream::connect("/tmp/clichess.socket") {
         Ok(sock) => sock,
         Err(_) => {
@@ -21,9 +28,8 @@ fn main() {
             return;
         }
     };
-    //send username and public key to server
+    clichess::write_to_stream(&mut stream, username).unwrap();
-    clichess::write_to_stream(&mut stream, username);
+    clichess::write_to_stream(&mut stream, public_key).unwrap();
-    clichess::write_to_stream(&mut stream, public_key);
     let (client, chess) = get_connection_info_from_stream(&stream);
     //First prompt when connecting to the server
     println!(
@@ -34,6 +40,8 @@ fn main() {
     //then we get the initial role of the connected client.
     let mut current_position = chess;
     let input_buffer = setup_input_buffer();
+    //start a thread to listen to server messages
+    let server_message = setup_server_message_recv(&stream).unwrap();
     loop {
         println!(
             "{}",
@@ -46,14 +54,18 @@ fn main() {
         if clichess::is_player_turn(&client, current_position.turn()) {
             //it's the user turn, taking user input
             let input = read_user_input(running.clone(), input_buffer.clone());
-            println!("trying to play {}", input);
+            clichess::write_to_stream(&mut stream, String::from(input.trim())).unwrap();
-            clichess::write_to_stream(&mut stream, String::from(input.trim()));
             if input.trim() == "exit" {
                 break;
             }
         }
         //update position after playing.
-        current_position = get_current_position(&stream);
+        match get_current_position(running.clone(), server_message.clone()) {
+            Ok(position) => current_position = position,
+            Err(RecvPositionError::UserCanceledError) => break,
+            Err(RecvPositionError::CommunicationError) => break,
+            Err(RecvPositionError::ParsePositionError) => break,
+        };
     }
     match current_position.outcome() {
         None => println!("Bye"),
@@ -100,7 +112,35 @@ fn setupctrlc() -> Arc<AtomicBool> {
     running
 }
 
+fn setup_server_message_recv(stream: &UnixStream) -> io::Result<Arc<Mutex<String>>> {
+    let buf = Arc::new(Mutex::new(String::new()));
+    let buf2 = buf.clone();
+    let thread_stream = stream.try_clone()?;
+
+    thread::spawn(move || {
+        loop {
+            //wait for server message
+            let buffer =
+                clichess::read_line_from_stream(&thread_stream).expect("Error message from server");
+            {
+                let mut server_message = buf2.lock().unwrap();
+                if server_message.is_empty() {
+                    *server_message = buffer;
+                } else {
+                    println!("Warning: server tried to send a message while the current one has not been computed yet.");
+                }
+            }
+        }
+    });
+    Ok(buf)
+}
+
 fn read_user_input(running: Arc<AtomicBool>, input_buffer: Arc<Mutex<String>>) -> String {
+    //clear input before waiting for a new one
+    {
+        let mut user_input = input_buffer.lock().unwrap();
+        user_input.clear();
+    }
     let mut input = String::new();
     while running.load(Ordering::SeqCst) {
         thread::sleep(time::Duration::from_millis(10));
@@ -122,9 +162,29 @@ fn read_user_input(running: Arc<AtomicBool>, input_buffer: Arc<Mutex<String>>) -
 }
 
 //wait for next position from server, then return the current board.
-fn get_current_position(stream: &UnixStream) -> Chess {
+fn get_current_position(
-    let response = clichess::read_line_from_stream(&stream).expect("Server disconnected.");
+    running: Arc<AtomicBool>,
-    parse_position(&response)
+    server_message: Arc<Mutex<String>>,
+) -> Result<Chess, RecvPositionError> {
+    let mut response = String::new();
+    while running.load(Ordering::SeqCst) {
+        thread::sleep(time::Duration::from_millis(10));
+        {
+            let mut server_response = server_message.lock().unwrap();
+            if server_response.is_empty() {
+                continue;
+            } else {
+                response = server_response.clone();
+                server_response.clear();
+                break;
+            }
+        }
+    }
+    if response.is_empty() {
+        Err(RecvPositionError::UserCanceledError)
+    } else {
+        Ok(parse_position(&response))
+    }
 }
 
 fn parse_position(string: &str) -> Chess {

src/bin/server.rs

@@ -67,7 +67,7 @@ fn handle_client(mut stream: UnixStream, server: Server) {
     {
         let chess = server.chess_position.lock().unwrap();
         player_turn = chess.turn();
-        clichess::write_to_stream(&mut stream, fen::fen(&*chess));
+        clichess::write_to_stream(&mut stream, fen::fen(&*chess)).unwrap();
     }
     println!("server {}, current position to the client sent", server.id);
     loop {
@@ -100,7 +100,7 @@ fn handle_client(mut stream: UnixStream, server: Server) {
                         cvar_buffer.1.notify_one();
                     }
                 }
-                clichess::write_to_stream(&mut stream, chessfen);
+                clichess::write_to_stream(&mut stream, chessfen).unwrap();
                 player_turn = chess.turn();
             }
         } else {
@@ -112,8 +112,13 @@ fn handle_client(mut stream: UnixStream, server: Server) {
                     buffer = cvar.wait(buffer).unwrap();
                 }
                 println!("server id: {}, sending {}", server.id, buffer);
-                clichess::write_to_stream(&mut stream, buffer.clone());
+                match clichess::write_to_stream(&mut stream, buffer.clone()) {
-                buffer.clear();
+                    Ok(()) => buffer.clear(),
+                    Err(e) => {
+                        println!("{}", e);
+                        break;
+                    }
+                }
                 let chess = server.chess_position.lock().unwrap();
                 player_turn = chess.turn();
             }

src/lib.rs

@@ -4,8 +4,8 @@ use shakmaty::san::ParseSanError;
 use shakmaty::san::San;
 use shakmaty::san::SanError;
 use shakmaty::{Chess, Color, IllegalMoveError, Position, Setup, Square};
-use std::io;
 use std::fmt;
+use std::io;
 use std::io::prelude::*;
 use std::io::{BufReader, Write};
 use std::os::unix::net::UnixStream;
@@ -195,8 +195,9 @@ pub fn read_line_from_stream(stream: &UnixStream) -> Result<String, io::Error> {
     Ok(String::from(result.trim_end()))
 }
 
-pub fn write_to_stream(stream: &mut UnixStream, msg: String) {
+pub fn write_to_stream(stream: &mut UnixStream, msg: String) -> io::Result<()> {
-    stream.write_all(&(msg + "\n").as_bytes()).unwrap();
+    stream.write_all(&(msg + "\n").as_bytes())?;
+    Ok(())
 }
 
 pub fn get_default_side(role: UserRole) -> Color {