From ba0218f88569f7cf300f30d31b2c0ac8972b5efd Mon Sep 17 00:00:00 2001 From: Joshua Higgins Date: Sun, 15 Mar 2026 04:36:39 +0000 Subject: [PATCH 1/4] misc: rework observers --- src/main.rs | 9 ++-- src/server.rs | 121 ++++++++++++++++++-------------------------------- src/types.rs | 2 - 3 files changed, 48 insertions(+), 84 deletions(-) diff --git a/src/main.rs b/src/main.rs index ae5bb6a..7d67bb6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -51,9 +51,6 @@ async fn handle_connection( let (mut ws_sender, mut ws_receiver) = ws_stream.split(); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); - // Store the client - sd.observers.write().await.insert(addr, tx.clone()); - // Spawn task to handle outgoing messages let send_task = tokio::spawn(async move { while let Some(msg) = rx.recv().await { @@ -103,6 +100,12 @@ async fn handle_connection( let _ = send(&tx, e.to_string().as_str()); } } + "OBSERVE" => { + if let Err(e) = sd.handle_observe(addr, tx.clone()).await { + error!("handle_observe: {}", e); + let _ = send(&tx, e.to_string().as_str()); + } + } "READY" => { if let Err(e) = sd.handle_ready(addr, tx.clone()).await { error!("handle_ready: {}", e); diff --git a/src/server.rs b/src/server.rs index f4384b7..aa0108f 100644 --- a/src/server.rs +++ b/src/server.rs @@ -80,8 +80,6 @@ impl Server { drop(clients_guard); - self.remove_observer_from_all_matches(addr).await; - self.observers.write().await.remove(&addr); self.usernames.write().await.insert(requested_username.clone(), addr); let _ = send(&tx, "CONNECT:ACK"); @@ -246,12 +244,27 @@ impl Server { } self.clients.write().await.remove(&addr); - self.observers.write().await.insert(addr, tx.clone()); let _ = send(&tx, "DISCONNECT:ACK"); Ok(()) } + pub async fn handle_observe( + &self, + addr: SocketAddr, + tx: UnboundedSender, + ) -> Result<(), anyhow::Error> { + let mut observers_guard = self.observers.write().await; + if observers_guard.remove(&addr).is_some() { + let _ = send(&tx, "OBSERVE:ACK:0"); + } else { + observers_guard.insert(addr, tx.clone()); + let _ = send(&tx, "OBSERVE:ACK:1"); + } + + Ok(()) + } + pub async fn handle_ready( &self, addr: SocketAddr, @@ -403,7 +416,6 @@ impl Server { if invalid { let current_match_id = current_match.id; let is_demo_mode = current_match.demo_mode; - let viewers = current_match.viewers.clone(); drop(current_match); drop(matches_guard); @@ -419,7 +431,8 @@ impl Server { let _ = send(&tx, "GAME:LOSS"); let _ = send(&opponent.connection, "GAME:WINS"); - self.broadcast_message(&viewers, &format!("GAME:WIN:{}", opponent.username)).await; + self.broadcast_message_all_observers(&format!("GAME:WIN:{}", opponent.username)) + .await; opponent.current_match = None; opponent.color = Color::None; @@ -448,10 +461,8 @@ impl Server { timeout_thread.abort(); } - let mut viewer_messages = Vec::new(); - let viewers = current_match.viewers.clone(); - - viewer_messages.push(format!("GAME:MOVE:{}:{}", client.username, column)); + let mut observer_messages = Vec::new(); + observer_messages.push(format!("GAME:MOVE:{}:{}", client.username, column)); // Check game end conditions let (winner, filled) = current_match.end_game_check(); @@ -464,7 +475,7 @@ impl Server { let opponent = opponent.read().await; let _ = send(&opponent.connection, "GAME:LOSS"); } - viewer_messages.push(format!("GAME:WIN:{}", client.username)); + observer_messages.push(format!("GAME:WIN:{}", client.username)); } else if filled { let _ = send(&tx, "GAME:DRAW"); if !current_match.demo_mode { @@ -472,7 +483,7 @@ impl Server { let opponent = opponent.read().await; let _ = send(&opponent.connection, "GAME:DRAW"); } - viewer_messages.push("GAME:DRAW".to_string()); + observer_messages.push("GAME:DRAW".to_string()); } // remove match from matchmaker @@ -564,19 +575,21 @@ impl Server { ); } - for msg in viewer_messages { - broadcast_message(&observers, &viewers, &msg).await; + for msg in observer_messages { + let observers_guard = observers.read().await; + for (_, tx) in observers_guard.iter() { + let _ = send(tx, &msg); + } } if demo_mode && no_winner { tokio::time::sleep(tokio::time::Duration::from_millis(default_waiting_time)).await; let _ = send(&client_tx, &format!("OPPONENT:{}", demo_move)); - broadcast_message( - &observers, - &viewers, - &format!("GAME:MOVE:{}:{}", SERVER_PLAYER_USERNAME, demo_move), - ) - .await; + let observers_guard = observers.read().await; + let msg = format!("GAME:MOVE:{}:{}", SERVER_PLAYER_USERNAME, demo_move); + for (_, tx) in observers_guard.iter() { + let _ = send(tx, &msg); + } } })); @@ -590,7 +603,6 @@ impl Server { let client_tx = tx.clone(); let client_addr = addr.clone(); let observers = self.observers.clone(); - let viewers = current_match.viewers.clone(); let opponent_move = opponent.clone(); current_match.timeout_thread = Some(tokio::spawn(async move { if demo_mode { @@ -610,12 +622,11 @@ impl Server { let opponent = opponent.read().await; let _ = send(&opponent.connection, "GAME:LOSS"); drop(opponent); - broadcast_message( - &observers, - &viewers, - &format!("GAME:WIN:{}", client_username), - ) - .await; + let observers_guard = observers.read().await; + let msg = format!("GAME:WIN:{}", client_username); + for (_, tx) in observers_guard.iter() { + let _ = send(tx, &msg); + } let mut clients_guard = clients.write().await; let mut client = clients_guard.get_mut(&client_addr).unwrap().write().await; @@ -709,10 +720,9 @@ impl Server { &self, tx: UnboundedSender, match_id: u32, - addr: SocketAddr, + _addr: SocketAddr, ) -> Result<(), anyhow::Error> { - let result = self.watch(match_id, addr).await; - if result.is_err() { + if self.matches.read().await.get(&match_id).is_none() { return Err(anyhow::anyhow!("ERROR:INVALID:WATCH")); } @@ -864,7 +874,8 @@ impl Server { timeout_thread.abort(); } - self.broadcast_message(&the_match.viewers, &format!("GAME:WIN:{}", winner_username)).await; + self.broadcast_message_all_observers(&format!("GAME:WIN:{}", winner_username)) + .await; let clients_guard = self.clients.read().await; if the_match.demo_mode { @@ -1187,54 +1198,6 @@ impl Server { Ok(()) } - pub async fn watch(&self, new_match_id: u32, addr: SocketAddr) -> Result<(), String> { - let matches_guard = self.matches.read().await; - - for match_guard in matches_guard.values() { - let mut found = false; - let mut a_match = match_guard.write().await; - for i in 0..a_match.viewers.len() { - if a_match.viewers[i] == addr { - a_match.viewers.remove(i); - found = true; - break; - } - } - - if found { - break; - } - } - - let result = matches_guard.get(&new_match_id); - if result.is_none() { - return Err("Match not found".to_string()); - } - result.unwrap().write().await.viewers.push(addr); - - Ok(()) - } - - pub async fn remove_observer_from_all_matches(&self, addr: SocketAddr) { - let matches_guard = self.matches.read().await; - - for match_guard in matches_guard.values() { - let mut found = false; - let mut a_match = match_guard.write().await; - for i in 0..a_match.viewers.len() { - if a_match.viewers[i] == addr { - a_match.viewers.remove(i); - found = true; - break; - } - } - - if found { - break; - } - } - } - pub async fn terminate_match(&self, match_id: u32) { let matches_guard = self.matches.read().await; let the_match = matches_guard.get(&match_id); @@ -1254,7 +1217,7 @@ impl Server { timeout_thread.abort(); } - self.broadcast_message(&the_match.viewers, "GAME:TERMINATED").await; + self.broadcast_message_all_observers("GAME:TERMINATED").await; let clients_guard = self.clients.read().await; if the_match.player1 != SERVER_PLAYER_ADDR.to_string().parse().unwrap() { diff --git a/src/types.rs b/src/types.rs index fef36f7..c0057e7 100644 --- a/src/types.rs +++ b/src/types.rs @@ -61,7 +61,6 @@ pub struct Match { pub id: u32, pub demo_mode: bool, pub board: Vec>, - pub viewers: Vec, pub ledger: Vec<(Color, usize, Instant)>, pub wait_thread: Option>, pub timeout_thread: Option>, @@ -81,7 +80,6 @@ impl Match { id, demo_mode, board: vec![vec![Color::None; 6]; 7], - viewers: Vec::new(), ledger: Vec::new(), wait_thread: None, timeout_thread: None, -- 2.49.1 From adc013db182c18b4e5ec6a4a707c4adbb1ea2d46 Mon Sep 17 00:00:00 2001 From: Joshua Higgins Date: Sun, 15 Mar 2026 04:41:34 +0000 Subject: [PATCH 2/4] include match_ids --- src/server.rs | 29 ++++++++++++++++++++--------- 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/src/server.rs b/src/server.rs index aa0108f..3254be8 100644 --- a/src/server.rs +++ b/src/server.rs @@ -431,8 +431,11 @@ impl Server { let _ = send(&tx, "GAME:LOSS"); let _ = send(&opponent.connection, "GAME:WINS"); - self.broadcast_message_all_observers(&format!("GAME:WIN:{}", opponent.username)) - .await; + self.broadcast_message_all_observers(&format!( + "GAME:{}:WIN:{}", + current_match_id, opponent.username + )) + .await; opponent.current_match = None; opponent.color = Color::None; @@ -462,7 +465,10 @@ impl Server { } let mut observer_messages = Vec::new(); - observer_messages.push(format!("GAME:MOVE:{}:{}", client.username, column)); + observer_messages.push(format!( + "GAME:{}:MOVE:{}:{}", + current_match.id, client.username, column + )); // Check game end conditions let (winner, filled) = current_match.end_game_check(); @@ -475,7 +481,7 @@ impl Server { let opponent = opponent.read().await; let _ = send(&opponent.connection, "GAME:LOSS"); } - observer_messages.push(format!("GAME:WIN:{}", client.username)); + observer_messages.push(format!("GAME:{}:WIN:{}", current_match.id, client.username)); } else if filled { let _ = send(&tx, "GAME:DRAW"); if !current_match.demo_mode { @@ -483,7 +489,7 @@ impl Server { let opponent = opponent.read().await; let _ = send(&opponent.connection, "GAME:DRAW"); } - observer_messages.push("GAME:DRAW".to_string()); + observer_messages.push(format!("GAME:{}:DRAW", current_match.id)); } // remove match from matchmaker @@ -558,6 +564,7 @@ impl Server { let observers = self.observers.clone(); let opponent_move = opponent.clone(); let client_tx = tx.clone(); + let wait_match_id = current_match.id; if current_match.demo_mode { current_match.ledger.push((!client.color, demo_move, Instant::now())); current_match.place_token(!client.color, demo_move); @@ -586,7 +593,10 @@ impl Server { tokio::time::sleep(tokio::time::Duration::from_millis(default_waiting_time)).await; let _ = send(&client_tx, &format!("OPPONENT:{}", demo_move)); let observers_guard = observers.read().await; - let msg = format!("GAME:MOVE:{}:{}", SERVER_PLAYER_USERNAME, demo_move); + let msg = format!( + "GAME:{}:MOVE:{}:{}", + wait_match_id, SERVER_PLAYER_USERNAME, demo_move + ); for (_, tx) in observers_guard.iter() { let _ = send(tx, &msg); } @@ -623,7 +633,7 @@ impl Server { let _ = send(&opponent.connection, "GAME:LOSS"); drop(opponent); let observers_guard = observers.read().await; - let msg = format!("GAME:WIN:{}", client_username); + let msg = format!("GAME:{}:WIN:{}", match_id, client_username); for (_, tx) in observers_guard.iter() { let _ = send(tx, &msg); } @@ -874,7 +884,7 @@ impl Server { timeout_thread.abort(); } - self.broadcast_message_all_observers(&format!("GAME:WIN:{}", winner_username)) + self.broadcast_message_all_observers(&format!("GAME:{}:WIN:{}", match_id, winner_username)) .await; let clients_guard = self.clients.read().await; @@ -1217,7 +1227,8 @@ impl Server { timeout_thread.abort(); } - self.broadcast_message_all_observers("GAME:TERMINATED").await; + self.broadcast_message_all_observers(&format!("GAME:{}:TERMINATED", match_id)) + .await; let clients_guard = self.clients.read().await; if the_match.player1 != SERVER_PLAYER_ADDR.to_string().parse().unwrap() { -- 2.49.1 From 511a104e1235e589db24240a9bc863d1bb48721c Mon Sep 17 00:00:00 2001 From: Joshua Higgins Date: Sun, 15 Mar 2026 04:57:03 +0000 Subject: [PATCH 3/4] fix compile errors --- src/lib.rs | 2 +- src/main.rs | 22 +++++++++++++++++++--- src/server.rs | 4 ++-- src/types.rs | 2 +- 4 files changed, 23 insertions(+), 7 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index cd7b1bd..937fbe3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,6 +1,6 @@ use std::{collections::HashMap, net::SocketAddr, sync::Arc}; -use rand::Rng; +use rand::RngExt; use tokio::sync::{ mpsc::{error::SendError, UnboundedSender}, RwLock, diff --git a/src/main.rs b/src/main.rs index 7d67bb6..a5af3e0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -49,7 +49,7 @@ async fn handle_connection( let ws_stream = accept_async(stream).await?; let (mut ws_sender, mut ws_receiver) = ws_stream.split(); - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); // Spawn task to handle outgoing messages let send_task = tokio::spawn(async move { @@ -262,7 +262,15 @@ async fn handle_connection( let player1_username = usernames[0].to_string(); let player2_username = usernames[1].to_string(); - if let Err(e) = sd.handle_reservation_add(tx.clone(), addr, player1_username, player2_username).await { + if let Err(e) = sd + .handle_reservation_add( + tx.clone(), + addr, + player1_username, + player2_username, + ) + .await + { error!("handle_reservation_add: {}", e); let _ = send(&tx, e.to_string().as_str()); } @@ -277,7 +285,15 @@ async fn handle_connection( let player1_username = usernames[0].to_string(); let player2_username = usernames[1].to_string(); - if let Err(e) = sd.handle_reservation_delete(tx.clone(), addr, player1_username, player2_username).await { + if let Err(e) = sd + .handle_reservation_delete( + tx.clone(), + addr, + player1_username, + player2_username, + ) + .await + { error!("handle_reservation_delete: {}", e); let _ = send(&tx, e.to_string().as_str()); } diff --git a/src/server.rs b/src/server.rs index 3254be8..dbed43d 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,4 +1,5 @@ use anyhow::anyhow; +use rand::RngExt; use std::time::Instant; use crate::{tournaments::*, types::*, *}; @@ -1227,8 +1228,7 @@ impl Server { timeout_thread.abort(); } - self.broadcast_message_all_observers(&format!("GAME:{}:TERMINATED", match_id)) - .await; + self.broadcast_message_all_observers(&format!("GAME:{}:TERMINATED", match_id)).await; let clients_guard = self.clients.read().await; if the_match.player1 != SERVER_PLAYER_ADDR.to_string().parse().unwrap() { diff --git a/src/types.rs b/src/types.rs index c0057e7..4411a97 100644 --- a/src/types.rs +++ b/src/types.rs @@ -1,4 +1,4 @@ -use rand::Rng; +use rand::RngExt; use std::net::SocketAddr; use std::time::Instant; use std::{ops, vec}; -- 2.49.1 From a7b08969a5698a6346beb26b8ee0197469fac62b Mon Sep 17 00:00:00 2001 From: Joshua Higgins Date: Sun, 15 Mar 2026 05:00:10 +0000 Subject: [PATCH 4/4] renames --- src/server.rs | 12 ++++++------ src/tournaments/round_robin.rs | 2 +- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/server.rs b/src/server.rs index dbed43d..db7df8a 100644 --- a/src/server.rs +++ b/src/server.rs @@ -432,7 +432,7 @@ impl Server { let _ = send(&tx, "GAME:LOSS"); let _ = send(&opponent.connection, "GAME:WINS"); - self.broadcast_message_all_observers(&format!( + self.broadcast(&format!( "GAME:{}:WIN:{}", current_match_id, opponent.username )) @@ -885,7 +885,7 @@ impl Server { timeout_thread.abort(); } - self.broadcast_message_all_observers(&format!("GAME:{}:WIN:{}", match_id, winner_username)) + self.broadcast(&format!("GAME:{}:WIN:{}", match_id, winner_username)) .await; let clients_guard = self.clients.read().await; @@ -1001,7 +1001,7 @@ impl Server { // Clear any pending reservations when a tournament starts self.reservations.write().await.clear(); - self.broadcast_message_all_observers(&format!("TOURNAMENT:START:{}", tournament_type)) + self.broadcast(&format!("TOURNAMENT:START:{}", tournament_type)) .await; Ok(()) } @@ -1020,7 +1020,7 @@ impl Server { tourney.write().await.cancel(&self).await; *tournament_guard = None; - self.broadcast_message_all_observers("TOURNAMENT:CANCEL").await; + self.broadcast("TOURNAMENT:CANCEL").await; Ok(()) } @@ -1228,7 +1228,7 @@ impl Server { timeout_thread.abort(); } - self.broadcast_message_all_observers(&format!("GAME:{}:TERMINATED", match_id)).await; + self.broadcast(&format!("GAME:{}:TERMINATED", match_id)).await; let clients_guard = self.clients.read().await; if the_match.player1 != SERVER_PLAYER_ADDR.to_string().parse().unwrap() { @@ -1263,7 +1263,7 @@ impl Server { } } - pub async fn broadcast_message_all_observers(&self, msg: &str) { + pub async fn broadcast(&self, msg: &str) { let observers_guard = self.observers.read().await; for (_, tx) in observers_guard.iter() { let _ = send(tx, msg); diff --git a/src/tournaments/round_robin.rs b/src/tournaments/round_robin.rs index 18c814f..c76d99e 100644 --- a/src/tournaments/round_robin.rs +++ b/src/tournaments/round_robin.rs @@ -167,7 +167,7 @@ impl Tournament for RoundRobin { } message.pop(); - server.broadcast_message_all_observers(&message).await; + server.broadcast(&message).await; if self.is_completed() { // Send scores -- 2.49.1