From 68e7365432c251396b50078626838401a1129e0d Mon Sep 17 00:00:00 2001 From: szymon Date: Sun, 16 Aug 2026 12:53:15 +0200 Subject: [PATCH] feat: add server message loop --- src/network/connection.rs | 61 +++++++++++++++++++------ src/network/server.rs | 93 ++++++++++++++++++++++++++++++++++----- 2 files changed, 130 insertions(+), 24 deletions(-) diff --git a/src/network/connection.rs b/src/network/connection.rs index 2e2e62f..1d4cf5b 100644 --- a/src/network/connection.rs +++ b/src/network/connection.rs @@ -2,7 +2,8 @@ use std::io; use std::marker::PhantomData; use std::net::SocketAddr; -use log::{debug, error}; +use log::{debug, error, info}; +use tokio::sync::mpsc::Receiver; use tokio::{net::TcpStream, sync::mpsc::Sender}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -20,9 +21,11 @@ pub enum ConnectionType { pub struct Connection where Inbound: MessageLike, + Outbound: MessageLike, { conn_type: ConnectionType, - sender: Sender>, + sender: Sender<(SocketAddr, Message)>, + reciever: Option>>, tcp: Option, peer: Option, outbound_type: PhantomData, @@ -35,11 +38,12 @@ where { pub fn new( conn_type: ConnectionType, - sender: Sender>, + sender: Sender<(SocketAddr, Message)>, ) -> Connection { Connection { conn_type, sender, + reciever: None, tcp: None, peer: None, outbound_type: PhantomData, @@ -79,7 +83,13 @@ where } } - pub async fn send(&mut self, msg: Message) { + pub fn get_sender(&mut self) -> Sender> { + let (sender, reciever) = tokio::sync::mpsc::channel(1024); + self.reciever = Some(reciever); + sender + } + + async fn send(&mut self, msg: Message) { let size = msg.size; let content = msg.content; @@ -105,17 +115,38 @@ where } pub async fn start_listening(&mut self) { - loop { - let result = self.read_header().await; + let mut receiver = self + .reciever + .take() + .expect("Tried listening, but get_sender wasn't called"); - if let Err(e) = result { - error!( - "Error while reading data from connection with: {:?} {}. Connection is broken. Disconnecting...", - self.peer, e - ); - return; + loop { + tokio::select! { + result = self.read_header() => { + if let Err(e) = result { + error!( + "Connection broken with {:?}: {}. Disconnecting...", + self.peer, e + ); + break; + } + }, + + msg_opt = receiver.recv() => { + match msg_opt { + Some(msg) => { + self.send(msg).await; + }, + None => { + info!("Server closed the channel for {:?}. Disconnecting...", self.peer); + break; + } + } + } } } + + self.reciever = Some(receiver); } async fn read_header(&mut self) -> Result<(), io::Error> { @@ -151,7 +182,11 @@ where match new_msg { Ok(msg) => { - if let Err(e) = self.sender.send(msg).await { + if let Err(e) = self + .sender + .send((self.peer.expect("Couldn't get peer addr"), msg)) + .await + { error!("Erro while sending message to main thread: {}", e); } return Ok(()); diff --git a/src/network/server.rs b/src/network/server.rs index 41ad2f5..0f0718c 100644 --- a/src/network/server.rs +++ b/src/network/server.rs @@ -1,5 +1,8 @@ use log::{error, info}; -use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::{ + collections::HashMap, + net::{IpAddr, Ipv4Addr, SocketAddr}, +}; use tokio::{ net::{TcpListener, TcpStream}, sync::mpsc::{Receiver, Sender}, @@ -9,7 +12,10 @@ use crate::{ logic::game_master::GameMaster, network::{ connection::{Connection, ConnectionType}, - message::{ClientMessage, Message, ServerMessage}, + message::{ + ClientMessage::{self, AnswerHandshake, RequestCapture, RequestMove}, + Message, ServerMessage, + }, }, }; @@ -22,9 +28,9 @@ pub struct ServerSettings { pub struct Server { listener: TcpListener, - connections: Vec>, - reciever: Receiver>, - sender: Sender>, + connections: HashMap>>, + reciever: Receiver<(SocketAddr, Message)>, + sender: Sender<(SocketAddr, Message)>, game_master: Option, settings: ServerSettings, } @@ -38,7 +44,8 @@ impl Server { allow_spectators: false, }; - let (sender, mut reciever) = tokio::sync::mpsc::channel::>(1024); // random + let (sender, mut reciever) = + tokio::sync::mpsc::channel::<(SocketAddr, Message)>(1024); // random // number Server { @@ -46,7 +53,7 @@ impl Server { .await .expect("Could not start the server"), // TODO! Handle this nicely, as this will // crash gui app - connections: vec![], + connections: HashMap::new(), sender, reciever, game_master: None, @@ -80,13 +87,77 @@ impl Server { Connection::new(ConnectionType::Client, self.sender.clone()); conn.delegate(socket, addr); - self.connections.push(conn); + let sender = conn.get_sender(); + + self.connections.insert(addr, sender); + + self.listen_client(conn); } - fn request_handshake(&self) { - let conn = self.connections.last().unwrap(); - info!("Requesting handshake from: {:?}", conn); + fn listen_client(&mut self, mut conn: Connection) { + tokio::spawn(async move { + conn.start_listening().await; + }); + } + /// Process incoming messages + pub async fn update(&mut self) { + let Some(oldest_message) = self.reciever.recv().await else { + return; + }; + + let (addr, msg) = oldest_message; + + match msg.content { + ClientMessage::AnswerHandshake { player_name } => { + self.process_handshake(); + } + ClientMessage::RequestCapture { capture_path } => { + self.process_capture(); + } + ClientMessage::RequestMove { from, to } => { + self.process_move(); + } + } + } + + /* ======= Processing messages :D helper functions ======== */ + + fn process_handshake(&self) { todo!(); } + + fn process_capture(&self) { + todo!(); + } + + fn process_move(&self) { + todo!(); + } + + async fn request_handshake(&mut self, addr: SocketAddr) { + let msg = Message::new(ServerMessage::RequestHandshake); + self.send_message(addr, msg).await; + } + + async fn send_message( + &mut self, + addr: SocketAddr, + msg: Result, postcard::Error>, + ) { + let conn_opt = self.connections.get_mut(&addr); + + let Some(conn) = conn_opt else { + error!("Could not get connection from connections list"); + return; + }; + + let _ = match msg { + Err(e) => { + error!("There was an error while creating message: {}", e); + return; + } + Ok(msg) => conn.send(msg).await, + }; + } }