From 6e2787d792de64c003eff03b62a2df798ef851c0 Mon Sep 17 00:00:00 2001 From: salmonik Date: Thu, 29 Oct 2026 20:43:09 +0100 Subject: v2.3.0: add crash with local multiplier and pushed crash event --- src/bin/client/connection.rs | 5 ++ src/bin/client/repl.rs | 123 +++++++++++++++++++++++++++++++++++-------- src/bin/client/tls.rs | 58 +++++++++++++++++++- src/bin/server/handler.rs | 55 +++++++++++++------ src/bin/server/tls.rs | 9 ++++ 5 files changed, 211 insertions(+), 39 deletions(-) (limited to 'src/bin') 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> { + self.inner.read_available_line() + } + pub fn set_read_timeout( &self, duration: Option, 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) -> 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 = 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 = Vec::new(); let result = (|| -> std::io::Result { 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, } impl Session { @@ -202,7 +204,61 @@ impl TlsStream { &self, duration: Option, ) -> 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> { + 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 { 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::io::Result<()> { + self.stream.get_ref().set_read_timeout(duration) + } } impl Read for TlsStream { -- cgit v1.2.3