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 rand::Rng;
use rand::RngExt;
use tokio::sync::{
mpsc::{error::SendError, UnboundedSender},
RwLock,

31
src/main.rs
View File

@@ -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::<Message>();
// 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());
}

138
src/server.rs
View File

@@ -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<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 +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<Message>,
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);

2
src/tournaments/round_robin.rs
View File

@@ -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

4
src/types.rs
View File

@@ -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<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,