made clients not have their own channel

SubAt0m1c · SubAt0m1c · commit 43a836b49292 · 2025-10-14T15:59:08.000-04:00
diff --git a/server/src/network/client.rs b/server/src/network/client.rs
@@ -1,112 +1,107 @@
-use crate::network::binary::var_int::{peek_var_int, VarInt};
-use crate::network::connection_state::ConnectionState;
-use crate::network::connection_state::ConnectionState::*;
-use crate::network::internal_packets::{ClientHandlerMessage, MainThreadMessage, NetworkThreadMessage};
-use crate::network::packets::packet_buffer::PacketBuffer;
-use crate::network::packets::packet_deserialize::PacketDeserializable;
-use crate::network::protocol::handshake::serverbound::Handshake;
-use crate::network::protocol::login::clientbound::LoginSuccess;
-use crate::network::protocol::login::serverbound::LoginStart;
-use crate::network::protocol::play::serverbound::Play;
-use crate::network::protocol::status::clientbound::{StatusPong, StatusResponse};
-use crate::network::protocol::status::serverbound::StatusPing;
-use crate::types::status::StatusBytes;
-use crate::player::player::{ClientId, GameProfile};
-use crate::GameProfileProperty;
+use std::collections::HashMap;
+
 use anyhow::bail;
 use bytes::{Buf, Bytes, BytesMut};
 use fstr::FString;
-use std::collections::HashMap;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::TcpStream;
-use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+use tokio::{io::{self, AsyncReadExt, AsyncWriteExt}, net::{TcpStream, tcp::{OwnedReadHalf, OwnedWriteHalf}}, sync::mpsc::UnboundedSender, task::JoinHandle};
 use uuid::Uuid;
 
+use crate::{ClientId, GameProfile, GameProfileProperty, network::{binary::var_int::{VarInt, peek_var_int}, connection_state::ConnectionState, internal_packets::{MainThreadMessage, NetworkThreadMessage}, packets::{packet_buffer::PacketBuffer, packet_deserialize::PacketDeserializable}, protocol::{handshake::serverbound::Handshake, login::{clientbound::LoginSuccess, serverbound::LoginStart}, play::serverbound::Play, status::{clientbound::{StatusPong, StatusResponse}, serverbound::StatusPing}}}, types::status::StatusBytes};
+
 #[derive(Debug, Clone)]
 pub struct Client {
     pub id: ClientId,
     pub connection_state: ConnectionState,
     // pub game_profile: Option<GameProfile>,
 }
 
+pub struct ClientHandler {
+    writer: OwnedWriteHalf,
+    handle: JoinHandle<()>
+}
 
-// main thread tx errors can be ignored since the network/client threads will close eachother properly regardless of client status, 
-// which will in turn close the client handlers.
+impl ClientHandler {
+    pub fn spawn(
+        client_id: ClientId, 
+        socket: TcpStream, 
+        network_tx: UnboundedSender<NetworkThreadMessage>, 
+        main_tx: UnboundedSender<MainThreadMessage>,        
+        status: StatusBytes
+    ) -> Self {
+        let (read, writer) = socket.into_split();
+        let handle = tokio::spawn(run_client(client_id, read, network_tx, main_tx, status));
+        Self { writer, handle }
+    }
+    
+    pub async fn send(&mut self, data: &Bytes) -> Result<(), io::Error> {
+        self.writer.write_all(data).await
+    }
+    
+    // this does not inform the network thread about the read tasks shutdown.
+    pub async fn disconnect(&mut self) -> Result<(), io::Error> {
+        self.writer.shutdown().await?;
+        self.handle.abort();
+        Ok(())
+    }
+    
+    pub fn abort(&self) {
+        self.handle.abort();
+    }
+}
 
