Revamp the NetworkClient-Server relationship

Cargo.lock

@@ -104,6 +104,7 @@ dependencies = [
  "serde_json",
  "tokio",
  "toml",
+ "uuid",
 ]
 
 [[package]]
@@ -586,6 +587,12 @@ version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f7fe0bb3479651439c9112f72b6c505038574c9fbb575ed1bf3b797fa39dd564"
 
+[[package]]
+name = "uuid"
+version = "0.8.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bc5cf98d8186244414c848017f0e2676b3fcb46807f6668a97dfe67359a3c4b7"
+
 [[package]]
 name = "wasi"
 version = "0.10.0+wasi-snapshot-preview1"

Cargo.toml

@@ -19,6 +19,7 @@ async-trait = "0.1.48"
 lazy_static = "1.4.0"
 ctrlc = "3.1.8"
 futures = "0.3.13"
+uuid = "0.8.2"
 # colorful = "0.2.1"
 # ozelot = "0.9.0" # Ozelot 0.9.0 supports protocol version 578 (1.15.2)
 # toml = "0.5.6"

src/entity/player.rs

@@ -1,8 +1,38 @@
-use crate::server::net::NetworkClient;
 use crate::world::location::Location;
+use uuid::Uuid;
 
+#[derive(Debug, Clone, PartialEq)]
 pub struct Player {
-    position: Location,
+    pub position: Location,
-    display_name: String,
+    pub uuid: Uuid,
-    connection: NetworkClient,
+    names: (String, Option<String>),
+}
+impl Player {
+    pub fn new() -> Player {
+        Player {
+            position: Location::zero(),
+            uuid: Uuid::nil(),
+            names: ("Player".into(), None),
+        }
+    }
+    pub fn username(&self) -> &String {
+        &self.names.0
+    }
+    pub fn username_mut(&mut self) -> &mut String {
+        &mut self.names.0
+    }
+    pub fn display_name(&self) -> &String {
+        match &self.names.1 {
+            Some(name) => name,
+            None => &self.names.0,
+        }
+    }
+    pub fn display_name_mut(&mut self) -> &mut Option<String> {
+        &mut self.names.1
+    }
+}
+impl Default for Player {
+    fn default() -> Self {
+        Player::new()
+    }
 }

src/mctypes/numbers.rs

