feat: add server message loop

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<Inbound, Outbound>
 where
     Inbound: MessageLike,
+    Outbound: MessageLike,
 {
     conn_type: ConnectionType,
-    sender: Sender<Message<Inbound>>,
+    sender: Sender<(SocketAddr, Message<Inbound>)>,
+    reciever: Option<Receiver<Message<Outbound>>>,
     tcp: Option<TcpStream>,
     peer: Option<SocketAddr>,
     outbound_type: PhantomData<Outbound>,
@@ -35,11 +38,12 @@ where
 {
     pub fn new(
         conn_type: ConnectionType,
-        sender: Sender<Message<Inbound>>,
+        sender: Sender<(SocketAddr, Message<Inbound>)>,
     ) -> Connection<Inbound, Outbound> {
         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<Outbound>) {
+    pub fn get_sender(&mut self) -> Sender<Message<Outbound>> {
+        let (sender, reciever) = tokio::sync::mpsc::channel(1024);
+        self.reciever = Some(reciever);
+        sender
+    }
+
+    async fn send(&mut self, msg: Message<Outbound>) {
         let size = msg.size;
         let content = msg.content;
 
@@ -105,17 +115,38 @@ where
     }
 
     pub async fn start_listening(&mut self) {
-        loop {
+        let mut receiver = self
-            let result = self.read_header().await;
+            .reciever
+            .take()
+            .expect("Tried listening, but get_sender wasn't called");
 
-            if let Err(e) = result {
+        loop {
-                error!(
+            tokio::select! {
-                    "Error while reading data from connection with: {:?} {}. Connection is broken. Disconnecting...",
+                result = self.read_header() => {
-                    self.peer, e
+                    if let Err(e) = result {
-                );
+                        error!(
-                return;
+                            "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(());

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<Connection<ClientMessage, ServerMessage>>,
+    connections: HashMap<SocketAddr, Sender<Message<ServerMessage>>>,
-    reciever: Receiver<Message<ClientMessage>>,
+    reciever: Receiver<(SocketAddr, Message<ClientMessage>)>,
-    sender: Sender<Message<ClientMessage>>,
+    sender: Sender<(SocketAddr, Message<ClientMessage>)>,
     game_master: Option<GameMaster>,
     settings: ServerSettings,
 }
@@ -38,7 +44,8 @@ impl Server {
             allow_spectators: false,
         };
 
-        let (sender, mut reciever) = tokio::sync::mpsc::channel::<Message<ClientMessage>>(1024); // random
+        let (sender, mut reciever) =
+            tokio::sync::mpsc::channel::<(SocketAddr, Message<ClientMessage>)>(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) {
+    fn listen_client(&mut self, mut conn: Connection<ClientMessage, ServerMessage>) {
-        let conn = self.connections.last().unwrap();
+        tokio::spawn(async move {
-        info!("Requesting handshake from: {:?}", conn);
+            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<Message<ServerMessage>, 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,
+        };
+    }
 }