Port to tokio

Cargo.toml

@@ -12,5 +12,5 @@ lto = true
 codegen-units = 1
 
 [workspace.dependencies]
-tokio = { version = "1.39.2", features = ["net", "macros", "rt-multi-thread", "fs", "io-util"] }
+tokio = { version = "1.39.3", features = ["net", "macros", "rt-multi-thread", "fs", "io-util"] }
 rayon = "1.10.0"

pumpkin-protocol/Cargo.toml

@@ -9,6 +9,7 @@ pumpkin-world =  { path = "../pumpkin-world" }
 pumpkin-text = { path = "../pumpkin-text" }
 
 bytes = "1.7"
+byteorder = "1.5.0"
 
 uuid = "1.10"
 

pumpkin-protocol/src/lib.rs

@@ -1,4 +1,5 @@
 use bytebuf::{packet_id::Packet, ByteBuffer, DeserializerError};
+use byteorder::ReadBytesExt;
 use bytes::Buf;
 use serde::{Deserialize, Serialize, Serializer};
 use std::io::{self, Write};
@@ -10,8 +11,8 @@ pub mod packet_decoder;
 pub mod packet_encoder;
 pub mod position;
 pub mod server;
-pub mod uuid;
 pub mod slot;
+pub mod uuid;
 
 pub const CURRENT_MC_PROTOCOL: u32 = 767;
 