-pub async fn handle_client(
-    client_id: ClientId,
-    mut socket: TcpStream,
-    tx: UnboundedSender<ClientHandlerMessage>,
-    mut rx: UnboundedReceiver<ClientHandlerMessage>,
+async fn run_client(
+    client_id: ClientId, 
+    mut reader: OwnedReadHalf, 
+    network_tx: UnboundedSender<NetworkThreadMessage>, 
     main_tx: UnboundedSender<MainThreadMessage>,
-    network_tx: UnboundedSender<NetworkThreadMessage>,
     status: StatusBytes
 ) {
     let mut client = Client {
         id: client_id,
-        connection_state: Handshaking,
+        connection_state: ConnectionState::Handshaking,
         // game_profile: None,
     };
     let mut bytes = BytesMut::new();
-
+    
     loop {
-        tokio::select! {
-            result = socket.read_buf(&mut bytes) => {
-                match result {
-                    Ok(0) => break, // channel closed normally
-                    Ok(_) => {
-                        // channel closes, if client sends invalid packets.
-                        if let Err(err) = read_packets(
-                            &mut bytes,
-                            &mut client,
-                            &tx,
-                            &main_tx,
-                            &status
-                        ).await {
-                            eprintln!("client {client_id:?} errored: {err}");
-                            break;
-                        }
-                    },
-                    Err(e) => {
-                        eprintln!("Client {client_id:?} read error: {e}");
-                        break;
-                    }
+        let res: Result<usize, io::Error> = reader.read_buf(&mut bytes).await;
+        match res {
+            Ok(0) => break,
+            Ok(_) => {
+                if let Err(err) = read_packets(
+                    &mut bytes,
+                    &mut client,
+                    &network_tx,
+                    &main_tx,
+                    &status
+                ).await {
+                    eprintln!("client {client_id:?} errored: {err}");
+                    break;
                 }
             }
-
-            Some(message) = rx.recv() => {
-                match message {
-                    ClientHandlerMessage::Send(data) => {
-                        if let Err(e) = socket.write_all(&data).await {
-                            eprintln!("write error: {e}");
-                            break
-                        }
-                    }
-                    ClientHandlerMessage::CloseHandler => break,
-                }
+            Err(e) => {
+                eprintln!("Client {client_id:?} read error: {e}");
+                break;
             }
         }
     }
-
-    if client.connection_state == ConnectionState::Play {
-        let _ = network_tx.send(NetworkThreadMessage::ConnectionClosed { client_id });
-    }
-    println!("handle client for {client_id:?} closed.");
+    println!("client reader closed");
+    let _ = network_tx.send(NetworkThreadMessage::ConnectionClosed { client_id, connection_state: client.connection_state });
 }
 
 async fn read_packets(
-    buffer: &mut BytesMut,
-    client: &mut Client,
-    client_tx: &UnboundedSender<ClientHandlerMessage>,
+    buffer: &mut BytesMut, 
+    client: &mut Client, 
+    network_tx: &UnboundedSender<NetworkThreadMessage>,
     main_tx: &UnboundedSender<MainThreadMessage>,
     status: &StatusBytes
 ) -> anyhow::Result<()> {
     while let Some(mut buffer) = try_read_packet_slice(buffer) {
         match client.connection_state {
-            Handshaking => handle_handshake(&mut buffer, client)?,
-            Status => handle_status(&mut buffer, client_tx, status)?,
-            Login => handle_login(&mut buffer, client, client_tx, main_tx)?,
-            Play => {
+            ConnectionState::Handshaking => handle_handshake(&mut buffer, client)?,
+            ConnectionState::Status => handle_status(client.id, &mut buffer, network_tx, status)?,
+            ConnectionState::Login => handle_login(&mut buffer, client, network_tx, main_tx)?,
+            ConnectionState::Play => {
                 let packet = Play::read(&mut buffer)?;
                 if let Play::Invalid(packet_id) = packet {
                     eprintln!("invalid packet: 0x{packet_id:02x}");
@@ -136,29 +131,20 @@ fn try_read_packet_slice(buf: &mut impl Buf) -> Option<Bytes> {
     Some(buf.copy_to_bytes(packet_len))
 }
 
-pub enum HandleResult {
-    Ok,
-    Disconnect,
-}
-
 fn handle_handshake(buffer: &mut impl Buf, client: &mut Client) -> anyhow::Result<()> {
-    if let Ok(packet_id) = VarInt::read(buffer) {
-        match *packet_id {
-            0x00 => {
-                let handshake = Handshake::read(buffer)?;
-                client.connection_state = ConnectionState::from_id(handshake.next_state.0)?;
-                Ok(())
-            },
-            _ => bail!("Unknown packet id during handshake")
+    match *VarInt::read(buffer)? {
+        0x00 => {
+            let handshake = Handshake::read(buffer)?;
+            client.connection_state = ConnectionState::from_id(handshake.next_state.0)?;
         }
-    } else {
-        bail!("Failed to read var_int packet id.")
+        _ => bail!("Unknown pack id during handshake."),
     }
+    Ok(())
 }
-
 fn handle_status(
+    client_id: ClientId,
     buffer: &mut impl Buf,
-    client_tx: &UnboundedSender<ClientHandlerMessage>,
+    network_tx: &UnboundedSender<NetworkThreadMessage>,
     status: &StatusBytes
 ) -> anyhow::Result<()> {
     let packet_id = *VarInt::read(buffer)?;
@@ -177,20 +163,19 @@ fn handle_status(
         }
         _ => bail!("Unknown packet id during status")
     }
-    // instead of sending to network tx,
-    // could just write directly to socket?
-    client_tx.send(ClientHandlerMessage::Send(packet_buffer.split_into_bytes()))?;
+    
+    // we cannot directly send to socket since the read task cant own the write half, at least not without Arc<Mutex<>> i think
+    network_tx.send(NetworkThreadMessage::SendPackets { client_id, buffer: packet_buffer.split_into_bytes() })?;
     Ok(())
 }
 
 const FLAME_OF_WAR: &str = "ewogICJ0aW1lc3RhbXAiIDogMTc1OTQzODI1MzM2OCwKICAicHJvZmlsZUlkIiA6ICI4YTdhZDkyMzc3MjI0ZjIyOGMwNDI4Y2I1YmQ5NzJkYSIsCiAgInByb2ZpbGVOYW1lIiA6ICJGbGFtZU9mV2FyIiwKICAic2lnbmF0dXJlUmVxdWlyZWQiIDogdHJ1ZSwKICAidGV4dHVyZXMiIDogewogICAgIlNLSU4iIDogewogICAgICAidXJsIiA6ICJodHRwOi8vdGV4dHVyZXMubWluZWNyYWZ0Lm5ldC90ZXh0dXJlL2JiNDg4Njc1YjMxYTQyZTc5MDI0ZGUzOGY1YmQ3ODZhMzlmNzVhMmE2ZGJhMDk0NDc5MmQ0NDNjNjA1ZDE4ZjkiCiAgICB9CiAgfQp9";
 const FLAME_OF_WAR_SIG: &str = "UvRQflcS0w4KTJSN+fpqYxVBTwo6wb66JMp6seThrmSGwUmbPfs8WEK2TPBIcipG0kBjWWdDMUpXFZ5YMBshnb7kHh588oPeL0gja/m9yHGEgtfucyqudL3m4sq3iZnJbdO3yKnF/00WqelBI5fZ3zc9SDyAjLUL4QHIXPm4U/z3UH1ZnVjGc5bZbV7qXILw7pF00al8ks1kpOUeds8zjSpVMRMTF9WQww89jNjbpvzcKP97KOOBXPJB1cuTUi3DEe3/9omZhcfgDyZDDJkmF3hTVZx1ijKtknlKRJqFcUEmsL1XUgRxqLSYNt1D1XCjEJeWAyT5YDVtvuj3Oa/zEeWQa9WVSXaUTGpVpQBRJrTJmtLH4O4hDMz4j7M2T0lsbOg7sIqvWVRvmKptKlLWKSWk8tlYXrx+Ef4YN5iva8/xhnKZmfe/JmT8uIKtNiv8Zcrj1WXasJ4wz0JCEQBOJDJXnEU548Sk1nxAcmX/W8jHkMnXArE3LKkLdxD7e++Hw60pv3GcyvTou5Mlrmgo6rHk188Li4CU826i+z0OuodRtdY+vsQIoFWLnnHu4HdqKA3IevcV7+Gl3FDzbzPXiSbUmSAV4drpLELTTPMnhhvMK85zS8138LTuScBiFRKVaSuXZJS7UIJ6VtjYK+iEuVblN9BJihP2NiuubCeL484=";
 
-
 fn handle_login(
     buffer: &mut impl Buf,
     client: &mut Client,
-    client_tx: &UnboundedSender<ClientHandlerMessage>,
+    network_tx: &UnboundedSender<NetworkThreadMessage>,
     main_tx: &UnboundedSender<MainThreadMessage>,
 ) -> anyhow::Result<()> {
     let packet_id = *VarInt::read(buffer)?;
@@ -222,7 +207,7 @@ fn handle_login(
                 uuid: uuid.hyphenated().to_string(),
                 name: game_profile.username.clone(),
             });
-            client_tx.send(ClientHandlerMessage::Send(packet_buffer.split_into_bytes()))?;
+            network_tx.send(NetworkThreadMessage::SendPackets { client_id: client.id, buffer: packet_buffer.split_into_bytes() })?;
             
             main_tx.send(MainThreadMessage::NewPlayer {
                 client_id: client.id,
diff --git a/server/src/network/internal_packets.rs b/server/src/network/internal_packets.rs
@@ -1,3 +1,4 @@
+use crate::network::connection_state::ConnectionState;
 use crate::network::protocol::play::serverbound::Play;
 use crate::player::player::{ClientId, GameProfile};
 use crate::types::status::StatusUpdate;
@@ -12,26 +13,21 @@ pub enum NetworkThreadMessage {
     },
     
     /// received when the client's handler is closed.
-    /// sends a client disconnected message to the main thread
+    /// sends a client disconnected message to the main thread if the connect state is play
     ConnectionClosed {
         client_id: ClientId,
+        connection_state: ConnectionState,
     },
 
     /// Disconnects the client from the server.
-    /// This sends a close handler message to the client's handler.
+    /// This aborts the client's read task and drops the write half.
     /// It should be sent after the vanilla disconnect packet is sent.
     /// the main thread should wait for a ClientDisconnected response to handle actually removing the player.
     DisconnectClient {
         client_id: ClientId,
     },
 }
 
-pub enum ClientHandlerMessage {
-    Send(Bytes),
-    /// Closes the handler for this client. This then sends a connection closed message to the network thread.
-    CloseHandler,
-}
-
 pub enum MainThreadMessage {
     PacketReceived {
         client_id: ClientId,
diff --git a/server/src/network/network.rs b/server/src/network/network.rs
@@ -1,17 +1,18 @@
-use crate::network::client::handle_client;
+use crate::network::client::{ClientHandler};
+use crate::network::connection_state::ConnectionState;
 use crate::types::status::Status;
 use core::panic;
 use std::collections::HashMap;
 use tokio::net::TcpListener;
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
 
-use crate::network::internal_packets::{ClientHandlerMessage, MainThreadMessage, NetworkThreadMessage};
+use crate::network::internal_packets::{MainThreadMessage, NetworkThreadMessage};
 use crate::player::player::ClientId;
 
 type Sender<T> = UnboundedSender<T>;
 type Receiver<T> = UnboundedReceiver<T>;
 
-type ClientMap = HashMap<ClientId, UnboundedSender<ClientHandlerMessage>>;
+type ClientMap = HashMap<ClientId, ClientHandler>;
 
 pub fn start_network(
     ip: &'static str,
@@ -45,20 +46,9 @@ async fn run_network_thread(
                 let Ok((socket, _)) = result else { continue };
                 let client_id: ClientId = client_id_counter;
                 client_id_counter += 1;
-
-                // do we need tx and rx for each client?
-                //
-                let (client_tx, client_rx) = unbounded_channel::<ClientHandlerMessage>();
-                clients.insert(client_id, client_tx.clone());
-                tokio::spawn(handle_client(
-                    client_id,
-                    socket,
-                    client_tx,
-                    client_rx,
-                    main_tx.clone(),
-                    network_tx.clone(),
-                    status.get()
-                ));
+                
+                let handler = ClientHandler::spawn(client_id, socket, network_tx.clone(), main_tx.clone(), status.get());
+                clients.insert(client_id, handler);
             }
 
             // this can never be none since this function owns a network_tx.
@@ -68,24 +58,30 @@ async fn run_network_thread(
                 match msg { 
                     NetworkThreadMessage::UpdateStatus(update) => status.set(update),
                     NetworkThreadMessage::SendPackets { client_id, buffer } => {
-                        if let Some(client_tx) = clients.get(&client_id) {
-                            if let Err(e) = client_tx.send(ClientHandlerMessage::Send(buffer)) {
-                                eprintln!("Client {} handler dropped its reciever: {}", client_id, e);
+                        if let Some(handler) = clients.get_mut(&client_id) {
+                            if let Err(e) = handler.send(&buffer).await {
+                                eprintln!("Client {} handler failed to send: {}", client_id, e);
                                 disconnect_client(client_id, &main_tx, &mut clients);
                             }
                         }
                     }
             
                     NetworkThreadMessage::DisconnectClient { client_id } => {
-                        if let Some(client_tx) = clients.get(&client_id) {
-                            if let Err(e) = client_tx.send(ClientHandlerMessage::CloseHandler) {
-                                eprintln!("Client {} handler dropped its reciever: {}", client_id, e);
-                                disconnect_client(client_id, &main_tx, &mut clients);
+                        if let Some(handler) = clients.get_mut(&client_id) {
+                            if let Err(e) = handler.disconnect().await {
+                                eprintln!("Client {} writer failed to shutdown: {}", client_id, e);
                             }
+                            disconnect_client(client_id, &main_tx, &mut clients);
                         }
                     }
             
-                    NetworkThreadMessage::ConnectionClosed { client_id } => disconnect_client(client_id, &main_tx, &mut clients),
+                    NetworkThreadMessage::ConnectionClosed { client_id, connection_state } => {
+                        if connection_state == ConnectionState::Play {
+                            // we probably shouldnt tell the main thread a client it never added got disconnected? 
+                            main_tx.send(MainThreadMessage::ClientDisconnected { client_id }).expect("Main thread should never drop its network reciever.");
+                        }
+                        clients.remove(&client_id);
+                    }
                 }
             }
         }
