commit 4f29e83d7962f664bea3dea4a1438e569781ad1e
parent b8e5176a875a15b95b15908f7a40baaa0fa1ae33
Author: Bradley Gannon <bradley@bradleygannon.com>
Date: Sat, 15 Aug 2026 18:08:30 -0400
Improve connection handling and player management
Diffstat:
10 files changed, 207 insertions(+), 223 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -90,6 +90,7 @@ version = "0.1.0"
dependencies = [
"anyhow",
"glam",
+ "mio",
"postcard",
"serde",
"uuid",
@@ -463,6 +464,7 @@ dependencies = [
"mio",
"postcard",
"serde",
+ "uuid",
]
[[package]]
diff --git a/client/src/main.rs b/client/src/main.rs
@@ -4,16 +4,12 @@ use std::{
time::{SystemTime, UNIX_EPOCH},
};
-use common::{
- codec::Codec,
- game::{Action, ActionKind, GameState},
-};
+use common::game::{Action, ActionKind, GameState, Response};
use macroquad::{prelude::*, rand::srand};
-use uuid::Uuid;
use crate::{
game::handle_input,
- net::{OnShutdown, handle_connection},
+ net::handle_connection,
render::{Camera, follow_local_player, load_sprites, render},
};
@@ -44,34 +40,36 @@ async fn main() {
.subsec_nanos() as u64,
);
- // TODO: generate this on the server
- let local_player_id = Uuid::new_v4();
-
- let _ = client_to_server_tx.send(Action::new(
- local_player_id,
- ActionKind::Join {
- name: local_player_id.to_string().chars().take(3).collect(),
- },
- ));
-
let mut camera = Camera::default();
- let mut codec = Codec::default();
let mut game = GameState::default();
- let _on_shutdown = OnShutdown::new(client_to_server_tx.clone(), local_player_id);
+ let mut local_player_id = None;
+ //let _on_shutdown = OnShutdown::new(client_to_server_tx.clone(), local_player_id);
loop {
- let bytes: Vec<u8> = server_to_client_rx.try_iter().flatten().collect();
- if let Some(decoded) = codec.decode(&bytes).last() {
- if let Ok(g) = postcard::from_bytes::<GameState>(decoded) {
- game = g;
- }
+ for response in server_to_client_rx.try_iter() {
+ match response {
+ Response::SetId { id } => {
+ let _ = client_to_server_tx.send(Action::new(
+ id,
+ ActionKind::Join {
+ name: id.to_string().chars().take(3).collect(),
+ },
+ ));
+ local_player_id = Some(id);
+ }
+ Response::UpdateGameState {
+ game_state: new_game_state,
+ } => game = new_game_state,
+ };
}
- if let Some(action) = handle_input(&local_player_id) {
- if let Err(_) = client_to_server_tx.send(action) {
- return;
+ if let Some(id) = local_player_id {
+ if let Some(action) = handle_input(&id) {
+ if let Err(_) = client_to_server_tx.send(action) {
+ return;
+ }
}
+ follow_local_player(&mut camera, &game, &id);
}
- follow_local_player(&mut camera, &game, &local_player_id);
render(&camera, &game, &spritesheet);
next_frame().await;
}
diff --git a/client/src/net.rs b/client/src/net.rs
@@ -1,5 +1,4 @@
use std::{
- io::{self, Read, Write},
net::SocketAddr,
sync::{
Arc, Mutex,
@@ -10,15 +9,14 @@ use std::{
};
use common::{
- codec::Codec,
- game::{Action, ActionKind},
+ game::{Action, Response},
+ net::{handle_network_read, handle_network_write},
};
-use mio::{Events, Interest, Poll, Token, net::TcpStream};
-use uuid::Uuid;
+use mio::{Interest, Poll, Token, net::TcpStream};
const TOKEN: Token = Token(0);
-pub(crate) fn handle_connection(addr: SocketAddr, tx: Sender<Vec<u8>>, rx: Receiver<Action>) {
+pub(crate) fn handle_connection(addr: SocketAddr, tx: Sender<Response>, rx: Receiver<Action>) {
let mut stream = TcpStream::connect(addr).unwrap();
let poll = Poll::new().unwrap();
poll.registry()
@@ -27,87 +25,9 @@ pub(crate) fn handle_connection(addr: SocketAddr, tx: Sender<Vec<u8>>, rx: Recei
let shared_stream = Arc::new(Mutex::new(stream));
let shared_stream_read = shared_stream.clone();
let shared_stream_write = shared_stream.clone();
- let read_thread = thread::spawn(|| handle_server_read(poll, shared_stream_read, tx));
- let write_thread = thread::spawn(|| handle_server_write(shared_stream_write, rx));
+ let read_thread = thread::spawn(|| handle_network_read(poll, shared_stream_read, tx));
+ let write_thread = thread::spawn(|| handle_network_write(shared_stream_write, rx));
while !read_thread.is_finished() && !write_thread.is_finished() {
thread::sleep(Duration::from_millis(500));
}
}
-
-pub(crate) fn handle_server_read(
- mut poll: Poll,
- stream: Arc<Mutex<TcpStream>>,
- tx: Sender<Vec<u8>>,
-) {
- let mut events = Events::with_capacity(128);
- let mut buffer = [0x00; 1024];
- let mut buffer_index;
- loop {
- // need an event loop here because otherwise this thread would hold a permanent lock on the
- // stream while blocking on reading it
- poll.poll(&mut events, None).unwrap();
- for event in events.iter() {
- if event.is_readable() {
- buffer_index = stream.lock().unwrap().read(&mut buffer).unwrap();
- if let Err(_) = tx.send(buffer[..buffer_index].to_vec()) {
- break;
- }
- }
- }
- }
-}
-
-fn handle_server_write(stream: Arc<Mutex<TcpStream>>, rx: Receiver<Action>) {
- let mut all_sent;
- loop {
- match rx.recv() {
- Ok(m) => match postcard::to_stdvec(&m) {
- Ok(to_write) => {
- all_sent = false;
- while !all_sent {
- match stream
- .lock()
- .unwrap()
- .write_all(Codec::encode(&to_write).as_slice())
- {
- Ok(_) => all_sent = true,
- Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
- all_sent = false;
- }
- Err(e) if e.kind() == io::ErrorKind::NotConnected => {
- all_sent = false;
- }
- Err(_) => {
- return;
- }
- }
- }
- }
- Err(_) => (),
- },
- Err(_) => return,
- }
- }
-}
-
-pub(crate) struct OnShutdown {
- tx: Sender<Action>,
- local_player_id: Uuid,
-}
-
-impl OnShutdown {
- pub(crate) fn new(tx: Sender<Action>, local_player_id: Uuid) -> Self {
- OnShutdown {
- tx,
- local_player_id,
- }
- }
-}
-
-impl Drop for OnShutdown {
- fn drop(&mut self) {
- let _ = self
- .tx
- .send(Action::new(self.local_player_id, ActionKind::Leave));
- }
-}
diff --git a/common/Cargo.toml b/common/Cargo.toml
@@ -6,6 +6,7 @@ edition = "2024"
[dependencies]
anyhow = "1.0.104"
glam = { version = "0.27.0", features = ["scalar-math", "serde"] }
+mio = { version = "1.2.2", features = ["net", "os-poll"] }
postcard = { version = "1.1.3", default-features = false, features = ["use-std"] }
serde = { version = "1.0.229", features = ["derive"] }
-uuid = { version = "1.24.1", features = ["v4", "serde"] }
+uuid = { version = "1.24.1", features = ["serde", "v4"] }
diff --git a/common/src/game.rs b/common/src/game.rs
@@ -4,7 +4,7 @@ use glam::Vec2;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub struct GameState {
pub timestamp: SystemTime,
pub sequence_number: u32,
@@ -21,14 +21,14 @@ impl Default for GameState {
}
}
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub struct Player {
pub name: String,
pub position: Vec2,
pub sprite_index: SpriteIndex,
}
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub struct Action {
pub player_id: Uuid,
pub kind: ActionKind,
@@ -40,14 +40,19 @@ impl Action {
}
}
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub enum ActionKind {
Join { name: String },
Move(Vec2),
- Leave,
}
-#[derive(Serialize, Deserialize, Debug, Clone)]
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
pub enum SpriteIndex {
Meeple,
}
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
+pub enum Response {
+ SetId { id: Uuid },
+ UpdateGameState { game_state: GameState },
+}
diff --git a/common/src/lib.rs b/common/src/lib.rs
@@ -1,2 +1,3 @@
pub mod codec;
pub mod game;
+pub mod net;
diff --git a/common/src/net.rs b/common/src/net.rs
@@ -0,0 +1,87 @@
+use std::{
+ io::{self, Read, Write},
+ sync::{
+ Arc, Mutex,
+ mpsc::{Receiver, Sender},
+ },
+};
+
+use mio::{Events, Poll, net::TcpStream};
+use serde::{Serialize, de::DeserializeOwned};
+
+use crate::codec::Codec;
+
+pub fn handle_network_read<T: DeserializeOwned>(
+ mut poll: Poll,
+ stream: Arc<Mutex<TcpStream>>,
+ tx: Sender<T>,
+) {
+ let mut events = Events::with_capacity(128);
+ let mut codec = Codec::default();
+ let mut buffer = [0x00; 1024];
+ let mut buffer_index;
+ loop {
+ // need an event loop here because otherwise the calling thread would hold a permanent lock
+ // on the stream while blocking on reading it
+ poll.poll(&mut events, None).unwrap();
+ for event in events.iter() {
+ if event.is_readable() {
+ let mut stream = stream.lock().unwrap();
+ match stream.read(&mut buffer) {
+ Ok(0) => return,
+ Ok(i) => {
+ buffer_index = i;
+ let messages = codec.decode(&buffer[..buffer_index]);
+ for message in messages {
+ match postcard::from_bytes(message.as_slice()) {
+ Ok(m) => {
+ if let Err(_) = tx.send(m) {
+ break;
+ }
+ }
+ Err(_) => (),
+ }
+ }
+ }
+ Err(e) => match e.kind() {
+ io::ErrorKind::WouldBlock
+ | io::ErrorKind::NotConnected
+ | io::ErrorKind::ResourceBusy
+ | io::ErrorKind::Interrupted => (),
+ _ => return,
+ },
+ }
+ }
+ }
+ }
+}
+
+pub fn handle_network_write<T: Serialize>(stream: Arc<Mutex<TcpStream>>, rx: Receiver<T>) {
+ loop {
+ match rx.recv() {
+ Ok(m) => match postcard::to_stdvec(&m) {
+ Ok(to_write) => {
+ let mut all_sent = false;
+ while !all_sent {
+ if let Ok(mut s) = stream.lock() {
+ match s.write_all(Codec::encode(&to_write).as_slice()) {
+ Ok(_) => all_sent = true,
+ Err(e) => match e.kind() {
+ io::ErrorKind::WouldBlock
+ | io::ErrorKind::NotConnected
+ | io::ErrorKind::ResourceBusy
+ | io::ErrorKind::Interrupted => {
+ all_sent = false;
+ }
+ _ => return,
+ },
+ }
+ }
+ }
+ }
+ Err(_) => (),
+ },
+ Err(_) => return,
+ }
+ }
+}
diff --git a/server/Cargo.toml b/server/Cargo.toml
@@ -10,3 +10,4 @@ glam = { version = "0.27.0", features = ["scalar-math", "serde"] }
mio = { version = "1.2.2", features = ["net", "os-poll"] }
postcard = { version = "1.1.3", default-features = false, features = ["use-std"] }
serde = { version = "1.0.229", features = ["derive"] }
+uuid = { version = "1.24.1", features = ["v4", "serde"] }
diff --git a/server/src/main.rs b/server/src/main.rs
@@ -5,12 +5,8 @@ use std::{
time::{Duration, Instant},
};
-use common::{
- codec::Codec,
- game::{ActionKind, GameState, Player, SpriteIndex},
-};
+use common::game::{ActionKind, GameState, Player, Response, SpriteIndex};
use glam::Vec2;
-use postcard::to_stdvec;
use crate::net::listen;
@@ -19,21 +15,21 @@ pub mod net;
const TICK_DURATION: Duration = Duration::from_millis(10);
fn main() {
- let handlers = Arc::new(Mutex::new(LinkedList::new()));
- let (handler_to_main_tx, handler_to_main_rx) = channel();
- let handlers_clone = handlers.clone();
- let _listener = thread::spawn(|| listen(handlers_clone, handler_to_main_tx));
+ let clients = Arc::new(Mutex::new(LinkedList::new()));
+ let (client_to_main_tx, client_to_main_rx) = channel();
+ let clients_clone = clients.clone();
+ let _listener = thread::spawn(|| listen(clients_clone, client_to_main_tx));
let mut next_tick = Instant::now() + TICK_DURATION;
- let mut game = GameState::default();
+ let mut game_state = GameState::default();
loop {
- let mut send_game_status = false;
+ let mut send_game_state = false;
// Process all pending player actions
- for action in handler_to_main_rx.try_iter() {
- send_game_status = true;
+ for action in client_to_main_rx.try_iter() {
+ send_game_state = true;
match action.kind {
ActionKind::Join { name } => {
- game.players.insert(
+ game_state.players.insert(
action.player_id.clone(),
Player {
name,
@@ -43,29 +39,28 @@ fn main() {
);
}
ActionKind::Move(velocity) => {
- if let Some(player) = game.players.get_mut(&action.player_id) {
+ if let Some(player) = game_state.players.get_mut(&action.player_id) {
player.position += velocity;
}
}
- ActionKind::Leave => {
- game.players.remove(&action.player_id);
- }
}
}
- // Only bother to send the latest game status to clients if we really need to, instead of
- // every tick
- if send_game_status {
- match handlers.lock() {
- Ok(mut h) => {
- let game_bytes = Codec::encode(&to_stdvec(&game).unwrap());
- for handler in h.iter() {
- let _ = handler.tx.send(game_bytes.clone());
- }
- h.extract_if(|handler| handler.thread.is_finished())
- .for_each(drop);
+ if let Ok(mut c) = clients.try_lock() {
+ // Clean up players associated with clients that have disconnected
+ for dropped_client in c.extract_if(|client| client.is_finished()) {
+ game_state.players.remove(&dropped_client.id());
+ send_game_state = true;
+ }
+
+ // Only bother to send the latest game status to clients if we really need to, instead
+ // of every tick
+ if send_game_state {
+ for client in c.iter() {
+ let _ = client.send(Response::UpdateGameState {
+ game_state: game_state.clone(),
+ });
}
- Err(_) => break,
}
}
diff --git a/server/src/net.rs b/server/src/net.rs
@@ -1,31 +1,59 @@
use std::{
collections::LinkedList,
- io::{self, Read, Write},
- net::SocketAddr,
sync::{
Arc, Mutex,
- mpsc::{Receiver, Sender, channel},
+ mpsc::{Receiver, SendError, Sender, channel},
},
thread::{self, JoinHandle},
time::Duration,
};
-use common::{codec::Codec, game::Action};
+use common::{
+ game::{Action, Response},
+ net::{handle_network_read, handle_network_write},
+};
use mio::{
Events, Interest, Poll, Token,
net::{TcpListener, TcpStream},
};
+use uuid::Uuid;
const TOKEN: Token = Token(0);
#[derive(Debug)]
-pub(crate) struct Handler {
- _addr: SocketAddr,
- pub thread: JoinHandle<()>,
- pub tx: Sender<Vec<u8>>,
+pub(crate) struct Client {
+ id: Uuid,
+ thread: JoinHandle<()>,
+ tx: Sender<Response>,
+}
+
+impl Client {
+ fn new(
+ stream: TcpStream,
+ client_to_main_tx: Sender<Action>,
+ main_to_client_tx: Sender<Response>,
+ main_to_client_rx: Receiver<Response>,
+ ) -> Self {
+ let new_client = Client {
+ id: Uuid::new_v4(),
+ thread: thread::spawn(|| handle_client(stream, main_to_client_rx, client_to_main_tx)),
+ tx: main_to_client_tx,
+ };
+ let _ = new_client.tx.send(Response::SetId { id: new_client.id });
+ new_client
+ }
+ pub(crate) fn id(&self) -> Uuid {
+ self.id
+ }
+ pub(crate) fn send(&self, t: Response) -> Result<(), SendError<Response>> {
+ self.tx.send(t)
+ }
+ pub(crate) fn is_finished(&self) -> bool {
+ self.thread.is_finished()
+ }
}
-pub(crate) fn listen(handlers: Arc<Mutex<LinkedList<Handler>>>, tx: Sender<Action>) {
+pub(crate) fn listen(clients: Arc<Mutex<LinkedList<Client>>>, tx: Sender<Action>) {
let mut poll = Poll::new().unwrap();
let mut events = Events::with_capacity(128);
let mut listener = TcpListener::bind("0.0.0.0:6767".parse().unwrap()).unwrap();
@@ -36,23 +64,18 @@ pub(crate) fn listen(handlers: Arc<Mutex<LinkedList<Handler>>>, tx: Sender<Actio
poll.poll(&mut events, None).unwrap();
for _ in events.iter() {
let (stream, _addr) = listener.accept().unwrap();
- let (main_to_handler_tx, main_to_handler_rx) = channel();
- if let Ok(addr) = stream.peer_addr() {
- let tx_clone = tx.clone();
- let thread =
- thread::spawn(|| handle_connection(stream, main_to_handler_rx, tx_clone));
- let handler = Handler {
- _addr: addr,
- thread,
- tx: main_to_handler_tx.clone(),
- };
- handlers.lock().unwrap().push_back(handler);
- }
+ let (main_to_client_tx, main_to_client_rx) = channel();
+ clients.lock().unwrap().push_back(Client::new(
+ stream,
+ tx.clone(),
+ main_to_client_tx,
+ main_to_client_rx,
+ ));
}
}
}
-fn handle_connection(mut stream: TcpStream, rx: Receiver<Vec<u8>>, tx: Sender<Action>) {
+fn handle_client(mut stream: TcpStream, rx: Receiver<Response>, tx: Sender<Action>) {
let poll = Poll::new().unwrap();
poll.registry()
.register(&mut stream, TOKEN, Interest::READABLE)
@@ -60,58 +83,9 @@ fn handle_connection(mut stream: TcpStream, rx: Receiver<Vec<u8>>, tx: Sender<Ac
let shared_stream = Arc::new(Mutex::new(stream));
let shared_stream_read = shared_stream.clone();
let shared_stream_write = shared_stream.clone();
- let read_thread = thread::spawn(|| handle_client_read(poll, shared_stream_read, tx));
- let write_thread = thread::spawn(|| handle_client_write(shared_stream_write, rx));
+ let read_thread = thread::spawn(|| handle_network_read(poll, shared_stream_read, tx));
+ let write_thread = thread::spawn(|| handle_network_write(shared_stream_write, rx));
while !read_thread.is_finished() && !write_thread.is_finished() {
thread::sleep(Duration::from_millis(500));
}
}
-
-fn handle_client_read(mut poll: Poll, stream: Arc<Mutex<TcpStream>>, tx: Sender<Action>) {
- let mut events = Events::with_capacity(128);
- let mut codec = Codec::default();
- let mut buffer = [0x00; 1024];
- let mut buffer_index;
- loop {
- poll.poll(&mut events, None).unwrap();
- for event in events.iter() {
- if event.is_readable() {
- let mut stream = stream.lock().unwrap();
- match stream.read(&mut buffer) {
- Ok(0) => return,
- Ok(i) => {
- buffer_index = i;
- let actions = codec.decode(&buffer[..buffer_index]);
- for action in actions {
- match postcard::from_bytes(action.as_slice()) {
- Ok(a) => {
- if let Err(_) = tx.send(a) {
- break;
- }
- }
- Err(_) => (),
- }
- }
- }
- Err(e) => match e.kind() {
- io::ErrorKind::WouldBlock => (),
- _ => return,
- },
- }
- }
- }
- }
-}
-
-fn handle_client_write(stream: Arc<Mutex<TcpStream>>, rx: Receiver<Vec<u8>>) {
- loop {
- match rx.recv() {
- Ok(bytes) => {
- if let Err(_) = stream.lock().unwrap().write_all(&bytes) {
- return;
- }
- }
- Err(_) => return,
- }
- }
-}