@@ -44,7 +45,7 @@ impl VarInt {
     pub fn decode_partial(r: &mut &[u8]) -> Result<i32, VarIntDecodeError> {
         let mut val = 0;
         for i in 0..Self::MAX_SIZE {
-            let byte = r.get_u8();
+            let byte = r.read_u8().map_err(|_| VarIntDecodeError::Incomplete)?;
             val |= (i32::from(byte) & 0b01111111) << (i * 7);
             if byte & 0b10000000 == 0 {
                 return Ok(val);

pumpkin-protocol/src/packet_decoder.rs

@@ -24,7 +24,6 @@ pub struct PacketDecoder {
 impl PacketDecoder {
     pub fn decode(&mut self) -> Result<Option<RawPacket>, PacketError> {
         let mut r = &self.buf[..];
-
         let packet_len = match VarInt::decode_partial(&mut r) {
             Ok(len) => len,
             Err(VarIntDecodeError::Incomplete) => return Ok(None),

pumpkin-world/src/world.rs

@@ -153,37 +153,36 @@ impl Level {
                 let modulus = |a: i32, b: i32| ((a % b) + b) % b;
                 let chunk_x = modulus(chunk.0, 32) as u32;
                 let chunk_z = modulus(chunk.1, 32) as u32;
-                let channel = channel.clone();
                 let table_entry = (chunk_x + chunk_z * 32) * 4;
+                let table_entry = table_entry as usize;
 
                 let mut offset = vec![0u8];
-                offset.extend_from_slice(
-                    &location_table[table_entry as usize..table_entry as usize + 3],
-                );
+                offset.extend_from_slice(&location_table[table_entry..table_entry + 3]);
                 let offset = u32::from_be_bytes(offset.try_into().unwrap()) as u64 * 4096;
-                let size = location_table[table_entry as usize + 3] as usize * 4096;
+                let size = location_table[table_entry + 3] as usize * 4096;
 
                 if offset == 0 && size == 0 {
                     let _ =
                         channel.blocking_send(((chunk.0, chunk.1), Err(WorldError::ChunkNotFound)));
                     return;
                 }
                 // Read the file using the offset and size
-                let mut file_buf = {
-                    let seek_result = region_file.seek(std::io::SeekFrom::Start(offset));
-                    if seek_result.is_err() {
+                let mut file_buf = match region_file.seek(std::io::SeekFrom::Start(offset)) {
+                    Ok(_) => vec![0; size],
+                    Err(_) => {
                         let _ = channel
                             .blocking_send(((chunk.0, chunk.1), Err(WorldError::RegionIsInvalid)));
                         return;
                     }
-                    let mut out = vec![0; size];
-                    let read_result = region_file.read_exact(&mut out);
-                    if read_result.is_err() {
+                };
+
+                match region_file.read_exact(&mut file_buf) {
+                    Ok(_) => (),
+                    Err(_) => {
                         let _ = channel
                             .blocking_send(((chunk.0, chunk.1), Err(WorldError::RegionIsInvalid)));
                         return;
                     }
-                    out
                 };
 
                 // TODO: check checksum to make sure chunk is not corrupted

pumpkin/Cargo.toml

@@ -58,7 +58,6 @@ simple_logger = "5.0.0"
 log = "0.4"
 
 # networking
-mio = { version = "1.0.1", features = ["os-poll", "net"]}
 crossbeam-channel = "0.5.13"
 
 uuid = { version = "1.10", features = ["serde", "v3"]}

pumpkin/src/client/client_packet.rs

@@ -34,39 +34,45 @@ use super::{
 /// Processes incoming Packets from the Client to the Server
 /// Implements the `Client` Packets
 impl Client {
-    pub fn handle_handshake(&mut self, _server: &mut Server, handshake: SHandShake) {
+    pub async fn handle_handshake(&mut self, _server: &mut Server, handshake: SHandShake) {
         dbg!("handshake");
         self.protocol_version = handshake.protocol_version.0;
         self.connection_state = handshake.next_state;
         if self.connection_state == ConnectionState::Login {
             if self.protocol_version < CURRENT_MC_PROTOCOL as i32 {
                 let protocol = self.protocol_version;
-                self.kick(&format!("Client outdated ({protocol}), Server uses Minecraft {CURRENT_MC_VERSION}, Protocol {CURRENT_MC_PROTOCOL}"));
+                self.kick(&format!("Client outdated ({protocol}), Server uses Minecraft {CURRENT_MC_VERSION}, Protocol {CURRENT_MC_PROTOCOL}")).await;
             } else if self.protocol_version > CURRENT_MC_PROTOCOL as i32 {
-                self.kick(&format!("Server outdated, Server uses Minecraft {CURRENT_MC_VERSION}, Protocol {CURRENT_MC_PROTOCOL}"));
+                self.kick(&format!("Server outdated, Server uses Minecraft {CURRENT_MC_VERSION}, Protocol {CURRENT_MC_PROTOCOL}")).await;
             }
         }
     }
 
-    pub fn handle_status_request(&mut self, server: &mut Server, _status_request: SStatusRequest) {
-        self.send_packet(&CStatusResponse::new(&server.status_response_json));
+    pub async fn handle_status_request(
+        &mut self,
+        server: &mut Server,
+        _status_request: SStatusRequest,
+    ) {
+        self.send_packet(&CStatusResponse::new(&server.status_response_json))
+            .await;
     }
 
-    pub fn handle_ping_request(&mut self, _server: &mut Server, ping_request: SPingRequest) {
+    pub async fn handle_ping_request(&mut self, _server: &mut Server, ping_request: SPingRequest) {
         dbg!("ping");
-        self.send_packet(&CPingResponse::new(ping_request.payload));
+        self.send_packet(&CPingResponse::new(ping_request.payload))
+            .await;
         self.close();
     }
 
     fn is_valid_player_name(name: &str) -> bool {
         name.len() <= 16 && name.chars().all(|c| c > 32 as char && c < 127 as char)
     }
 
-    pub fn handle_login_start(&mut self, server: &mut Server, login_start: SLoginStart) {
+    pub async fn handle_login_start(&mut self, server: &mut Server, login_start: SLoginStart) {
         dbg!("login start");
 
         if !Self::is_valid_player_name(&login_start.name) {
-            self.kick("Invalid characters in username");
+            self.kick("Invalid characters in username").await;
             return;
         }
         // default game profile, when no online mode
@@ -80,7 +86,7 @@ impl Client {
         let proxy = &server.advanced_config.proxy;
         if proxy.enabled {
             if proxy.velocity.enabled {
-                velocity_login(self)
+                velocity_login(self).await
             }
             return;
         }
@@ -94,7 +100,7 @@ impl Client {
             &verify_token,
             server.base_config.online_mode, // TODO
         );
-        self.send_packet(&packet);
+        self.send_packet(&packet).await;
     }
 
     pub async fn handle_encryption_response(
@@ -107,8 +113,12 @@ impl Client {
             .decrypt(Pkcs1v15Encrypt, &encryption_response.shared_secret)
             .map_err(|_| EncryptionError::FailedDecrypt)
             .unwrap();
-        self.enable_encryption(&shared_secret)
-            .unwrap_or_else(|e| self.kick(&e.to_string()));
+        match self.enable_encryption(&shared_secret) {
+            Ok(d) => d,
+            Err(e) => {
+                self.kick(&e.to_string()).await;
+            }
+        }
 
         if server.base_config.online_mode {
             let hash = Sha1::new()
@@ -135,7 +145,7 @@ impl Client {
                             .allow_banned_players
                         {
                             if !p.is_empty() {
-                                self.kick("Your account can't join");
+                                self.kick("Your account can't join").await;
                             }
                         } else {
                             for allowed in server
@@ -146,14 +156,14 @@ impl Client {
                                 .clone()
                             {
                                 if !p.contains(&allowed) {
-                                    self.kick("Your account can't join");
+                                    self.kick("Your account can't join").await;
                                 }
                             }
                         }
                     }
                     self.gameprofile = Some(p);
                 }
-                Err(e) => self.kick(&e.to_string()),
+                Err(e) => self.kick(&e.to_string()).await,
             }
         }
         for ele in self.gameprofile.as_ref().unwrap().properties.clone() {
@@ -168,32 +178,33 @@ impl Client {
                 .packet_compression
                 .compression_threshold;
             let level = server.advanced_config.packet_compression.compression_level;
-            self.send_packet(&CSetCompression::new(threshold.into()));
+            self.send_packet(&CSetCompression::new(threshold.into()))
+                .await;
             self.set_compression(Some((threshold, level)));
         }
 
         if let Some(profile) = self.gameprofile.as_ref().cloned() {
             let packet = CLoginSuccess::new(profile.id, &profile.name, &profile.properties, false);
-            self.send_packet(&packet);
+            self.send_packet(&packet).await;
         } else {
-            self.kick("game profile is none");
+            self.kick("game profile is none").await;
         }
     }
 
-    pub fn handle_plugin_response(
+    pub async fn handle_plugin_response(
         &mut self,
         _server: &mut Server,
         _plugin_response: SLoginPluginResponse,
     ) {
     }
 
-    pub fn handle_login_acknowledged(
+    pub async fn handle_login_acknowledged(
         &mut self,
         server: &mut Server,
         _login_acknowledged: SLoginAcknowledged,
     ) {
         self.connection_state = ConnectionState::Config;
-        server.send_brand(self);
+        server.send_brand(self).await;
 
         let resource_config = &server.advanced_config.resource_pack;
         if resource_config.enabled {
@@ -208,18 +219,20 @@ impl Client {
                 resource_config.resource_pack_sha1.clone(),
                 resource_config.force,
                 prompt_message,
-            ));
+            ))
+            .await;
         }
 
         // known data packs
         self.send_packet(&CKnownPacks::new(&[KnownPack {
             namespace: "minecraft",
             id: "core",
             version: "1.21",
-        }]));
+        }]))
+        .await;
         dbg!("login achnowlaged");
     }
-    pub fn handle_client_information_config(
+    pub async fn handle_client_information_config(
         &mut self,
         _server: &mut Server,
         client_information: SClientInformationConfig,
@@ -237,29 +250,38 @@ impl Client {
         });
     }
 
-    pub fn handle_plugin_message(&mut self, _server: &mut Server, plugin_message: SPluginMessage) {
+    pub async fn handle_plugin_message(
+        &mut self,
+        _server: &mut Server,
+        plugin_message: SPluginMessage,
+    ) {
         if plugin_message.channel.starts_with("minecraft:brand")
             || plugin_message.channel.starts_with("MC|Brand")
         {
             dbg!("got a client brand");
             match String::from_utf8(plugin_message.data) {
                 Ok(brand) => self.brand = Some(brand),
-                Err(e) => self.kick(&e.to_string()),
+                Err(e) => self.kick(&e.to_string()).await,
             }
         }
     }
 
-    pub fn handle_known_packs(&mut self, server: &mut Server, _config_acknowledged: SKnownPacks) {
+    pub async fn handle_known_packs(
+        &mut self,
+        server: &mut Server,
+        _config_acknowledged: SKnownPacks,
+    ) {
         for registry in &server.cached_registry {
             self.send_packet(&CRegistryData::new(
                 &registry.registry_id,
                 &registry.registry_entries,
-            ));
+            ))
+            .await;
         }
 
         // We are done with configuring
         dbg!("finish config");
-        self.send_packet(&CFinishConfig::new());
+        self.send_packet(&CFinishConfig::new()).await;
     }
 
     pub async fn handle_config_acknowledged(