joshuafhiggins/connect4-moderator-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Releases 2 Activity

misc: rework observers #24

Closed
joshuafhiggins wants to merge 4 commits from feature/misc-rework-observers into main
pull from: feature/misc-rework-observers
merge into: joshuafhiggins:main
Conversation 0 Commits 4 Files Changed 5 +84 -93
5 changed files with 84 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/lib.rs
View File

@@ -1,6 +1,6 @@
use std::{collections::HashMap, net::SocketAddr, sync::Arc}; use std::{collections::HashMap, net::SocketAddr, sync::Arc};
use rand::Rng; use rand::RngExt;
use tokio::sync::{ use tokio::sync::{
mpsc::{error::SendError, UnboundedSender}, mpsc::{error::SendError, UnboundedSender},
RwLock, RwLock,

31
src/main.rs
View File

@@ -49,10 +49,7 @@ async fn handle_connection(
let ws_stream = accept_async(stream).await?; let ws_stream = accept_async(stream).await?;
let (mut ws_sender, mut ws_receiver) = ws_stream.split(); 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::<Message>();
// Store the client
sd.observers.write().await.insert(addr, tx.clone());
// Spawn task to handle outgoing messages // Spawn task to handle outgoing messages
let send_task = tokio::spawn(async move { let send_task = tokio::spawn(async move {
@@ -103,6 +100,12 @@ async fn handle_connection(
let _ = send(&tx, e.to_string().as_str()); 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" => { "READY" => {
if let Err(e) = sd.handle_ready(addr, tx.clone()).await { if let Err(e) = sd.handle_ready(addr, tx.clone()).await {
error!("handle_ready: {}", e); error!("handle_ready: {}", e);
@@ -259,7 +262,15 @@ async fn handle_connection(
let player1_username = usernames[0].to_string(); let player1_username = usernames[0].to_string();
let player2_username = usernames[1].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); error!("handle_reservation_add: {}", e);
let _ = send(&tx, e.to_string().as_str()); let _ = send(&tx, e.to_string().as_str());
} }
@@ -274,7 +285,15 @@ async fn handle_connection(
let player1_username = usernames[0].to_string(); let player1_username = usernames[0].to_string();
let player2_username = usernames[1].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); error!("handle_reservation_delete: {}", e);
let _ = send(&tx, e.to_string().as_str()); let _ = send(&tx, e.to_string().as_str());
} }

138
src/server.rs
View File

@@ -1,4 +1,5 @@
use anyhow::anyhow; use anyhow::anyhow;
use rand::RngExt;
use std::time::Instant; use std::time::Instant;
use crate::{tournaments::*, types::*, *}; use crate::{tournaments::*, types::*, *};
@@ -80,8 +81,6 @@ impl Server {
drop(clients_guard); 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); self.usernames.write().await.insert(requested_username.clone(), addr);
let _ = send(&tx, "CONNECT:ACK"); let _ = send(&tx, "CONNECT:ACK");
@@ -246,12 +245,27 @@ impl Server {
} }
self.clients.write().await.remove(&addr); self.clients.write().await.remove(&addr);
self.observers.write().await.insert(addr, tx.clone());
let _ = send(&tx, "DISCONNECT:ACK"); let _ = send(&tx, "DISCONNECT:ACK");
Ok(()) 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( pub async fn handle_ready(
&self, &self,
addr: SocketAddr, addr: SocketAddr,
@@ -403,7 +417,6 @@ impl Server {
if invalid { if invalid {
let current_match_id = current_match.id; let current_match_id = current_match.id;
let is_demo_mode = current_match.demo_mode; let is_demo_mode = current_match.demo_mode;
let viewers = current_match.viewers.clone();
drop(current_match); drop(current_match);
drop(matches_guard); drop(matches_guard);
@@ -419,7 +432,11 @@ impl Server {
let _ = send(&tx, "GAME:LOSS"); let _ = send(&tx, "GAME:LOSS");
let _ = send(&opponent.connection, "GAME:WINS"); 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.current_match = None;
opponent.color = Color::None; opponent.color = Color::None;
@@ -448,10 +465,11 @@ impl Server {
timeout_thread.abort(); 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:{}:{}",
viewer_messages.push(format!("GAME:MOVE:{}:{}", client.username, column)); current_match.id, client.username, column
));
// Check game end conditions // Check game end conditions
let (winner, filled) = current_match.end_game_check(); let (winner, filled) = current_match.end_game_check();
@@ -464,7 +482,7 @@ impl Server {
let opponent = opponent.read().await; let opponent = opponent.read().await;
let _ = send(&opponent.connection, "GAME:LOSS"); 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 { } else if filled {
let _ = send(&tx, "GAME:DRAW"); let _ = send(&tx, "GAME:DRAW");
if !current_match.demo_mode { if !current_match.demo_mode {
@@ -472,7 +490,7 @@ impl Server {
let opponent = opponent.read().await; let opponent = opponent.read().await;
let _ = send(&opponent.connection, "GAME:DRAW"); 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 // remove match from matchmaker
@@ -547,6 +565,7 @@ impl Server {
let observers = self.observers.clone(); let observers = self.observers.clone();
let opponent_move = opponent.clone(); let opponent_move = opponent.clone();
let client_tx = tx.clone(); let client_tx = tx.clone();
let wait_match_id = current_match.id;
if current_match.demo_mode { if current_match.demo_mode {
current_match.ledger.push((!client.color, demo_move, Instant::now())); current_match.ledger.push((!client.color, demo_move, Instant::now()));
current_match.place_token(!client.color, demo_move); current_match.place_token(!client.color, demo_move);
@@ -564,19 +583,24 @@ 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 { if demo_mode && no_winner {
tokio::time::sleep(tokio::time::Duration::from_millis(default_waiting_time)).await; tokio::time::sleep(tokio::time::Duration::from_millis(default_waiting_time)).await;
let _ = send(&client_tx, &format!("OPPONENT:{}", demo_move)); let _ = send(&client_tx, &format!("OPPONENT:{}", demo_move));
broadcast_message( let observers_guard = observers.read().await;
&observers, let msg = format!(
&viewers, "GAME:{}:MOVE:{}:{}",
&format!("GAME:MOVE:{}:{}", SERVER_PLAYER_USERNAME, demo_move), wait_match_id, SERVER_PLAYER_USERNAME, demo_move
) );
.await; for (_, tx) in observers_guard.iter() {
let _ = send(tx, &msg);
}
} }
})); }));
@@ -590,7 +614,6 @@ impl Server {
let client_tx = tx.clone(); let client_tx = tx.clone();
let client_addr = addr.clone(); let client_addr = addr.clone();
let observers = self.observers.clone(); let observers = self.observers.clone();
let viewers = current_match.viewers.clone();
let opponent_move = opponent.clone(); let opponent_move = opponent.clone();
current_match.timeout_thread = Some(tokio::spawn(async move { current_match.timeout_thread = Some(tokio::spawn(async move {
if demo_mode { if demo_mode {
@@ -610,12 +633,11 @@ impl Server {
let opponent = opponent.read().await; let opponent = opponent.read().await;
let _ = send(&opponent.connection, "GAME:LOSS"); let _ = send(&opponent.connection, "GAME:LOSS");
drop(opponent); drop(opponent);
broadcast_message( let observers_guard = observers.read().await;
&observers, let msg = format!("GAME:{}:WIN:{}", match_id, 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 clients_guard = clients.write().await;
let mut client = clients_guard.get_mut(&client_addr).unwrap().write().await; let mut client = clients_guard.get_mut(&client_addr).unwrap().write().await;
@@ -709,10 +731,9 @@ impl Server {
&self, &self,
tx: UnboundedSender<Message>, tx: UnboundedSender<Message>,
match_id: u32, match_id: u32,
addr: SocketAddr, _addr: SocketAddr,
) -> Result<(), anyhow::Error> { ) -> 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")); return Err(anyhow::anyhow!("ERROR:INVALID:WATCH"));
} }
@@ -864,7 +885,8 @@ impl Server {
timeout_thread.abort(); 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; let clients_guard = self.clients.read().await;
if the_match.demo_mode { if the_match.demo_mode {
@@ -979,7 +1001,7 @@ impl Server {
// Clear any pending reservations when a tournament starts // Clear any pending reservations when a tournament starts
self.reservations.write().await.clear(); self.reservations.write().await.clear();
self.broadcast_message_all_observers(&format!("TOURNAMENT:START:{}", tournament_type)) self.broadcast(&format!("TOURNAMENT:START:{}", tournament_type))
.await; .await;
Ok(()) Ok(())
} }
@@ -998,7 +1020,7 @@ impl Server {
tourney.write().await.cancel(&self).await; tourney.write().await.cancel(&self).await;
*tournament_guard = None; *tournament_guard = None;
self.broadcast_message_all_observers("TOURNAMENT:CANCEL").await; self.broadcast("TOURNAMENT:CANCEL").await;
Ok(()) Ok(())
} }
@@ -1187,54 +1209,6 @@ impl Server {
Ok(()) 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) { pub async fn terminate_match(&self, match_id: u32) {
let matches_guard = self.matches.read().await; let matches_guard = self.matches.read().await;
let the_match = matches_guard.get(&match_id); let the_match = matches_guard.get(&match_id);
@@ -1254,7 +1228,7 @@ impl Server {
timeout_thread.abort(); 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; let clients_guard = self.clients.read().await;
if the_match.player1 != SERVER_PLAYER_ADDR.to_string().parse().unwrap() { 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; let observers_guard = self.observers.read().await;
for (_, tx) in observers_guard.iter() { for (_, tx) in observers_guard.iter() {
let _ = send(tx, msg); let _ = send(tx, msg);

2
src/tournaments/round_robin.rs
View File

@@ -167,7 +167,7 @@ impl Tournament for RoundRobin {
} }
message.pop(); message.pop();
server.broadcast_message_all_observers(&message).await; server.broadcast(&message).await;
if self.is_completed() { if self.is_completed() {
// Send scores // Send scores

4
src/types.rs
View File

@@ -1,4 +1,4 @@
use rand::Rng; use rand::RngExt;
use std::net::SocketAddr; use std::net::SocketAddr;
use std::time::Instant; use std::time::Instant;
use std::{ops, vec}; use std::{ops, vec};
@@ -61,7 +61,6 @@ pub struct Match {
pub id: u32, pub id: u32,
pub demo_mode: bool, pub demo_mode: bool,
pub board: Vec<Vec<Color>>, pub board: Vec<Vec<Color>>,
pub viewers: Vec<SocketAddr>,
pub ledger: Vec<(Color, usize, Instant)>, pub ledger: Vec<(Color, usize, Instant)>,
pub wait_thread: Option<tokio::task::JoinHandle<()>>, pub wait_thread: Option<tokio::task::JoinHandle<()>>,
pub timeout_thread: Option<tokio::task::JoinHandle<()>>, pub timeout_thread: Option<tokio::task::JoinHandle<()>>,
@@ -81,7 +80,6 @@ impl Match {
id, id,
demo_mode, demo_mode,
board: vec![vec![Color::None; 6]; 7], board: vec![vec![Color::None; 6]; 7],
viewers: Vec::new(),
ledger: Vec::new(), ledger: Vec::new(),
wait_thread: None, wait_thread: None,
timeout_thread: None, timeout_thread: None,