5 files changed, 211 insertions, 39 deletions

diff --git a/src/bin/client/connection.rs b/src/bin/client/connection.rs
index 6459de5..363a2a9 100644
--- a/src/bin/client/connection.rs
+++ b/src/bin/client/connection.rs
@@ -125,6 +125,11 @@ impl TlsStream {
         self.inner.write_line(msg)
     }
 
+    /// One line if the socket already has one, without blocking.
+    pub fn read_available_line(&mut self) -> std::io::Result<Option<String>> {
+        self.inner.read_available_line()
+    }
+
     pub fn set_read_timeout(
         &self,
         duration: Option<std::time::Duration>,
diff --git a/src/bin/client/repl.rs b/src/bin/client/repl.rs
index f85af47..7fa462a 100644
--- a/src/bin/client/repl.rs
+++ b/src/bin/client/repl.rs
@@ -102,29 +102,76 @@ impl ClientState {
         self.apply_prompt();
 
         loop {
-            let event = read().unwrap();
-            match event {
-                Event::Key(key) => {
-                    if key.kind != KeyEventKind::Press {
-                        continue;
-                    }
-                    let Some(key) = map_key(key) else {
-                        continue;
-                    };
-                    let effects = self.core.handle_key(key);
-                    if self.apply_effects(effects) {
-                        break;
+            let mut quit = false;
+
+            if crossterm::event::poll(std::time::Duration::from_millis(25)).unwrap_or(false) {
+                match read().unwrap() {
+                    Event::Key(key) => {
+                        if key.kind == KeyEventKind::Press {
+                            if let Some(key) = map_key(key) {
+                                let effects = self.core.handle_key(key);
+                                quit = self.apply_effects(effects);
+                                self.apply_prompt();
+                            }
+                        }
                     }
-                    self.apply_prompt();
+                    Event::Resize(_, _) => self.apply_prompt(),
+                    _ => {}
+                }
+            }
+
+            if !quit && self.core.expects_push() {
+                quit = self.pump_async();
+            }
+
+            if !quit {
+                let effects = self.core.tick();
+                if !effects.is_empty() {
+                    quit = self.apply_effects(effects);
                 }
-                Event::Resize(_, _) => self.apply_prompt(),
-                _ => {}
+            }
+
+            if quit {
+                break;
             }
         }
 
         restore_terminal(&self.cleanup_done);
     }
 
+    /// Read any server-initiated lines that are already waiting and apply them.
+    /// Returns true when an effect asked the program to quit.
+    fn pump_async(&mut self) -> bool {
+        loop {
+            let Some(mut stream) = self.stream.take() else {
+                return false;
+            };
+            let line = match stream.read_available_line() {
+                Ok(line) => {
+                    self.stream = Some(stream);
+                    line
+                }
+                Err(_) => {
+                    self.core.set_connected(false);
+                    self.stream = None;
+                    return false;
+                }
+            };
+
+            let Some(line) = line else {
+                return false;
+            };
+
+            let effects = match self.core.handle_async_line(&line) {
+                Some(effects) => effects,
+                None => self.core.handle_reply(&line),
+            };
+            if self.apply_effects(effects) {
+                return true;
+            }
+        }
+    }
+
     fn apply_effects(&mut self, effects: Vec<Effect>) -> bool {
         for effect in effects {
             match effect {
@@ -250,6 +297,9 @@ impl ClientState {
         let roulette = lines
             .first()
             .is_some_and(|first| first.contains("ROULETTE"));
+        let crash = lines
+            .first()
+            .is_some_and(|first| first.contains("CRASH"));
         let rendered: Vec<String> = lines
             .iter()
             .map(|line| {
@@ -263,6 +313,18 @@ impl ClientState {
                     } else {
                         style_roulette_numbers(line)
                     }
+                } else if crash {
+                    if line.contains("CRASHED") {
+                        line.as_str().bold().red().to_string()
+                    } else if line.starts_with("Cashed out") {
+                        line.as_str().bold().green().to_string()
+                    } else if line.starts_with("Multiplier:") {
+                        line.as_str().yellow().to_string()
+                    } else if line.starts_with("ERROR:") {
+                        line.as_str().red().to_string()
+                    } else {
+                        line.clone()
+                    }
                 } else if line.contains('{') {
                     style_winning_cells(line)
                 } else if line.contains("WIN!") || line.starts_with("Won ") {
@@ -361,22 +423,37 @@ impl ClientState {
             return "ERROR Not connected".to_string();
         };
 
+        // A crash event can be pushed while a request is in flight. Route those
+        // lines as they arrive and keep waiting for the actual reply.
+        let mut pushed: Vec<String> = Vec::new();
         let result = (|| -> std::io::Result<String> {
             stream.write_line(line)?;
-            let mut response = String::new();
-            let read = stream.read_line(&mut response)?;
-            if read == 0 {
-                return Err(std::io::Error::new(
-                    std::io::ErrorKind::UnexpectedEof,
-                    "connection closed",
-                ));
+            loop {
+                let mut response = String::new();
+                let read = stream.read_line(&mut response)?;
+                if read == 0 {
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::UnexpectedEof,
+                        "connection closed",
+                    ));
+                }
+                let response = response.trim_end().to_string();
+                if self.core.is_async(&response) {
+                    pushed.push(response);
+                    continue;
+                }
+                return Ok(response);
             }
-            Ok(response)
         })();
 
         match result {
             Ok(response) => {
                 self.stream = Some(stream);
+                for line in pushed {
+                    if let Some(effects) = self.core.handle_async_line(&line) {
+                        self.apply_effects(effects);
+                    }
+                }
                 response
             }
             Err(error) => {
diff --git a/src/bin/client/tls.rs b/src/bin/client/tls.rs
index 8c052f7..b46f52b 100644
--- a/src/bin/client/tls.rs
+++ b/src/bin/client/tls.rs
@@ -64,6 +64,7 @@ impl TlsConnector {
                 sock,
                 tls: conn,
                 lines: LineBuffer::new(),
+                read_timeout: Some(std::time::Duration::from_secs(10)),
             })),
         })
     }
@@ -150,6 +151,7 @@ struct Session {
     sock: TcpStream,
     tls: rustls::ClientConnection,
     lines: LineBuffer,
+    read_timeout: Option<std::time::Duration>,
 }
 
 impl Session {
@@ -202,7 +204,61 @@ impl TlsStream {
         &self,
         duration: Option<std::time::Duration>,
     ) -> std::io::Result<()> {
-        self.session.lock().sock.set_read_timeout(duration)
+        let mut s = self.session.lock();
+        s.read_timeout = duration;
+        s.sock.set_read_timeout(duration)
+    }
+
+    /// Return one complete line if the socket already has one, without
+    /// blocking. Used to pick up server pushes between requests.
+    pub fn read_available_line(&mut self) -> std::io::Result<Option<String>> {
+        let mut s = self.session.lock();
+        if let Some(line) = s.lines.next_line() {
+            return Ok(Some(line));
+        }
+
+        // `set_read_timeout(Some(0))` is rejected by std, so poll with the
+        // shortest timeout the socket accepts instead of a nonblocking mode.
+        s.sock
+            .set_read_timeout(Some(std::time::Duration::from_millis(1)))?;
+        let drained = (|| -> std::io::Result<()> {
+            loop {
+                let mut buf = [0u8; 4096];
+                match s.tls.reader().read(&mut buf) {
+                    Ok(0) => return Ok(()),
+                    Ok(n) => s.lines.push(&buf[..n]),
+                    // The reader only drains plaintext rustls has already
+                    // decrypted; `drive` moves bytes off the socket.
+                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                        match s.drive() {
+                            Ok(()) => continue,
+                            Err(error)
+                                if matches!(
+                                    error.kind(),
+                                    std::io::ErrorKind::WouldBlock
+                                        | std::io::ErrorKind::TimedOut
+                                ) =>
+                            {
+                                return Ok(());
+                            }
+                            Err(error) => return Err(error),
+                        }
+                    }
+                    Err(error)
+                        if matches!(
+                            error.kind(),
+                            std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
+                        ) =>
+                    {
+                        return Ok(());
+                    }
+                    Err(error) => return Err(error),
+                }
+            }
+        })();
+        s.sock.set_read_timeout(s.read_timeout)?;
+        drained?;
+        Ok(s.lines.next_line())
     }
 }
 
diff --git a/src/bin/server/handler.rs b/src/bin/server/handler.rs
index 86b09cb..dffae69 100644
--- a/src/bin/server/handler.rs
+++ b/src/bin/server/handler.rs
@@ -1,6 +1,11 @@
+use std::time::Duration;
+
 use crate::tls::TlsStream;
 use zagan_bet::server::core::{ConnectionState, ServerCore};
 
+/// How long the connection waits for a request before advancing timed games.
+const POLL_INTERVAL: Duration = Duration::from_millis(50);
+
 pub fn handle_client_tls(mut stream: TlsStream) {
     let client_addr = match stream.peer_addr() {
         Ok(addr) => addr.to_string(),
@@ -19,34 +24,54 @@ pub fn handle_client_tls(mut stream: TlsStream) {
         }
     };
 
+    if let Err(error) = stream.set_read_timeout(Some(POLL_INTERVAL)) {
+        eprintln!("Could not set read timeout: {error}");
+        return;
+    }
+
     let mut state = ConnectionState::default();
     println!("[TLS Client connected] {client_addr}");
 
-    loop {
+    'connection: loop {
         let mut line = String::new();
         match stream.read_line(&mut line) {
             Ok(0) => break,
-            Ok(_) => {}
+            Ok(_) => {
+                let line = line.trim();
+                if !line.is_empty() {
+                    let user = state.user().unwrap_or("guest").to_string();
+                    println!("[{user} @ {client_addr}] Received: {line}");
+                    let response = core.handle_message(&mut state, line);
+                    if let Err(error) = stream.write_line(&response) {
+                        eprintln!("Write error: {error}");
+                        break;
+                    }
+                    let user = state.user().unwrap_or("guest");
+                    println!("[{user} @ {client_addr}] Sent: {response}");
+                }
+            }
+            Err(error)
+                if matches!(
+                    error.kind(),
+                    std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
+                ) =>
+            {
+                // No request yet: fall through and let timed games advance.
+            }
             Err(error) => {
                 eprintln!("Read error: {error}");
                 break;
             }
         }
 
-        let line = line.trim();
-        if line.is_empty() {
-            continue;
-        }
-
-        let user = state.user().unwrap_or("guest").to_string();
-        println!("[{user} @ {client_addr}] Received: {line}");
-        let response = core.handle_message(&mut state, line);
-        if let Err(error) = stream.write_line(&response) {
-            eprintln!("Write error: {error}");
-            break;
+        for push in core.poll(&state) {
+            if let Err(error) = stream.write_line(&push) {
+                eprintln!("Write error: {error}");
+                break 'connection;
+            }
+            let user = state.user().unwrap_or("guest");
+            println!("[{user} @ {client_addr}] Pushed: {push}");
         }
-        let user = state.user().unwrap_or("guest");
-        println!("[{user} @ {client_addr}] Sent: {response}");
     }
 
     core.disconnect(&state);
diff --git a/src/bin/server/tls.rs b/src/bin/server/tls.rs
index 292ebf7..5d84589 100644
--- a/src/bin/server/tls.rs
+++ b/src/bin/server/tls.rs
@@ -77,6 +77,15 @@ impl TlsStream {
     pub fn peer_addr(&self) -> std::io::Result<std::net::SocketAddr> {
         self.stream.get_ref().peer_addr()
     }
+
+    /// Read timeout for the socket. The connection loop uses a short one so it
+    /// can advance crash rounds while no request is in flight.
+    pub fn set_read_timeout(
+        &self,
+        duration: Option<std::time::Duration>,
+    ) -> std::io::Result<()> {
+        self.stream.get_ref().set_read_timeout(duration)
+    }
 }
 
 impl Read for TlsStream {