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

feat: connections & ready up

Browse Source
This commit is contained in:
Joshua Higgins
2025-11-14 11:41:17 -05:00
Unverified
parent 9426fed5b3
commit be4e18323a
3 changed files with 208 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@@ -18,11 +18,9 @@ target
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore # be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear # and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder. # option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/ .idea/
# Added by cargo # Added by cargo
/target /target
/.idea

256
src/main.rs
View File

@@ -1,123 +1,209 @@
mod types; mod types;
use crate::types::{Color, MatchMaker, Role, AI};
use futures_util::{SinkExt, StreamExt};
use std::collections::HashMap;
use std::net::SocketAddr; use std::net::SocketAddr;
use std::sync::Arc; use std::sync::Arc;
use tokio::net::{TcpListener, TcpStream}; use tokio::net::{TcpListener, TcpStream};
use tokio::sync::RwLock; use tokio::sync::RwLock;
use tokio_tungstenite::{accept_async, tungstenite::Message}; use tokio_tungstenite::{accept_async, tungstenite::Message};
use futures_util::{SinkExt, StreamExt}; use tracing::{error, info, warn};
use std::collections::HashMap;
use tracing::{info, error, warn};
use types::Client; use types::Client;
use crate::types::{Color, Role, AI};
type Clients = Arc<RwLock<HashMap<SocketAddr, Client>>>; type Clients = Arc<RwLock<HashMap<SocketAddr, Client>>>;
#[tokio::main] #[tokio::main]
async fn main() -> Result<(), anyhow::Error> { async fn main() -> Result<(), anyhow::Error> {
// Initialize logging // Initialize logging
tracing_subscriber::fmt::init(); tracing_subscriber::fmt::init();
let addr = "127.0.0.1:8080"; let addr = "0.0.0.0:8080";
let listener = TcpListener::bind(&addr).await?; let listener = TcpListener::bind(&addr).await?;
info!("WebSocket server listening on: {}", addr); info!("WebSocket server listening on: {}", addr);
let mut clients: Clients = Arc::new(RwLock::new(HashMap::new())); let mut clients: Clients = Arc::new(RwLock::new(HashMap::new()));
let mut match_maker: Arc<RwLock<MatchMaker>> = Arc::new(RwLock::new(MatchMaker::new()));
while let Ok((stream, addr)) = listener.accept().await { while let Ok((stream, addr)) = listener.accept().await {
tokio::spawn(handle_connection(stream, addr, clients.clone())); tokio::spawn(handle_connection(
} stream,
addr,
clients.clone(),
match_maker.clone(),
));
}
Ok(()) Ok(())
} }
async fn handle_connection( async fn handle_connection(
stream: TcpStream, stream: TcpStream,
addr: SocketAddr, addr: SocketAddr,
clients: Clients, clients: Clients,
match_maker: Arc<RwLock<MatchMaker>>,
) -> Result<(), anyhow::Error> { ) -> Result<(), anyhow::Error> {
info!("New WebSocket connection from: {}", addr); info!("New WebSocket connection from: {}", addr);
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();
// Store the client // Store the client
{ {
clients.write().await.insert(addr, Client::new(String::new(), Role::Observer, tx)); clients
} .write()
.await
.insert(addr, Client::new(String::new(), Role::Observer, tx));
}
// Spawn task to handle outgoing messages // Spawn task to handle outgoing messages
let mut send_task = tokio::spawn(async move { let mut send_task = tokio::spawn(async move {
while let Some(msg) = rx.recv().await { while let Some(msg) = rx.recv().await {
if ws_sender.send(msg).await.is_err() { if ws_sender.send(msg).await.is_err() {
break; break;
} }
} }
}); });
// Handle incoming messages // Handle incoming messages
while let Some(msg) = ws_receiver.next().await { while let Some(msg) = ws_receiver.next().await {
match msg { match msg {
Ok(Message::Text(text)) => { Ok(Message::Text(text)) => {
info!("Received text from {}: {}", addr, text); info!("Received text from {}: {}", addr, text);
if text.starts_with("CONNECT") { if text.starts_with("CONNECT") {
let requested_username = text.split(":").collect::<Vec<&str>>()[1].to_string(); let requested_username = text.split(":").collect::<Vec<&str>>()[1].to_string();
let mut is_taken = false; let mut is_taken = false;
for client in clients.read().await.values() { for client in clients.read().await.values() {
if let Some(username) = &client.username { if let Some(username) = &client.username {
if *username == requested_username { if *username == requested_username {
is_taken = true; let _ = clients
break; .read()
} .await
.get(&addr)
.unwrap()
.send(&format!("ERROR:INVALID:ID:{}", requested_username));
is_taken = true;
break;
}
}
}
if is_taken { continue; }
// not taken
clients.write().await.get_mut(&addr).unwrap().role = Role::Player;
clients.write().await.get_mut(&addr).unwrap().username = Some(requested_username);
let _ = clients
.read()
.await
.get(&addr)
.unwrap()
.send("CONNECT:ACK");
}
else if text.starts_with("READY") {
if clients.read().await.get(&addr).unwrap().role != Role::Player {
let _ = clients
.read()
.await
.get(&addr)
.unwrap()
.send("ERROR:INVALID");
continue;
}
let mut already_ready = false;
for ready_player in &match_maker.read().await.ready_players {
if ready_player.username.eq(clients.read().await.get(&addr).unwrap().username.as_ref().unwrap()) {
let _ = clients
.read()
.await
.get(&addr)
.unwrap()
.send("ERROR:INVALID");
already_ready = true;
break;
} }
} }
if is_taken { if already_ready { continue; }
let _ = clients.read().await.get(&addr).unwrap().send("ERROR:INVALID:ID:".to_string() + &*requested_username);
continue;
}
// not taken match_maker.write().await.ready_players.push(AI::new(
clients.write().await.get_mut(&addr).unwrap().role = Role::Player; clients
clients.write().await.get_mut(&addr).unwrap().username = Some(requested_username); .read()
} .await
else if text.starts_with("READY") { .get(&addr)
// TODO! .unwrap()
} .username
.as_ref()
.unwrap(),
Color::None,
));
let _ = clients
.read()
.await
.get(&addr)
.unwrap()
.send("READY:ACK");
}
else if text.starts_with("PLAY") { else if text.starts_with("PLAY") {
// TODO! // TODO!
} // Check if client is valid
// Check if it's their move
// Check if valid move
// Place it
// Check game end conditions
// Broadcast move
}
else { else {
let _ = clients.read().await.get(&addr).unwrap().send("ERROR:UNKNOWN".to_string()); let _ = clients
} .read()
} .await
Ok(Message::Close(_)) => { .get(&addr)
info!("Client {} disconnected", addr); .unwrap()
break; .send("ERROR:UNKNOWN");
} }
Ok(_) => { let _ = clients.read().await.get(&addr).unwrap().send("ERROR:UNKNOWN".to_string()); } }
Err(e) => { Ok(Message::Close(_)) => {
error!("WebSocket error for {}: {}", addr, e); info!("Client {} disconnected", addr);
break; break;
} }
} Ok(_) => {
} let _ = clients
.read()
.await
.get(&addr)
.unwrap()
.send("ERROR:UNKNOWN");
}
Err(e) => {
error!("WebSocket error for {}: {}", addr, e);
break;
}
}
}
// Clean up // Clean up
send_task.abort(); send_task.abort();
clients.write().await.remove(&addr);
info!("Client {} removed", addr);
Ok(()) // TODO: Remove and terminate any matches
clients.write().await.remove(&addr);
info!("Client {} removed", addr);
Ok(())
} }
async fn broadcast_message(clients: &Clients, msg: Message) { async fn broadcast_message(clients: &Clients, msg: Message) {
let clients = clients.read().await; let clients = clients.read().await;
for (_, client) in clients.iter() { for (_, client) in clients.iter() {
if client.role == Role::Admin || client.role == Role::Observer { if client.role == Role::Admin || client.role == Role::Observer {
client.connection.send(msg.clone()).ok(); client.connection.send(msg.clone()).ok();
} }
} }
} }

