misc: rework observers

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);

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<Message>,
+    ) -> 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 mut observer_messages = Vec::new();
-        let viewers = current_match.viewers.clone();
+        observer_messages.push(format!("GAME:MOVE:{}:{}", client.username, column));
-
-        viewer_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 {
+            for msg in observer_messages {
-                broadcast_message(&observers, &viewers, &msg).await;
+                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(
+                let observers_guard = observers.read().await;
-                    &observers,
+                let msg = format!("GAME:MOVE:{}:{}", SERVER_PLAYER_USERNAME, demo_move);
-                    &viewers,
+                for (_, tx) in observers_guard.iter() {
-                    &format!("GAME:MOVE:{}:{}", SERVER_PLAYER_USERNAME, demo_move),
+                    let _ = send(tx, &msg);
-                )
+                }
-                .await;
             }
         }));
 
@@ -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(
+                    let observers_guard = observers.read().await;
-                        &observers,
+                    let msg = format!("GAME:WIN:{}", client_username);
-                        &viewers,
+                    for (_, tx) in observers_guard.iter() {
-                        &format!("GAME:WIN:{}", client_username),
+                        let _ = send(tx, &msg);
-                    )
+                    }
-                    .await;
 
                     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<Message>,
         match_id: u32,
-        addr: SocketAddr,
+        _addr: SocketAddr,
     ) -> Result<(), anyhow::Error> {
-        let result = self.watch(match_id, addr).await;
+        if self.matches.read().await.get(&match_id).is_none() {
-        if result.is_err() {
             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() {

src/types.rs

@@ -61,7 +61,6 @@ pub struct Match {
     pub id: u32,
     pub demo_mode: bool,
     pub board: Vec<Vec<Color>>,
-    pub viewers: Vec<SocketAddr>,
     pub ledger: Vec<(Color, usize, Instant)>,
     pub wait_thread: Option<tokio::task::JoinHandle<()>>,
     pub timeout_thread: Option<tokio::task::JoinHandle<()>>,
@@ -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,