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 ae5bb6a..a5af3e0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -49,10 +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(); - - // Store the client - sd.observers.write().await.insert(addr, tx.clone()); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); // Spawn task to handle outgoing messages let send_task = tokio::spawn(async move { @@ -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); @@ -259,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()); } @@ -274,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 f4384b7..db7df8a 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::*, *}; @@ -80,8 +81,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 +245,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 +417,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 +432,11 @@ 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(&format!( + "GAME:{}:WIN:{}", + current_match_id, opponent.username + )) + .await; opponent.current_match = None; opponent.color = Color::None; @@ -448,10 +465,11 @@ 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:{}:{}", + current_match.id, client.username, column + )); // Check game end conditions let (winner, filled) = current_match.end_game_check(); @@ -464,7 +482,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:{}", current_match.id, client.username)); } else if filled { let _ = send(&tx, "GAME:DRAW"); if !current_match.demo_mode { @@ -472,7 +490,7 @@ impl Server { let opponent = opponent.read().await; let _ = send(&opponent.connection, "GAME:DRAW"); } - viewer_messages.push("GAME:DRAW".to_string()); + observer_messages.push(format!("GAME:{}:DRAW", current_match.id)); } // remove match from matchmaker @@ -547,6 +565,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); @@ -564,19 +583,24 @@ 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:{}:{}", + wait_match_id, SERVER_PLAYER_USERNAME, demo_move + ); + for (_, tx) in observers_guard.iter() { + let _ = send(tx, &msg); + } } })); @@ -590,7 +614,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 +633,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:{}", match_id, 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 +731,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 +885,8 @@ impl Server { timeout_thread.abort(); } - self.broadcast_message(&the_match.viewers, &format!("GAME:WIN:{}", winner_username)).await; + self.broadcast(&format!("GAME:{}:WIN:{}", match_id, winner_username)) + .await; let clients_guard = self.clients.read().await; if the_match.demo_mode { @@ -979,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(()) } @@ -998,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(()) } @@ -1187,54 +1209,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 +1228,7 @@ impl Server { timeout_thread.abort(); } - self.broadcast_message(&the_match.viewers, "GAME:TERMINATED").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() { @@ -1289,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 diff --git a/src/types.rs b/src/types.rs index fef36f7..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}; @@ -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,