@@ -636,26 +636,28 @@ pub struct MCVarInt {
 impl MCVarInt {
     pub async fn read(t: &mut TcpStream) -> tokio::io::Result<MCVarInt> {
         let mut num_read = 0;
-        let mut result = 0i32;
+        let mut result = 0i128;
         let mut read = 0u8;
         let mut run_once = false;
         while (read & 0b10000000) != 0 || !run_once {
             run_once = true;
             read = read_byte(t).await?;
-            let value = (read & 0b01111111) as i32;
+            let value = (read & 0b01111111) as i128;
             result |= value << (7 * num_read);
             num_read += 1;
             if num_read > 5 {
                 return Err(io_error("MCVarInt is too big"));
             }
         }
-        Ok(MCVarInt { value: result })
+        Ok(MCVarInt {
+            value: result as i32,
+        })
     }
     pub fn from_bytes(_v: Vec<u8>) -> MCVarInt {
         unimplemented!()
     }
     pub fn to_bytes(&self) -> Vec<u8> {
-        unimplemented!()
+        Into::<Vec<u8>>::into(*self)
     }
 }
 impl From<u8> for MCVarInt {

src/server/messages.rs

@@ -5,10 +5,7 @@ pub enum ServerboundMessage {
 }
 
 #[derive(Debug, PartialEq, Clone)]
-pub enum ClientboundMessage {}
+pub enum ClientboundMessage {
-
-#[derive(Debug, PartialEq, Clone)]
-pub enum BroadcastMessage {
     Chat(String),       // The chat message.
     Disconnect(String), // The reason for disconnecting.
 }

src/server/mod.rs

@@ -4,25 +4,28 @@ pub mod messages;
 pub mod net;
 
 use crate::entity::player::Player;
+use crate::{mctypes::*, CONFIG, FAVICON};
 use log::*;
 use messages::*;
-use net::*;
+use net::{
+    packets::{self, Packet, PacketCommon},
+    *,
+};
+use serde_json::json;
 use std::sync::mpsc::{self, Receiver, TryRecvError};
+use std::time::Duration;
 use tokio::net::{TcpListener, ToSocketAddrs};
 
 /// The struct containing all the data and running all the updates.
 pub struct Server {
     network_clients: Vec<NetworkClient>,
     network_receiver: Receiver<NetworkClient>,
-    message_receiver: Receiver<ServerboundMessage>,
-    // message_sender: Bus<BroadcastMessage>,
     pub players: Vec<Player>,
 }
 impl Server {
     pub fn new<A: 'static + ToSocketAddrs + Send>(addr: A) -> Server {
         let (network_client_tx, network_client_rx) = mpsc::channel();
-        let (serverbound_message_tx, serverbound_message_rx) = mpsc::channel();
+        let (serverbound_message_tx, _serverbound_message_rx) = mpsc::channel();
-        // let mut broadcast_message_tx = Bus::new(1_000); // Hold up to 1,000 messages in the queue.
         tokio::task::spawn(async move {
             let listener = TcpListener::bind(addr)
                 .await
@@ -38,7 +41,6 @@ impl Server {
                         stream,
                         id as u128,
                         serverbound_message_tx.clone(),
-                        // broadcast_message_tx.add_rx(),
                     ))
                     .expect("Network receiver disconnected");
                 id += 1;
@@ -48,7 +50,7 @@ impl Server {
         Server {
             network_receiver: network_client_rx,
             network_clients: vec![],
-            message_receiver: serverbound_message_rx,
+            // message_receiver: serverbound_message_rx,
             players: vec![],
         }
     }
@@ -61,82 +63,210 @@ impl Server {
             "Server shutting down. Uptime: {:?}",
             crate::START_TIME.elapsed()
         );
-        self.broadcast_message(BroadcastMessage::Disconnect(
+        self.broadcast_message(ClientboundMessage::Disconnect(
             "The server is shutting down".into(),
         ))
         .await;
     }
 
-    /// Update the network server.
+    pub fn num_players(&self) -> usize {
-    ///
+        let mut num = 0;
-    /// Update each client in `self.network_clients`.
+        for client in &self.network_clients {
-    async fn update_network(&mut self) -> tokio::io::Result<()> {
+            if client.state == NetworkClientState::Play {
-        // Read new clients from the network.
+                num += 1;
+            }
+        }
+        num
+    }
+
+    /// Send a `ClientboundMessage` to all connected network clients.
+    pub async fn broadcast_message(&mut self, message: ClientboundMessage) {
+        let mut v = Vec::new();
+        for client in self.network_clients.iter_mut() {
+            v.push(client.handle_message(message.clone()));
+        }
+        futures::future::join_all(v).await;
+    }
+
+    /// Get a client from their id.
+    pub fn client_from_id(&mut self, client_id: u128) -> Option<&mut NetworkClient> {
+        // Find the client based on id.
+        let mut client_index = -1isize;
+        for (i, c) in self.network_clients.iter().enumerate() {
+            if c.id == client_id {
+                client_index = i as isize;
+                break;
+            }
+        }
+        if client_index == -1 {
+            return None;
+        }
+        Some(&mut self.network_clients[client_index as usize])
+    }
+
+    /// Update the game server.
+    pub async fn update(&mut self) -> tokio::io::Result<()> {
+        // Get new clients from the network listener thread.
         loop {
             match self.network_receiver.try_recv() {
                 Ok(client) => {
                     info!(
                         "Got client at {}",
-                        client.stream.peer_addr().expect("Could not get peer addr")
+                        client.stream.peer_addr().expect("could not get peer addr")
                     );
                     self.network_clients.push(client)
                 }
                 Err(TryRecvError::Empty) => break,
-                Err(TryRecvError::Disconnected) => panic!("Network sender disconnected"),
+                Err(TryRecvError::Disconnected) => panic!("network sender disconnected"),
             }
         }
-        // Count the number of players in the Play state.
+        // Read new packets from each client.
-        let num_players = self.network_clients.iter().fold(0, |acc, nc| {
-            if nc.state == NetworkClientState::Play {
-                acc + 1
-            } else {
-                acc
-            }
-        });
-        // Update each client, disconnecting those with errors.
         for client in self.network_clients.iter_mut() {
-            if client.update(num_players).await.is_err() {
+            // Update and ignore errors.
+            if client.update().await.is_err() {
+                client.force_disconnect();
+            }
+        }
+        // Update the client and server according to each packet.
+        let mut packets = vec![];
+        for client in self.network_clients.iter_mut() {
+            while let Some(packet) = client.read_packet() {
+                packets.push((client.id, packet));
+            }
+        }
+        for (client_id, packet) in packets {
+            if self.handle_packet(client_id, packet).await.is_err() {
+                self.client_from_id(client_id).unwrap().force_disconnect();
+            }
+        }
+        // Disconnect clients when necessary.
+        for client in self.network_clients.iter_mut() {
+            if client.state == NetworkClientState::Disconnected {
+                client.force_disconnect();
+            } else if client.last_keep_alive.elapsed() > Duration::from_secs(20) {
+                debug!("Disconnecting client for timing out");
+                client.state = NetworkClientState::Disconnected;
                 client.force_disconnect();
             }
         }
         // Remove disconnected clients.
         self.network_clients
             .retain(|nc| nc.state != NetworkClientState::Disconnected);
-        // Read new messages from the clients.
+        Ok(())
-        loop {
-            match self.message_receiver.try_recv() {
-                Ok(message) => match message {
-                    ServerboundMessage::Chat(msg) => {
-                        self.broadcast_message(BroadcastMessage::Chat(msg)).await;
     }
-                    ServerboundMessage::PlayerJoin(_uuid, username) => {
+
-                        self.broadcast_message(BroadcastMessage::Chat(format!(
+    /// Handle a packet.
-                            "Welcome {} to the server!",
+    pub async fn handle_packet<P: PacketCommon>(
-                            username
+        &mut self,
+        client_id: u128,
+        packet: P,
+    ) -> tokio::io::Result<()> {
+        let num_players = self.num_players();
+        let mut client = self.client_from_id(client_id).unwrap();
+        match packet.as_packet() {
+            // Handshaking.
+            Packet::Handshake(handshake) => {
+                if handshake.next_state == 1 {
+                    client.state = NetworkClientState::Status;
+                } else if handshake.next_state == 2 {
+                    client.state = NetworkClientState::Login;
+                } else {
+                    client.state = NetworkClientState::Disconnected;
+                }
+                if handshake.protocol_version != 47 {
+                    let mut logindisconnect = packets::LoginDisconnect::new();
+                    logindisconnect.reason = MCChat {
+                        text: MCString::from("Incompatible client! Server is on 1.8.9"),
+                    };
+                    client.send_packet(logindisconnect).await?;
+                    client.state = NetworkClientState::Disconnected;
+                }
+            }
+            // Status.
+            Packet::StatusRequest(_statusrequest) => {
+                let mut statusresponse = packets::StatusResponse::new();
+                statusresponse.json_response = json!({
+                    "version": {
+                        "name": "1.8.9",
+                        "protocol": 47,
+                    },
+                    "players": {
+                        "max": CONFIG.max_players,
+                        "online": num_players,
+                        "sample": [
+                            {
+                                "name": "shvr",
+                                "id": "e3f58380-60bb-4714-91f2-151d525e64aa"
+                            }
+                        ]
+                    },
+                    "description": {
+                        "text": CONFIG.motd
+                    },
+                    "favicon": format!("data:image/png;base64,{}", if FAVICON.is_ok() { radix64::STD.encode(FAVICON.as_ref().unwrap().as_slice()) } else { "".to_owned() })
+                })
+                .to_string()
+                .into();
+                client.send_packet(statusresponse).await?;
+            }
+            Packet::StatusPing(statusping) => {
+                let mut statuspong = packets::StatusPong::new();
+                statuspong.payload = statusping.payload;
+                client.send_packet(statuspong).await?;
+            }
+            // Login.
+            Packet::LoginStart(loginstart) => {
+                client.player = Some(crate::entity::player::Player::new());
+                let player = client.player.as_mut().unwrap();
+                *player.username_mut() = loginstart.player_name.into();
+                // Offline mode skips encryption and compression.
+                // TODO: Encryption and compression
+                let mut loginsuccess = packets::LoginSuccess::new();
+                // We're in offline mode, so this is a temporary uuid.
+                // TODO: Get uuid and username from Mojang servers.
+                loginsuccess.uuid = player.uuid.clone().to_hyphenated().to_string().into();
+                loginsuccess.username = player.username().clone().into();
+                client.send_packet(loginsuccess).await?;
+                client.state = NetworkClientState::Play;
+                client.send_packet(packets::JoinGame::new()).await?;
+            }
+            Packet::ClientSettings(_clientsettings) => {
+                // TODO: Handle the packet.
+                client.send_packet(packets::HeldItemChange::new()).await?;
+                client
+                    .send_packet(packets::ClientboundPlayerPositionAndLook::new())
+                    .await?;
+                client.send_packet(packets::SpawnPosition::new()).await?;
+            }
+            // Play.
+            Packet::KeepAlivePong(_keepalivepong) => {
+                // TODO: Handle the packet.
+            }
+            Packet::ServerboundChatMessage(chatmessage) => {
+                let player_name = client.player.as_ref().unwrap().username().clone();
+                info!("<{}> {}", player_name, chatmessage.text);
+                self.broadcast_message(ClientboundMessage::Chat(format!(
+                    "<{}> {}",
+                    player_name, chatmessage.text
                 )))
                 .await;
+                // TODO: Handle the packet.
             }
-                },
+            Packet::Player(_player) => {
-                Err(TryRecvError::Empty) => break,
+                // TODO: Handle the packet.
-                Err(TryRecvError::Disconnected) => panic!("Message sender disconnected"),
             }
+            Packet::PlayerPosition(_playerposition) => {
+                // TODO: Handle the packet.
+            }
+            Packet::PlayerLook(_playerlook) => {
+                // TODO: Handle the packet.
+            }
+            Packet::ServerboundPlayerPositionAndLook(_playerpositionandlook) => {
+                // TODO: Handle the packet.
+            }
+            // Other.
+            _ => error!("handling unknown packet type: {:?}", packet),
         }
         Ok(())
     }
-
-    pub async fn broadcast_message(&mut self, message: BroadcastMessage) {
-        let mut v = Vec::new();
-        for client in self.network_clients.iter_mut() {
-            v.push(client.handle_broadcast_message(message.clone()));
-        }
-        futures::future::join_all(v).await;
-    }
-
-    /// Update the game server.
-    ///
-    /// Start by updating the network.
-    pub async fn update(&mut self) -> tokio::io::Result<()> {
-        self.update_network().await?;
-        Ok(())
-    }
 }

src/server/net/mod.rs

@@ -2,12 +2,15 @@
 pub mod packets;
 
 use super::messages::*;
-use crate::{mctypes::*, CONFIG, FAVICON};
+use crate::mctypes::*;
 use log::*;
 use packets::*;
-use serde_json::json;
+// use serde_json::json;
-use std::sync::mpsc::Sender;
+use std::{
-use std::time::{Duration, Instant};
+    collections::VecDeque,
+    sync::mpsc::Sender,
+    time::{Duration, Instant},
+};
 use tokio::net::TcpStream;
 
 /// The network client can only be in a few states,
@@ -29,10 +32,10 @@ pub struct NetworkClient {
     pub connected: bool,
     pub stream: TcpStream,
     pub state: NetworkClientState,
-    pub uuid: Option<String>,
-    pub username: Option<String>,
     pub last_keep_alive: Instant,
     pub message_sender: Sender<ServerboundMessage>,
+    packets: VecDeque<Packet>,
+    pub player: Option<crate::entity::player::Player>,
 }
 impl NetworkClient {
     /// Create a new `NetworkClient`
@@ -46,174 +49,101 @@ impl NetworkClient {
             connected: true,
             stream,
             state: NetworkClientState::Handshake,
-            uuid: None,
-            username: None,
             last_keep_alive: Instant::now(),
             message_sender,
+            packets: VecDeque::new(),
+            player: None,
         }
     }
 
-    /// Update the client.
+    /// Try to read a new packet into the processing queue.
-    ///
+    pub async fn update(&mut self) -> tokio::io::Result<()> {
-    /// Updating could mean connecting new clients, reading packets,
+        // Don't try to read packets if disconnected.
-    /// writing packets, or disconnecting clients.
+        if self.state == NetworkClientState::Disconnected {
-    pub async fn update(&mut self, num_players: usize) -> tokio::io::Result<()> {
+            return Ok(());
-        // println!("{:?}", self);
-        match self.state {
-            NetworkClientState::Handshake => {
-                let (_packet_length, _packet_id) = read_packet_header(&mut self.stream).await?;
-                let handshake = self.get_packet::<Handshake>().await?;
-                // Minecraft versions 1.8 - 1.8.9 use protocol version 47.
-                let compatible_versions = handshake.protocol_version == 47;
-                let next_state = match handshake.next_state.into() {
-                    1 => NetworkClientState::Status,
-                    2 => NetworkClientState::Login,
-                    _ => NetworkClientState::Disconnected,
-                };
-                self.state = next_state;
-                // If incompatible versions or wrong next state
-                if !compatible_versions {
-                    let mut logindisconnect = LoginDisconnect::new();
-                    logindisconnect.reason = MCChat {
-                        text: MCString::from("Incompatible client! Server is on 1.8.9"),
-                    };
-                    self.send_packet(logindisconnect).await?;
-                    self.state = NetworkClientState::Disconnected;
         }
+        if self.stream.peek(&mut [0u8; 4096]).await? > 0 {
+            // Read the packet header.
+            let (_packet_length, packet_id) = read_packet_header(&mut self.stream).await?;
+            // Get the packet based on packet_id.
+            let packet = match packet_id.value {
+                0x00 => match self.state {
+                    NetworkClientState::Handshake => {
+                        Some(self.get_wrapped_packet::<Handshake>().await)
                     }
                     NetworkClientState::Status => {
-                let (_packet_length, _packet_id) = read_packet_header(&mut self.stream).await?;
+                        Some(self.get_wrapped_packet::<StatusRequest>().await)
-                let _statusrequest = self.get_packet::<StatusRequest>().await?;
-                let mut statusresponse = StatusResponse::new();
-                statusresponse.json_response = json!({
-                    "version": {
-                        "name": "1.8.9",
-                        "protocol": 47,
-                    },
-                    "players": {
-                        "max": CONFIG.max_players,
-                        "online": num_players,
-                        "sample": [
-                            {
-                                "name": "shvr",
-                                "id": "e3f58380-60bb-4714-91f2-151d525e64aa"
-                            }
-                        ]
-                    },
-                    "description": {
-                        "text": CONFIG.motd
-                    },
-                    "favicon": format!("data:image/png;base64,{}", if FAVICON.is_ok() { radix64::STD.encode(FAVICON.as_ref().unwrap().as_slice()) } else { "".to_owned() })
-                })
-                .to_string()
-                .into();
-                self.send_packet(statusresponse).await?;
-                let (_packet_length, _packet_id) = read_packet_header(&mut self.stream).await?;
-                let statusping = self.get_packet::<StatusPing>().await?;
-                let mut statuspong = StatusPong::new();
-                statuspong.payload = statusping.payload;
-                self.send_packet(statuspong).await?;
-                self.state = NetworkClientState::Disconnected;
                     }
                     NetworkClientState::Login => {
-                let (_packet_length, _packet_id) = read_packet_header(&mut self.stream).await?;
+                        Some(self.get_wrapped_packet::<LoginStart>().await)
-                let loginstart = self.get_packet::<LoginStart>().await?;
-                // Offline mode skips encryption and compression.
-                // TODO: Encryption and compression
-                let mut loginsuccess = LoginSuccess::new();
-                // We're in offline mode, so this is a temporary uuid.
-                // TODO: Get uuid and username from Mojang servers.
-                loginsuccess.uuid = "00000000-0000-3000-0000-000000000000".into();
-                loginsuccess.username = loginstart.player_name;
-                self.uuid = Some(loginsuccess.uuid.clone().into());
-                self.username = Some(loginsuccess.username.clone().into());
-                self.send_packet(loginsuccess).await?;
-                self.state = NetworkClientState::Play;
-                let joingame = JoinGame::new();
-                // TODO: Fill out `joingame` with actual information.
-                self.send_packet(joingame).await?;
-                let (_packet_length, _packet_id) = read_packet_header(&mut self.stream).await?;
-                let _clientsettings = self.get_packet::<ClientSettings>().await?;
-                // TODO: Actually use client settings.
-                let helditemchange = HeldItemChange::new();
-                // TODO: Retrieve selected slot from storage.
-                self.send_packet(helditemchange).await?;
-                // TODO: S->C Declare Recipes (1.16?)
-                // TODO: S->C Tags (1.16?)
-                // TODO: S->C Entity Status (optional?)
-                // TODO: S->C Declare Commands (1.16?)
-                // TODO: S->C Unlock Recipes (1.16?)
-                // TODO: S->C Player Position and Look
-                let playerpositionandlook = ClientboundPlayerPositionAndLook::new();
-                // TODO: Retrieve player position from storage.
-                self.send_packet(playerpositionandlook).await?;
-                // TODO: S->C Player Info (Add Player action) (1.16?)
-                // TODO: S->C Player Info (Update latency action) (1.16?)
-                // TODO: S->C Update View Position (1.16?)
-                // TODO: S->C Update Light (1.16?)
-                // TODO: S->C Chunk Data
-                // TODO: S->C World Border
-                // TODO: S->C Spawn Position
-                let spawnposition = SpawnPosition::new();
-                self.send_packet(spawnposition).await?;
-                // Send initial keep alive.
-                self.keep_alive().await?;
-                // TODO: S->C Player Position and Look
-                // TODO: C->S Teleport Confirm
-                // TODO: C->S Player Position and Look
-                // TODO: C->S Client Status
-                // TODO: S->C inventories, entities, etc.
-                self.message_sender
-                    .send(ServerboundMessage::PlayerJoin(
-                        self.uuid
-                            .as_ref()
-                            .unwrap_or(&"00000000-0000-3000-0000-000000000000".to_owned())
-                            .to_string(),
-                        self.username
-                            .as_ref()
-                            .unwrap_or(&"unknown".to_owned())
-                            .to_string(),
-                    ))
-                    .expect("Message receiver disconnected");
                     }
                     NetworkClientState::Play => {
+                        Some(self.get_wrapped_packet::<KeepAlivePong>().await)
+                    }
+                    _ => None,
+                },
+                0x01 => {
+                    match self.state {
+                        NetworkClientState::Status => {
+                            Some(self.get_wrapped_packet::<StatusPing>().await)
+                        }
+                        NetworkClientState::Login => None, // TODO: 0x01 Encryption Response
+                        NetworkClientState::Play => {
+                            Some(self.get_wrapped_packet::<ServerboundChatMessage>().await)
+                        }
+                        _ => None,
+                    }
+                }
+                // The rest of the packets are all always in the play state.
+                0x02 => None, // TODO: 0x02 Use Entity
+                0x03 => Some(self.get_wrapped_packet::<Player>().await),
+                0x04 => Some(self.get_wrapped_packet::<PlayerPosition>().await),
+                0x05 => Some(self.get_wrapped_packet::<PlayerLook>().await),
+                0x06 => Some(
+                    self.get_wrapped_packet::<ServerboundPlayerPositionAndLook>()
+                        .await,
+                ),
+                0x07 => None, // TODO: 0x07 Player Digging
+                0x08 => None, // TODO: 0x08 Player Block Placement
+                0x09 => None, // TODO: 0x09 Held Item Change
+                0x0a => None, // TODO: 0x0a Animation
+                0x0b => None, // TODO: 0x0b Entity Action
+                0x0c => None, // TODO: 0x0c Steer Vehicle
+                0x0d => None, // TODO: 0x0d Close Window
+                0x0e => None, // TODO: 0x0e Click Window
+                0x0f => None, // TODO: 0x0f Confirm Transaction
+                0x10 => None, // TODO: 0x10 Creative Inventory Action
+                0x11 => None, // TODO: 0x11 Enchant Item
+                0x12 => None, // TODO: 0x12 Update Sign
+                0x13 => None, // TODO: 0x13 Player Abilities
+                0x14 => None, // TODO: 0x14 Tab-Complete
+                0x15 => Some(self.get_wrapped_packet::<ClientSettings>().await),
+                0x16 => None, // TODO: 0x16 Client Status
+                0x17 => None, // TODO: 0x17 Plugin Message
+                0x18 => None, // TODO: 0x18 Spectate
+                0x19 => None, // TODO: 0x19 Resource Pack Status
+                _ => None,
+            };
+            if let Some(Ok(packet)) = packet {
+                // Add it to the internal queue to be processed.
+                self.packets.push_back(packet);
+            }
+        }
         if self.last_keep_alive.elapsed() > Duration::from_millis(1000) {
+            debug!(
+                "Sending keep alive, last one was {:?} ago",
+                self.last_keep_alive.elapsed()
+            );
             self.keep_alive().await?;
         }
-                let (packet_length, packet_id) = read_packet_header(&mut self.stream).await?;
-                // debug!("{}", packet_id);
-                if packet_id == Player::id() {
-                    let _player = self.get_packet::<Player>().await?;
-                } else if packet_id == PlayerPosition::id() {
-                    let _playerposition = self.get_packet::<PlayerPosition>().await?;
-                } else if packet_id == PlayerLook::id() {
-                    let _playerlook = self.get_packet::<PlayerLook>().await?;
-                } else if packet_id == ServerboundPlayerPositionAndLook::id() {
-                    let _playerpositionandlook = self
-                        .get_packet::<ServerboundPlayerPositionAndLook>()
-                        .await?;
-                } else if packet_id == ServerboundChatMessage::id() {
-                    let serverboundchatmessage =
-                        self.get_packet::<ServerboundChatMessage>().await?;
-                    let reply = format!("<{}> {}", self.get_name(), serverboundchatmessage.text);
-                    info!("{}", reply);
-                    self.message_sender
-                        .send(ServerboundMessage::Chat(reply))
-                        .expect("Message receiver disconnected");
-                } else {
-                    let _ = read_bytes(&mut self.stream, Into::<i32>::into(packet_length) as usize)
-                        .await?;
-                }
-            }
-            NetworkClientState::Disconnected => {
-                if self.connected {
-                    self.disconnect("Disconnected").await?;
-                }
-            }
-        }
         Ok(())
     }
 
+    /// Pop a packet from the queue.
+    pub fn read_packet(&mut self) -> Option<Packet> {
+        self.packets.pop_front()
+    }
+
     /// Send a generic packet to the client.
     pub async fn send_packet<P: PacketCommon>(&mut self, packet: P) -> tokio::io::Result<()> {
         debug!("Sent {:?} {:#04X?} {:?}", self.state, P::id(), packet);
@@ -221,12 +151,19 @@ impl NetworkClient {
     }
 
     /// Read a generic packet from the network.
-    pub async fn get_packet<T: PacketCommon>(&mut self) -> tokio::io::Result<T> {
+    async fn get_packet<T: PacketCommon>(&mut self) -> tokio::io::Result<T> {
         let packet = T::read(&mut self.stream).await?;
         debug!("Got {:?} {:#04X?} {:?}", self.state, T::id(), packet);
         Ok(packet)
     }
 
+    /// Read a generic packet from the network and wrap it in `Packet`.
+    async fn get_wrapped_packet<T: PacketCommon>(&mut self) -> tokio::io::Result<Packet> {
+        let packet = T::read(&mut self.stream).await?;
+        debug!("Got {:?} {:#04X?} {:?}", self.state, T::id(), packet);
+        Ok(packet.as_packet())
+    }
+
     /// Send the client a message in chat.
     pub async fn send_chat_message<C: Into<MCChat>>(
         &mut self,
@@ -258,32 +195,20 @@ impl NetworkClient {
     /// Send a keep alive packet to the client.
     pub async fn keep_alive(&mut self) -> tokio::io::Result<()> {
         if cfg!(debug_assertions) {
-            self.send_chat_message("keep alive").await?;
+            // self.send_chat_message("keep alive").await?;
         }
         // Keep alive ping to client.
-        let clientboundkeepalive = KeepAlivePing::new();
+        self.send_packet(KeepAlivePing::new()).await?;
-        self.send_packet(clientboundkeepalive).await?;
         // Keep alive pong to server.
         let (_packet_length, _packet_id) = read_packet_header(&mut self.stream).await?;
-        let _serverboundkeepalive = self.get_packet::<KeepAlivePong>().await?;
+        let _ = self.get_packet::<KeepAlivePong>().await?;
         self.last_keep_alive = Instant::now();
         Ok(())
     }
 
-    /// Helper function to get the name of the player.
+    /// Receives messages from the server.
-    pub fn get_name(&self) -> String {
+    pub async fn handle_message(&mut self, message: ClientboundMessage) -> tokio::io::Result<()> {
-        self.username
+        use ClientboundMessage::*;
-            .as_ref()
-            .unwrap_or(&"unknown".to_owned())
-            .to_string()
-    }
-
-    /// Receives broadcast messages from the server.
-    pub async fn handle_broadcast_message(
-        &mut self,
-        message: BroadcastMessage,
-    ) -> tokio::io::Result<()> {
-        use BroadcastMessage::*;
         match message {
             Chat(s) => self.send_chat_message(s).await?,
             Disconnect(reason) => self.disconnect(reason).await?,

src/server/net/packets/mod.rs

@@ -96,12 +96,30 @@ register_packets!(
 );
 
 #[async_trait::async_trait]
-pub trait PacketCommon: Into<Packet> + core::fmt::Debug
+pub trait PacketCommon: Into<Packet> + core::fmt::Debug + Clone
 where
     Self: Sized,
 {
     fn new() -> Self;
     fn id() -> u8;
+    fn as_packet(&self) -> Packet {
+        self.clone().into()
+    }
     async fn read(t: &mut TcpStream) -> tokio::io::Result<Self>;
     async fn write(&self, t: &mut TcpStream) -> tokio::io::Result<()>;
 }
+#[async_trait::async_trait]
+impl PacketCommon for Packet {
+    fn new() -> Self {
+        Packet::new()
+    }
+    fn id() -> u8 {
+        255 // The generic `Packet` doesn't really have an id, but I can't leave it blank.
+    }
+    async fn read(_t: &mut TcpStream) -> tokio::io::Result<Self> {
+        panic!("cannot PacketCommon::read a generic Packet")
+    }
+    async fn write(&self, _t: &mut TcpStream) -> tokio::io::Result<()> {
+        panic!("cannot PacketCommon::write a generic Packet")
+    }
+}

src/world/location.rs

@@ -1,4 +1,5 @@
 /// Used to store a point in a world.
+#[derive(Debug, Clone, PartialEq)]
 pub struct Location {
     pub x: f64,
     pub y: f64,