38
src/types.rs
View File

@@ -1,3 +1,5 @@
use std::collections::HashMap;
use std::net::SocketAddr;
use tokio::sync::mpsc::error::SendError; use tokio::sync::mpsc::error::SendError;
use tokio::sync::mpsc::UnboundedSender; use tokio::sync::mpsc::UnboundedSender;
use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::tungstenite::Message;
@@ -33,7 +35,7 @@ impl Client {
} }
} }
pub fn send(&self, text: String) -> Result<(), SendError<Message>> { pub fn send(&self, text: &str) -> Result<(), SendError<Message>> {
self.connection.send(Message::text(text)) self.connection.send(Message::text(text))
} }
} }
@@ -51,14 +53,42 @@ impl AI {
} }
pub struct Match { pub struct Match {
pub id: u32,
pub board: Vec<Vec<Color>>, pub board: Vec<Vec<Color>>,
pub viewers: Vec<SocketAddr>,
pub player1: AI, pub player1: AI,
pub player2: AI, pub player2: AI,
} }
impl Match { impl Match {
pub fn new(id: u32, player1: AI, player2: AI) -> Match { pub fn new(player1: AI, player2: AI) -> Match {
Match { id, board: Vec::new(), player1, player2 } Match { board: Vec::new(), viewers: Vec::new(), player1, player2 }
}
}
pub struct MatchMaker {
pub matches: HashMap<u32, Match>,
pub ready_players: Vec<AI>,
}
impl MatchMaker {
pub fn new() -> MatchMaker {
MatchMaker { matches: HashMap::new(), ready_players: Vec::new() }
}
pub fn watch(&mut self, viewer: SocketAddr, match_id: u32) {
for aMatch in &mut self.matches.values_mut() {
let mut found = false;
for i in 0..aMatch.viewers.len() {
if aMatch.viewers[i] == viewer {
aMatch.viewers.remove(i);
found = true;
break;
}
}
if found { break; }
}
self.matches.get_mut(&match_id).unwrap().viewers.push(viewer);
} }
} }