3 files changed, 162 insertions, 24 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())
     }
 }