diff --git a/pumpkin-inventory/Cargo.toml b/pumpkin-inventory/Cargo.toml index abeb7cfd1..c4dc9c41b 100644 --- a/pumpkin-inventory/Cargo.toml +++ b/pumpkin-inventory/Cargo.toml @@ -11,5 +11,5 @@ num-traits = "0.2" num-derive = "0.4" thiserror = "1.0.63" itertools.workspace = true -parking_lot.workspace = true crossbeam.workspace = true +tokio.workspace = true diff --git a/pumpkin-inventory/src/drag_handler.rs b/pumpkin-inventory/src/drag_handler.rs index 8479c3635..1d7fffa8b 100644 --- a/pumpkin-inventory/src/drag_handler.rs +++ b/pumpkin-inventory/src/drag_handler.rs @@ -2,10 +2,10 @@ use crate::container_click::MouseDragType; use crate::{Container, InventoryError}; use itertools::Itertools; use num_traits::Euclid; -use parking_lot::{Mutex, RwLock}; use pumpkin_world::item::ItemStack; use std::collections::HashMap; use std::sync::Arc; +use tokio::sync::{Mutex, RwLock}; #[derive(Debug, Default)] pub struct DragHandler(RwLock>>>); @@ -13,7 +13,7 @@ impl DragHandler { pub fn new() -> Self { Self(RwLock::new(HashMap::new())) } - pub fn new_drag( + pub async fn new_drag( &self, container_id: u64, player: i32, @@ -24,21 +24,21 @@ impl DragHandler { drag_type, slots: vec![], }; - let mut drags = self.0.write(); + let mut drags = self.0.write().await; drags.insert(container_id, Arc::new(Mutex::new(drag))); Ok(()) } - pub fn add_slot( + pub async fn add_slot( &self, container_id: u64, player: i32, slot: usize, ) -> Result<(), InventoryError> { - let drags = self.0.read(); + let drags = self.0.read().await; match drags.get(&container_id) { Some(drag) => { - let mut drag = drag.lock(); + let mut drag = drag.lock().await; if drag.player != player { Err(InventoryError::MultiplePlayersDragging)? } @@ -51,7 +51,7 @@ impl DragHandler { Ok(()) } - pub fn apply_drag( + pub async fn apply_drag( &self, maybe_carried_item: &mut Option, container: &mut T, @@ -63,11 +63,11 @@ impl DragHandler { return Ok(()); } - let mut drags = self.0.write(); + let mut drags = self.0.write().await; let Some((_, drag)) = drags.remove_entry(container_id) else { Err(InventoryError::OutOfOrderDragging)? }; - let drag = drag.lock(); + let drag = drag.lock().await; if player != drag.player { Err(InventoryError::MultiplePlayersDragging)? diff --git a/pumpkin-inventory/src/open_container.rs b/pumpkin-inventory/src/open_container.rs index 261ba3b74..882b04998 100644 --- a/pumpkin-inventory/src/open_container.rs +++ b/pumpkin-inventory/src/open_container.rs @@ -1,7 +1,7 @@ use crate::{Container, WindowType}; -use parking_lot::Mutex; use pumpkin_world::item::ItemStack; use std::sync::Arc; +use tokio::sync::Mutex; pub struct OpenContainer { players: Vec, diff --git a/pumpkin-protocol/src/lib.rs b/pumpkin-protocol/src/lib.rs index f51cb750f..a59645ec0 100644 --- a/pumpkin-protocol/src/lib.rs +++ b/pumpkin-protocol/src/lib.rs @@ -142,8 +142,8 @@ pub enum PacketError { FailedWrite(String), #[error("failed to flush decoder")] FailedFinish, - #[error("failed to write encoded packet to connection: {0}")] - ConnectionWrite(String), + #[error("failed to write encoded packet to connection")] + ConnectionWrite, #[error("packet exceeds maximum length")] TooLong, #[error("packet length is out of bounds")] diff --git a/pumpkin/Cargo.toml b/pumpkin/Cargo.toml index 8e737ab63..45669b78e 100644 --- a/pumpkin/Cargo.toml +++ b/pumpkin/Cargo.toml @@ -59,10 +59,6 @@ png = "0.17.14" simple_logger = { version = "5.0.0", features = ["threads"] } log.workspace = true -# networking -mio = { version = "1.0.2", features = ["net", "os-poll"] } - -parking_lot.workspace = true crossbeam.workspace = true uuid.workspace = true tokio.workspace = true diff --git a/pumpkin/src/client/client_packet.rs b/pumpkin/src/client/client_packet.rs index aba49e761..a0daf76f9 100644 --- a/pumpkin/src/client/client_packet.rs +++ b/pumpkin/src/client/client_packet.rs @@ -34,35 +34,36 @@ use super::{authentication::AuthError, Client, PlayerConfig}; /// NEVER TRUST THE CLIENT. HANDLE EVERY ERROR, UNWRAP/EXPECT /// TODO: REMOVE ALL UNWRAPS impl Client { - pub fn handle_handshake(&self, handshake: SHandShake) { + pub async fn handle_handshake(&self, handshake: SHandShake) { dbg!("handshake"); let version = handshake.protocol_version.0; self.protocol_version .store(version, std::sync::atomic::Ordering::Relaxed); - *self.server_address.lock() = handshake.server_address; + *self.server_address.lock().await = handshake.server_address; self.connection_state.store(handshake.next_state); if self.connection_state.load() != ConnectionState::Status { let protocol = version; match protocol.cmp(&(CURRENT_MC_PROTOCOL as i32)) { std::cmp::Ordering::Less => { - 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; } std::cmp::Ordering::Equal => {} std::cmp::Ordering::Greater => { - 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(&self, server: &Server, _status_request: SStatusRequest) { - self.send_packet(&server.get_status()); + pub async fn handle_status_request(&self, server: &Server, _status_request: SStatusRequest) { + self.send_packet(&server.get_status()).await; } - pub fn handle_ping_request(&self, ping_request: SStatusPingRequest) { + pub async fn handle_ping_request(&self, ping_request: SStatusPingRequest) { dbg!("ping"); - self.send_packet(&CPingResponse::new(ping_request.payload)); + self.send_packet(&CPingResponse::new(ping_request.payload)) + .await; self.close(); } @@ -73,28 +74,28 @@ impl Client { .all(|c| c > 32_u8 as char && c < 127_u8 as char) } - pub fn handle_login_start(&self, server: &Server, login_start: SLoginStart) { + pub async fn handle_login_start(&self, server: &Server, login_start: SLoginStart) { log::debug!("login start, State {:?}", self.connection_state); 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 // TODO: make offline uuid - let mut gameprofile = self.gameprofile.lock(); + let mut gameprofile = self.gameprofile.lock().await; let proxy = &ADVANCED_CONFIG.proxy; if proxy.enabled { if proxy.velocity.enabled { - velocity_login(self); + velocity_login(self).await; } else if proxy.bungeecord.enabled { - match bungeecord::bungeecord_login(self, login_start.name) { + match bungeecord::bungeecord_login(self, login_start.name).await { Ok((_ip, profile)) => { // self.address.lock() = ip; - self.finish_login(&profile); + self.finish_login(&profile).await; *gameprofile = Some(profile); } - Err(error) => self.kick(&error.to_string()), + Err(error) => self.kick(&error.to_string()).await, } } } else { @@ -107,7 +108,8 @@ impl Client { // TODO: check config for encryption let verify_token: [u8; 4] = rand::random(); - self.send_packet(&server.encryption_request(&verify_token, BASIC_CONFIG.online_mode)); + self.send_packet(&server.encryption_request(&verify_token, BASIC_CONFIG.online_mode)) + .await; } } @@ -118,10 +120,11 @@ impl Client { ) { let shared_secret = server.decrypt(&encryption_response.shared_secret).unwrap(); - self.set_encryption(Some(&shared_secret)) - .unwrap_or_else(|e| self.kick(&e.to_string())); - - let mut gameprofile = self.gameprofile.lock(); + if let Err(error) = self.set_encryption(Some(&shared_secret)).await { + self.kick(&error.to_string()).await; + return; + } + let mut gameprofile = self.gameprofile.lock().await; if BASIC_CONFIG.online_mode { match self @@ -130,30 +133,31 @@ impl Client { { Ok(profile) => *gameprofile = Some(profile), Err(e) => { - self.kick(&e.to_string()); + self.kick(&e.to_string()).await; } } } if let Some(profile) = gameprofile.as_ref() { if ADVANCED_CONFIG.packet_compression.enabled { - self.enable_compression(); + self.enable_compression().await; } - self.finish_login(profile); + self.finish_login(profile).await; } else { - self.kick("No Game profile"); + self.kick("No Game profile").await; } } - fn enable_compression(&self) { + async fn enable_compression(&self) { let compression = ADVANCED_CONFIG.packet_compression.compression_info.clone(); - self.send_packet(&CSetCompression::new(compression.threshold.into())); - self.set_compression(Some(compression)); + self.send_packet(&CSetCompression::new(compression.threshold.into())) + .await; + self.set_compression(Some(compression)).await; } - fn finish_login(&self, profile: &GameProfile) { + async fn finish_login(&self, profile: &GameProfile) { let packet = CLoginSuccess::new(&profile.id, &profile.name, &profile.properties, false); - self.send_packet(&packet); + self.send_packet(&packet).await; } async fn autenticate( @@ -164,7 +168,7 @@ impl Client { ) -> Result { if let Some(auth_client) = &server.auth_client { let hash = server.digest_secret(shared_secret); - let ip = self.address.lock().ip(); + let ip = self.address.lock().await.ip(); let profile = authentication::authenticate(username, &hash, &ip, auth_client).await?; // Check if player should join @@ -199,32 +203,32 @@ impl Client { Err(AuthError::MissingAuthClient) } - pub fn handle_plugin_response(&self, plugin_response: SLoginPluginResponse) { + pub async fn handle_plugin_response(&self, plugin_response: SLoginPluginResponse) { let velocity_config = &ADVANCED_CONFIG.proxy.velocity; if velocity_config.enabled { - let mut address = self.address.lock(); + let mut address = self.address.lock().await; match velocity::receive_velocity_plugin_response( address.port(), velocity_config, plugin_response, ) { Ok((profile, new_address)) => { - self.finish_login(&profile); - *self.gameprofile.lock() = Some(profile); + self.finish_login(&profile).await; + *self.gameprofile.lock().await = Some(profile); *address = new_address } - Err(error) => self.kick(&error.to_string()), + Err(error) => self.kick(&error.to_string()).await, } } } - pub fn handle_login_acknowledged( + pub async fn handle_login_acknowledged( &self, server: &Server, _login_acknowledged: SLoginAcknowledged, ) { self.connection_state.store(ConnectionState::Config); - self.send_packet(&server.get_branding()); + self.send_packet(&server.get_branding()).await; let resource_config = &ADVANCED_CONFIG.resource_pack; if resource_config.enabled { @@ -243,7 +247,7 @@ impl Client { }, ); - self.send_packet(&resource_pack); + self.send_packet(&resource_pack).await; } // known data packs @@ -251,16 +255,20 @@ impl Client { namespace: "minecraft", id: "core", version: "1.21", - }])); + }])) + .await; dbg!("login acknowledged"); } - pub fn handle_client_information_config(&self, client_information: SClientInformationConfig) { + pub async fn handle_client_information_config( + &self, + client_information: SClientInformationConfig, + ) { dbg!("got client settings"); if let (Some(main_hand), Some(chat_mode)) = ( Hand::from_i32(client_information.main_hand.into()), ChatMode::from_i32(client_information.chat_mode.into()), ) { - *self.config.lock() = Some(PlayerConfig { + *self.config.lock().await = Some(PlayerConfig { locale: client_information.locale, view_distance: client_information.view_distance, chat_mode, @@ -271,33 +279,34 @@ impl Client { server_listing: client_information.server_listing, }); } else { - self.kick("Invalid hand or chat type") + self.kick("Invalid hand or chat type").await } } - pub fn handle_plugin_message(&self, plugin_message: SPluginMessage) { + pub async fn handle_plugin_message(&self, 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.lock() = Some(brand), - Err(e) => self.kick(&e.to_string()), + Ok(brand) => *self.brand.lock().await = Some(brand), + Err(e) => self.kick(&e.to_string()).await, } } } - pub fn handle_known_packs(&self, server: &Server, _config_acknowledged: SKnownPacks) { + pub async fn handle_known_packs(&self, server: &Server, _config_acknowledged: SKnownPacks) { for registry in &server.cached_registry { self.send_packet(&CRegistryData::new( ®istry.registry_id, ®istry.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(&self, _config_acknowledged: SAcknowledgeFinishConfig) { diff --git a/pumpkin/src/client/container.rs b/pumpkin/src/client/container.rs index 3e91a383a..60b6243c0 100644 --- a/pumpkin/src/client/container.rs +++ b/pumpkin/src/client/container.rs @@ -1,8 +1,6 @@ use crate::entity::player::Player; use crate::server::Server; use itertools::Itertools; -use parking_lot::Mutex; -use pumpkin_core::text::TextComponent; use pumpkin_core::GameMode; use pumpkin_inventory::container_click::{ Click, ClickType, KeyClick, MouseClick, MouseDragState, MouseDragType, @@ -12,7 +10,7 @@ use pumpkin_inventory::window_property::{WindowProperty, WindowPropertyTrait}; use pumpkin_inventory::Container; use pumpkin_inventory::{container_click, InventoryError, OptionallyCombinedContainer}; use pumpkin_protocol::client::play::{ - CCloseContainer, COpenScreen, CSetContainerContent, CSetContainerProperty, CSetContainerSlot, + CCloseContainer, CSetContainerContent, CSetContainerProperty, CSetContainerSlot, }; use pumpkin_protocol::server::play::SClickContainer; use pumpkin_protocol::slot::Slot; @@ -20,40 +18,40 @@ use pumpkin_world::item::ItemStack; use std::sync::Arc; impl Player { - pub fn open_container(&self, server: &Server, minecraft_menu_id: &str) { - let inventory = self.inventory.lock(); + pub async fn open_container(&self, server: &Server, minecraft_menu_id: &str) { + let inventory = self.inventory.lock().await; inventory .state_id .store(0, std::sync::atomic::Ordering::Relaxed); let total_opened_containers = inventory.total_opened_containers; let container = self.get_open_container(server); - let mut container = container.as_ref().map(|container| container.lock()); - let menu_protocol_id = (*pumpkin_world::global_registry::REGISTRY - .get("minecraft:menu") - .unwrap() - .entries - .get(minecraft_menu_id) - .expect("Should be a valid menu id") - .get("protocol_id") - .unwrap()) - .into(); - let window_title = container - .as_ref() - .map(|container| container.window_name()) - .unwrap_or_else(|| inventory.window_name()); - let title = TextComponent::text(window_title); + let container = container.as_ref().map(|container| container.lock()); + // let menu_protocol_id = (*pumpkin_world::global_registry::REGISTRY + // .get("minecraft:menu") + // .unwrap() + // .entries + // .get(minecraft_menu_id) + // .expect("Should be a valid menu id") + // .get("protocol_id") + // .unwrap()) + // .into(); + // let window_title = container + // .as_ref() + // .map(|container| container.window_name()) + // .unwrap_or_else(|| inventory.window_name()); + // let title = TextComponent::text(window_title); - self.client.send_packet(&COpenScreen::new( - total_opened_containers.into(), - menu_protocol_id, - title, - )); + // self.client.send_packet(&COpenScreen::new( + // total_opened_containers.into(), + // menu_protocol_id, + // title, + // )); drop(inventory); - self.set_container_content(container.as_deref_mut()); + // self.set_container_content(container.as_deref_mut()); } - pub fn set_container_content(&self, container: Option<&mut Box>) { - let mut inventory = self.inventory.lock(); + pub async fn set_container_content(&self, container: Option<&mut Box>) { + let mut inventory = self.inventory.lock().await; let total_opened_containers = inventory.total_opened_containers; let container = OptionallyCombinedContainer::new(&mut inventory, container); @@ -80,27 +78,30 @@ impl Player { &slots, &carried_item, ); - self.client.send_packet(&packet); + self.client.send_packet(&packet).await; } /// The official Minecraft client is weird, and will always just close *any* window that is opened when this gets sent - pub fn close_container(&self) { - let mut inventory = self.inventory.lock(); + pub async fn close_container(&self) { + let mut inventory = self.inventory.lock().await; inventory.total_opened_containers += 1; self.client .send_packet(&CCloseContainer::new(inventory.total_opened_containers)) + .await } - pub fn set_container_property( + pub async fn set_container_property( &mut self, window_property: WindowProperty, ) { let (id, value) = window_property.into_tuple(); - self.client.send_packet(&CSetContainerProperty::new( - self.inventory.lock().total_opened_containers, - id, - value, - )); + self.client + .send_packet(&CSetContainerProperty::new( + self.inventory.lock().await.total_opened_containers, + id, + value, + )) + .await; } pub async fn handle_click_container( @@ -109,22 +110,23 @@ impl Player { packet: SClickContainer, ) -> Result<(), InventoryError> { let opened_container = self.get_open_container(server); - let mut opened_container = opened_container.as_ref().map(|container| container.lock()); + let opened_container = opened_container.as_ref().map(|container| container.lock()); let drag_handler = &server.drag_handler; let state_id = self .inventory .lock() + .await .state_id .load(std::sync::atomic::Ordering::Relaxed); // This is just checking for regular desync, client hasn't done anything malicious if state_id != packet.state_id.0 as u32 { - self.set_container_content(opened_container.as_deref_mut()); + // self.set_container_content(opened_container.as_deref_mut()); return Ok(()); } if opened_container.is_some() { - if packet.window_id != self.inventory.lock().total_opened_containers { + if packet.window_id != self.inventory.lock().await.total_opened_containers { return Err(InventoryError::ClosedContainerInteract(self.entity_id())); } } else if packet.window_id != 0 { @@ -144,49 +146,56 @@ impl Player { match click.click_type { ClickType::MouseClick(mouse_click) => { - self.mouse_click(opened_container.as_deref_mut(), mouse_click, click.slot) + // self.mouse_click(opened_container.as_deref_mut(), mouse_click, click.slot).await + todo!() } ClickType::ShiftClick => { - self.shift_mouse_click(opened_container.as_deref_mut(), click.slot) + // self.shift_mouse_click(opened_container.as_deref_mut(), click.slot).await + todo!() + } + ClickType::KeyClick(key_click) => { + todo!() + // container_click::Slot::Normal(slot) => { + // self.number_button_pressed(opened_container.as_deref_mut(), key_click, slot).await + // } + // container_click::Slot::OutsideInventory => Err(InventoryError::InvalidPacket), } - ClickType::KeyClick(key_click) => match click.slot { - container_click::Slot::Normal(slot) => { - self.number_button_pressed(opened_container.as_deref_mut(), key_click, slot) - } - container_click::Slot::OutsideInventory => Err(InventoryError::InvalidPacket), - }, ClickType::CreativePickItem => { - if let container_click::Slot::Normal(slot) = click.slot { - self.creative_pick_item(opened_container.as_deref_mut(), slot) - } else { - Err(InventoryError::InvalidPacket) - } + // if let container_click::Slot::Normal(slot) = click.slot { + // self.creative_pick_item(opened_container.as_deref_mut(), slot).await + // } else { + // Err(InventoryError::InvalidPacket) + // } + todo!() } ClickType::DoubleClick => { update_whole_container = true; - if let container_click::Slot::Normal(slot) = click.slot { - self.double_click(opened_container.as_deref_mut(), slot) - } else { - Err(InventoryError::InvalidPacket) - } + // if let container_click::Slot::Normal(slot) = click.slot { + // self.double_click(opened_container.as_deref_mut(), slot) + // } else { + // Err(InventoryError::InvalidPacket) + // } + todo!() } ClickType::MouseDrag { drag_state } => { if drag_state == MouseDragState::End { update_whole_container = true; } - self.mouse_drag(drag_handler, opened_container.as_deref_mut(), drag_state) + todo!() + // self.mouse_drag(drag_handler, opened_container.as_deref_mut(), drag_state) } ClickType::DropType(_drop_type) => { dbg!("todo"); Ok(()) } }?; - if let Some(mut opened_container) = opened_container { + if let Some(opened_container) = opened_container { if update_whole_container { drop(opened_container); self.send_whole_container_change(server).await?; } else if let container_click::Slot::Normal(slot_index) = click.slot { - let mut inventory = self.inventory.lock(); + let mut inventory = self.inventory.lock().await; + let mut opened_container = opened_container.await; let combined_container = OptionallyCombinedContainer::new(&mut inventory, Some(&mut opened_container)); if let Some(slot) = combined_container.get_slot_excluding_inventory(slot_index) { @@ -200,13 +209,13 @@ impl Player { Ok(()) } - fn mouse_click( + async fn mouse_click( &self, opened_container: Option<&mut Box>, mouse_click: MouseClick, slot: container_click::Slot, ) -> Result<(), InventoryError> { - let mut inventory = self.inventory.lock(); + let mut inventory = self.inventory.lock().await; let mut container = OptionallyCombinedContainer::new(&mut inventory, opened_container); match slot { @@ -220,12 +229,12 @@ impl Player { } } - fn shift_mouse_click( + async fn shift_mouse_click( &self, opened_container: Option<&mut Box>, slot: container_click::Slot, ) -> Result<(), InventoryError> { - let mut inventory = self.inventory.lock(); + let mut inventory = self.inventory.lock().await; let mut container = OptionallyCombinedContainer::new(&mut inventory, opened_container); match slot { @@ -267,7 +276,7 @@ impl Player { Ok(()) } - fn number_button_pressed( + async fn number_button_pressed( &self, opened_container: Option<&mut Box>, key_click: KeyClick, @@ -277,7 +286,7 @@ impl Player { KeyClick::Slot(slot) => slot, KeyClick::Offhand => 45, }; - let mut inventory = self.inventory.lock(); + let mut inventory = self.inventory.lock().await; let mut changing_item_slot = inventory.get_slot(changing_slot as usize)?.to_owned(); let mut container = OptionallyCombinedContainer::new(&mut inventory, opened_container); @@ -286,7 +295,7 @@ impl Player { Ok(()) } - fn creative_pick_item( + async fn creative_pick_item( &self, opened_container: Option<&mut Box>, slot: usize, @@ -294,7 +303,7 @@ impl Player { if self.gamemode.load() != GameMode::Creative { return Err(InventoryError::PermissionError); } - let mut inventory = self.inventory.lock(); + let mut inventory = self.inventory.lock().await; let mut container = OptionallyCombinedContainer::new(&mut inventory, opened_container); if let Some(Some(item)) = container.all_slots().get_mut(slot) { self.carried_item.store(Some(item.to_owned())); @@ -302,12 +311,12 @@ impl Player { Ok(()) } - fn double_click( + async fn double_click( &self, opened_container: Option<&mut Box>, slot: usize, ) -> Result<(), InventoryError> { - let mut inventory = self.inventory.lock(); + let mut inventory = self.inventory.lock().await; let mut container = OptionallyCombinedContainer::new(&mut inventory, opened_container); let mut slots = container.all_slots(); @@ -340,7 +349,7 @@ impl Player { Ok(()) } - fn mouse_drag( + async fn mouse_drag( &self, drag_handler: &DragHandler, opened_container: Option<&mut Box>, @@ -357,20 +366,21 @@ impl Player { { Err(InventoryError::PermissionError)? } - drag_handler.new_drag(container_id, player_id, drag_type) + drag_handler + .new_drag(container_id, player_id, drag_type) + .await + } + MouseDragState::AddSlot(slot) => { + drag_handler.add_slot(container_id, player_id, slot).await } - MouseDragState::AddSlot(slot) => drag_handler.add_slot(container_id, player_id, slot), MouseDragState::End => { - let mut inventory = self.inventory.lock(); + let mut inventory = self.inventory.lock().await; let mut container = OptionallyCombinedContainer::new(&mut inventory, opened_container); let mut carried_item = self.carried_item.load(); - let res = drag_handler.apply_drag( - &mut carried_item, - &mut container, - &container_id, - player_id, - ); + let res = drag_handler + .apply_drag(&mut carried_item, &mut container, &container_id, player_id) + .await; self.carried_item.store(carried_item); res } @@ -379,7 +389,7 @@ impl Player { async fn get_current_players_in_container(&self, server: &Server) -> Vec> { let player_ids = { - let open_containers = server.open_containers.read(); + let open_containers = server.open_containers.read().await; open_containers .get(&self.open_container.load().unwrap()) .unwrap() @@ -399,6 +409,7 @@ impl Player { .world .current_players .lock() + .await .iter() .filter_map(|(token, player)| { if *token != player_token { @@ -423,7 +434,7 @@ impl Player { slot: Slot, ) -> Result<(), InventoryError> { for player in self.get_current_players_in_container(server).await { - let inventory = player.inventory.lock(); + let inventory = player.inventory.lock().await; let total_opened_containers = inventory.total_opened_containers; // Returns previous value @@ -446,15 +457,17 @@ impl Player { for player in players { let container = player.get_open_container(server); - let mut container = container.as_ref().map(|v| v.lock()); - player.set_container_content(container.as_deref_mut()); + let container = container.as_ref().map(|v| v.lock()); + // player.set_container_content(container.as_deref_mut()); } Ok(()) } - pub fn get_open_container(&self, server: &Server) -> Option>>> { - self.open_container - .load() - .map_or_else(|| None, |id| server.try_get_container(self.entity_id(), id)) + pub fn get_open_container( + &self, + server: &Server, + ) -> Option>>> { + // self.open_container .load().map_or_else(|| None, |id| server.try_get_container(self.entity_id(), id).await) + None } } diff --git a/pumpkin/src/client/mod.rs b/pumpkin/src/client/mod.rs index b2bb4f47b..9bf10162c 100644 --- a/pumpkin/src/client/mod.rs +++ b/pumpkin/src/client/mod.rs @@ -1,6 +1,5 @@ use std::{ collections::VecDeque, - io::{self, Write}, net::SocketAddr, sync::{ atomic::{AtomicBool, AtomicI32}, @@ -15,8 +14,6 @@ use crate::{ use authentication::GameProfile; use crossbeam::atomic::AtomicCell; -use mio::{event::Event, net::TcpStream}; -use parking_lot::Mutex; use pumpkin_config::compression::CompressionInfo; use pumpkin_core::text::TextComponent; use pumpkin_protocol::{ @@ -32,8 +29,9 @@ use pumpkin_protocol::{ }, ClientPacket, ConnectionState, PacketError, RawPacket, ServerPacket, }; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::sync::Mutex; -use std::io::Read; use thiserror::Error; pub mod authentication; @@ -105,7 +103,9 @@ pub struct Client { /// A unique id identifying the client. pub id: usize, /// The underlying TCP connection to the client. - pub connection: Arc>, + pub connection_reader: Arc>, + pub connection_writer: Arc>, + /// The client's IP address. pub address: Mutex, /// The packet encoder for outgoing packets. @@ -126,10 +126,11 @@ pub struct Client { impl Client { pub fn new( id: usize, - connection: TcpStream, + connection: tokio::net::TcpStream, address: SocketAddr, keep_alive_sender: Arc>, ) -> Self { + let (connection_reader, connection_writer) = connection.into_split(); Self { protocol_version: AtomicI32::new(0), gameprofile: Mutex::new(None), @@ -139,7 +140,8 @@ impl Client { id, address: Mutex::new(address), connection_state: AtomicCell::new(ConnectionState::HandShake), - connection: Arc::new(Mutex::new(connection)), + connection_reader: Arc::new(Mutex::new(connection_reader)), + connection_writer: Arc::new(Mutex::new(connection_writer)), enc: Arc::new(Mutex::new(PacketEncoder::default())), dec: Arc::new(Mutex::new(PacketDecoder::default())), encryption: AtomicBool::new(false), @@ -152,13 +154,13 @@ impl Client { } /// Adds a Incoming packet to the queue - pub fn add_packet(&self, packet: RawPacket) { - let mut client_packets_queue = self.client_packets_queue.lock(); + pub async fn add_packet(&self, packet: RawPacket) { + let mut client_packets_queue = self.client_packets_queue.lock().await; client_packets_queue.push_back(packet); } /// Sets the Packet encryption - pub fn set_encryption( + pub async fn set_encryption( &self, shared_secret: Option<&[u8]>, // decrypted ) -> Result<(), EncryptionError> { @@ -168,55 +170,64 @@ impl Client { let crypt_key: [u8; 16] = shared_secret .try_into() .map_err(|_| EncryptionError::SharedWrongLength)?; - self.dec.lock().set_encryption(Some(&crypt_key)); - self.enc.lock().set_encryption(Some(&crypt_key)); + self.dec.lock().await.set_encryption(Some(&crypt_key)); + self.enc.lock().await.set_encryption(Some(&crypt_key)); } else { - self.dec.lock().set_encryption(None); - self.enc.lock().set_encryption(None); + self.dec.lock().await.set_encryption(None); + self.enc.lock().await.set_encryption(None); } Ok(()) } /// Sets the Packet compression - pub fn set_compression(&self, compression: Option) { - self.dec.lock().set_compression(compression.is_some()); - self.enc.lock().set_compression(compression); + pub async fn set_compression(&self, compression: Option) { + self.dec.lock().await.set_compression(compression.is_some()); + self.enc.lock().await.set_compression(compression); } /// Send a Clientbound Packet to the Client - pub fn send_packet(&self, packet: &P) { + pub async fn send_packet(&self, packet: &P) { // assert!(!self.closed); - let mut enc = self.enc.lock(); - enc.append_packet(packet) - .unwrap_or_else(|e| self.kick(&e.to_string())); - self.connection + let mut enc = self.enc.lock().await; + if let Err(error) = enc.append_packet(packet) { + self.kick(&error.to_string()).await; + return; + } + if let Err(error) = self + .connection_writer .lock() + .await .write_all(&enc.take()) - .map_err(|e| PacketError::ConnectionWrite(e.to_string())) - .unwrap_or_else(|e| self.kick(&e.to_string())); + .await + .map_err(|_| PacketError::ConnectionWrite) + { + self.kick(&error.to_string()).await; + } } - pub fn try_send_packet(&self, packet: &P) -> Result<(), PacketError> { + pub async fn try_send_packet(&self, packet: &P) -> Result<(), PacketError> { // assert!(!self.closed); - let mut enc = self.enc.lock(); + let mut enc = self.enc.lock().await; enc.append_packet(packet)?; - self.connection + self.connection_writer .lock() + .await .write_all(&enc.take()) - .map_err(|e| PacketError::ConnectionWrite(e.to_string()))?; + .await + .map_err(|_| PacketError::ConnectionWrite)?; Ok(()) } /// Processes all packets send by the client pub async fn process_packets(&self, server: &Arc) { - while let Some(mut packet) = self.client_packets_queue.lock().pop_front() { - let _ = self.handle_packet(server, &mut packet).await.map_err(|e| { + while let Some(mut packet) = self.client_packets_queue.lock().await.pop_front() { + if let Err(error) = self.handle_packet(server, &mut packet).await { dbg!("{:?}", packet.id); - let text = format!("Error while reading incoming packet {}", e); + let text = format!("Error while reading incoming packet {}", error); log::error!("{}", text); - self.kick(&text) - }); + self.kick(&text).await + }; } } @@ -227,8 +238,12 @@ impl Client { packet: &mut RawPacket, ) -> Result<(), DeserializerError> { match self.connection_state.load() { - pumpkin_protocol::ConnectionState::HandShake => self.handle_handshake_packet(packet), - pumpkin_protocol::ConnectionState::Status => self.handle_status_packet(server, packet), + pumpkin_protocol::ConnectionState::HandShake => { + self.handle_handshake_packet(packet).await + } + pumpkin_protocol::ConnectionState::Status => { + self.handle_status_packet(server, packet).await + } // TODO: Check config if transfer is enabled pumpkin_protocol::ConnectionState::Login | pumpkin_protocol::ConnectionState::Transfer => { @@ -244,11 +259,14 @@ impl Client { } } - fn handle_handshake_packet(&self, packet: &mut RawPacket) -> Result<(), DeserializerError> { + async fn handle_handshake_packet( + &self, + packet: &mut RawPacket, + ) -> Result<(), DeserializerError> { let bytebuf = &mut packet.bytebuf; match packet.id.0 { SHandShake::PACKET_ID => { - self.handle_handshake(SHandShake::read(bytebuf)?); + self.handle_handshake(SHandShake::read(bytebuf)?).await; Ok(()) } _ => { @@ -261,7 +279,7 @@ impl Client { } } - fn handle_status_packet( + async fn handle_status_packet( &self, server: &Arc, packet: &mut RawPacket, @@ -269,11 +287,13 @@ impl Client { let bytebuf = &mut packet.bytebuf; match packet.id.0 { SStatusRequest::PACKET_ID => { - self.handle_status_request(server, SStatusRequest::read(bytebuf)?); + self.handle_status_request(server, SStatusRequest::read(bytebuf)?) + .await; Ok(()) } SStatusPingRequest::PACKET_ID => { - self.handle_ping_request(SStatusPingRequest::read(bytebuf)?); + self.handle_ping_request(SStatusPingRequest::read(bytebuf)?) + .await; Ok(()) } _ => { @@ -294,7 +314,8 @@ impl Client { let bytebuf = &mut packet.bytebuf; match packet.id.0 { SLoginStart::PACKET_ID => { - self.handle_login_start(server, SLoginStart::read(bytebuf)?); + self.handle_login_start(server, SLoginStart::read(bytebuf)?) + .await; Ok(()) } SEncryptionResponse::PACKET_ID => { @@ -303,11 +324,13 @@ impl Client { Ok(()) } SLoginPluginResponse::PACKET_ID => { - self.handle_plugin_response(SLoginPluginResponse::read(bytebuf)?); + self.handle_plugin_response(SLoginPluginResponse::read(bytebuf)?) + .await; Ok(()) } SLoginAcknowledged::PACKET_ID => { - self.handle_login_acknowledged(server, SLoginAcknowledged::read(bytebuf)?); + self.handle_login_acknowledged(server, SLoginAcknowledged::read(bytebuf)?) + .await; Ok(()) } _ => { @@ -328,11 +351,13 @@ impl Client { let bytebuf = &mut packet.bytebuf; match packet.id.0 { SClientInformationConfig::PACKET_ID => { - self.handle_client_information_config(SClientInformationConfig::read(bytebuf)?); + self.handle_client_information_config(SClientInformationConfig::read(bytebuf)?) + .await; Ok(()) } SPluginMessage::PACKET_ID => { - self.handle_plugin_message(SPluginMessage::read(bytebuf)?); + self.handle_plugin_message(SPluginMessage::read(bytebuf)?) + .await; Ok(()) } SAcknowledgeFinishConfig::PACKET_ID => { @@ -341,7 +366,8 @@ impl Client { Ok(()) } SKnownPacks::PACKET_ID => { - self.handle_known_packs(server, SKnownPacks::read(bytebuf)?); + self.handle_known_packs(server, SKnownPacks::read(bytebuf)?) + .await; Ok(()) } _ => { @@ -356,70 +382,59 @@ impl Client { /// Reads the connection until our buffer of len 4096 is full, then decode /// Close connection when an error occurs or when the Client closed the connection - pub async fn poll(&self, event: &Event) { - if event.is_readable() { - let mut received_data = vec![]; - let mut buf = [0; 4096]; - loop { - let connection = self.connection.clone(); - let mut connection = connection.lock(); - match connection.read(&mut buf) { - Ok(0) => { - // Reading 0 bytes means the other side has closed the - // connection or is done writing, then so are we. - self.close(); - break; - } - Ok(n) => received_data.extend(&buf[..n]), - // Would block "errors" are the OS's way of saying that the - // connection is not actually ready to perform this I/O operation. - Err(ref err) if would_block(err) => break, - Err(ref err) if interrupted(err) => continue, - // Other errors we'll consider fatal. - Err(_) => self.close(), + pub async fn poll(&self) { + loop { + let mut dec = self.dec.lock().await; + if let Ok(packet) = dec.decode() { + if let Some(packet) = packet { + dbg!("a"); + self.add_packet(packet).await; + return; } + }; + + dec.reserve(4096); + let mut buf = dec.take_capacity(); + + if self + .connection_reader + .lock() + .await + .read_buf(&mut buf) + .await + .unwrap() + == 0 + { + self.close(); + return; } - if !received_data.is_empty() { - let mut dec = self.dec.lock(); - dec.queue_slice(&received_data); - loop { - match dec.decode() { - Ok(packet) => { - if let Some(packet) = packet { - self.add_packet(packet); - } else { - break; - } - } - Err(err) => { - self.kick(&err.to_string()); - break; - } - } - } - dec.clear(); - } + // This should always be an O(1) unsplit because we reserved space earlier and + // the call to `read_buf` shouldn't have grown the allocation. + dec.queue_bytes(buf); } } /// Kicks the Client with a reason depending on the connection state - pub fn kick(&self, reason: &str) { + pub async fn kick(&self, reason: &str) { dbg!(reason); match self.connection_state.load() { ConnectionState::Login => { self.try_send_packet(&CLoginDisconnect::new( &serde_json::to_string_pretty(&reason).unwrap_or_else(|_| "".into()), )) + .await .unwrap_or_else(|_| self.close()); } ConnectionState::Config => { self.try_send_packet(&CConfigDisconnect::new(reason)) + .await .unwrap_or_else(|_| self.close()); } // This way players get kicked when players using client functions (e.g. poll, send_packet) ConnectionState::Play => { self.try_send_packet(&CPlayDisconnect::new(&TextComponent::text(reason))) + .await .unwrap_or_else(|_| self.close()); } _ => { @@ -443,11 +458,3 @@ pub enum EncryptionError { #[error("shared secret has the wrong length")] SharedWrongLength, } - -fn would_block(err: &io::Error) -> bool { - err.kind() == io::ErrorKind::WouldBlock -} - -pub fn interrupted(err: &io::Error) -> bool { - err.kind() == io::ErrorKind::Interrupted -} diff --git a/pumpkin/src/client/player_packet.rs b/pumpkin/src/client/player_packet.rs index ee9920f9e..cc5fdc819 100644 --- a/pumpkin/src/client/player_packet.rs +++ b/pumpkin/src/client/player_packet.rs @@ -41,8 +41,8 @@ fn modulus(a: f32, b: f32) -> f32 { /// Handles all Play Packets send by a real Player /// NEVER TRUST THE CLIENT. HANDLE EVERY ERROR, UNWRAP/EXPECT ARE FORBIDDEN impl Player { - pub fn handle_confirm_teleport(&self, confirm_teleport: SConfirmTeleport) { - let mut awaiting_teleport = self.awaiting_teleport.lock(); + pub async fn handle_confirm_teleport(&self, confirm_teleport: SConfirmTeleport) { + let mut awaiting_teleport = self.awaiting_teleport.lock().await; if let Some((id, position)) = awaiting_teleport.as_ref() { if id == &confirm_teleport.teleport_id { // we should set the pos now to that we requested in the teleport packet, Is may fixed issues when the client sended position packets while being teleported @@ -52,12 +52,13 @@ impl Player { *awaiting_teleport = None; } else { - self.kick(TextComponent::text("Wrong teleport id")) + self.kick(TextComponent::text("Wrong teleport id")).await } } else { self.kick(TextComponent::text( "Send Teleport confirm, but we did not teleport", )) + .await } } @@ -71,7 +72,7 @@ impl Player { pub async fn handle_position(&self, position: SPlayerPosition) { if position.x.is_nan() || position.feet_y.is_nan() || position.z.is_nan() { - self.kick(TextComponent::text("Invalid movement")); + self.kick(TextComponent::text("Invalid movement")).await; return; } let entity = &self.living_entity.entity; @@ -104,16 +105,18 @@ impl Player { // return; // } // send new position to all other players - world.broadcast_packet_expect( - &[self.client.id], - &CUpdateEntityPos::new( - entity_id.into(), - x.mul_add(4096.0, -(lastx * 4096.0)) as i16, - y.mul_add(4096.0, -(lasty * 4096.0)) as i16, - z.mul_add(4096.0, -(lastz * 4096.0)) as i16, - position.ground, - ), - ); + world + .broadcast_packet_expect( + &[self.client.id], + &CUpdateEntityPos::new( + entity_id.into(), + x.mul_add(4096.0, -(lastx * 4096.0)) as i16, + y.mul_add(4096.0, -(lasty * 4096.0)) as i16, + z.mul_add(4096.0, -(lastz * 4096.0)) as i16, + position.ground, + ), + ) + .await; player_chunker::update_position(entity, self).await; } @@ -122,11 +125,11 @@ impl Player { || position_rotation.feet_y.is_nan() || position_rotation.z.is_nan() { - self.kick(TextComponent::text("Invalid movement")); + self.kick(TextComponent::text("Invalid movement")).await; return; } if position_rotation.yaw.is_infinite() || position_rotation.pitch.is_infinite() { - self.kick(TextComponent::text("Invalid rotation")); + self.kick(TextComponent::text("Invalid rotation")).await; return; } let entity = &self.living_entity.entity; @@ -170,28 +173,32 @@ impl Player { // } // send new position to all other players - world.broadcast_packet_expect( - &[self.client.id], - &CUpdateEntityPosRot::new( - entity_id.into(), - x.mul_add(4096.0, -(lastx * 4096.0)) as i16, - y.mul_add(4096.0, -(lasty * 4096.0)) as i16, - z.mul_add(4096.0, -(lastz * 4096.0)) as i16, - yaw as u8, - pitch as u8, - position_rotation.ground, - ), - ); - world.broadcast_packet_expect( - &[self.client.id], - &CHeadRot::new(entity_id.into(), yaw as u8), - ); + world + .broadcast_packet_expect( + &[self.client.id], + &CUpdateEntityPosRot::new( + entity_id.into(), + x.mul_add(4096.0, -(lastx * 4096.0)) as i16, + y.mul_add(4096.0, -(lasty * 4096.0)) as i16, + z.mul_add(4096.0, -(lastz * 4096.0)) as i16, + yaw as u8, + pitch as u8, + position_rotation.ground, + ), + ) + .await; + world + .broadcast_packet_expect( + &[self.client.id], + &CHeadRot::new(entity_id.into(), yaw as u8), + ) + .await; player_chunker::update_position(entity, self).await; } pub async fn handle_rotation(&self, rotation: SPlayerRotation) { if !rotation.yaw.is_finite() || !rotation.pitch.is_finite() { - self.kick(TextComponent::text("Invalid rotation")); + self.kick(TextComponent::text("Invalid rotation")).await; return; } let entity = &self.living_entity.entity; @@ -211,9 +218,13 @@ impl Player { let world = &entity.world; let packet = CUpdateEntityRot::new(entity_id.into(), yaw as u8, pitch as u8, rotation.ground); - world.broadcast_packet_expect(&[self.client.id], &packet); + world + .broadcast_packet_expect(&[self.client.id], &packet) + .await; let packet = CHeadRot::new(entity_id.into(), yaw as u8); - world.broadcast_packet_expect(&[self.client.id], &packet); + world + .broadcast_packet_expect(&[self.client.id], &packet) + .await; } pub fn handle_chat_command(&self, server: &Server, command: SChatCommand) { @@ -280,6 +291,7 @@ impl Player { } } else { self.kick(TextComponent::text("Invalid player command")) + .await } } @@ -292,13 +304,15 @@ impl Player { }; let id = self.entity_id(); let world = &self.living_entity.entity.world; - world.broadcast_packet_expect( - &[self.client.id], - &CEntityAnimation::new(id.into(), animation as u8), - ) + world + .broadcast_packet_expect( + &[self.client.id], + &CEntityAnimation::new(id.into(), animation as u8), + ) + .await } None => { - self.kick(TextComponent::text("Invalid hand")); + self.kick(TextComponent::text("Invalid hand")).await; } }; } @@ -308,7 +322,7 @@ impl Player { let message = chat_message.message; if message.len() > 256 { - self.kick(TextComponent::text("Oversized message")); + self.kick(TextComponent::text("Oversized message")).await; return; } @@ -317,20 +331,22 @@ impl Player { let entity = &self.living_entity.entity; let world = &entity.world; - world.broadcast_packet_all(&CPlayerChatMessage::new( - gameprofile.id, - 1.into(), - chat_message.signature.as_deref(), - &message, - chat_message.timestamp, - chat_message.salt, - &[], - Some(TextComponent::text(&message)), - FilterType::PassThrough, - 1.into(), - TextComponent::text(&gameprofile.name), - None, - )) + world + .broadcast_packet_all(&CPlayerChatMessage::new( + gameprofile.id, + 1.into(), + chat_message.signature.as_deref(), + &message, + chat_message.timestamp, + chat_message.salt, + &[], + Some(TextComponent::text(&message)), + FilterType::PassThrough, + 1.into(), + TextComponent::text(&gameprofile.name), + None, + )) + .await /* server.broadcast_packet( self, @@ -343,12 +359,12 @@ impl Player { ) */ } - pub fn handle_client_information_play(&self, client_information: SClientInformationPlay) { + pub async fn handle_client_information_play(&self, client_information: SClientInformationPlay) { if let (Some(main_hand), Some(chat_mode)) = ( Hand::from_i32(client_information.main_hand.into()), ChatMode::from_i32(client_information.chat_mode.into()), ) { - *self.config.lock() = PlayerConfig { + *self.config.lock().await = PlayerConfig { locale: client_information.locale, view_distance: client_information.view_distance, chat_mode, @@ -360,6 +376,7 @@ impl Player { }; } else { self.kick(TextComponent::text("Invalid hand or chat type")) + .await } } @@ -377,7 +394,8 @@ impl Player { let config = &ADVANCED_CONFIG.pvp; if config.enabled { let world = &entity.world; - let attacked_player = world.get_player_by_entityid(entity_id.0 as EntityId); + let attacked_player = + world.get_player_by_entityid(entity_id.0 as EntityId).await; if let Some(player) = attacked_player { let victem_entity = &player.living_entity.entity; if config.protect_creative @@ -407,17 +425,20 @@ impl Player { .store(velocity.multiply(0.6, 1.0, 0.6)); victem_entity.velocity.store(saved_velo); - player.client.send_packet(packet); + player.client.send_packet(packet).await; } if config.hurt_animation { - world.broadcast_packet_all(&CHurtAnimation::new( - &entity_id, - entity.yaw.load(), - )) + world + .broadcast_packet_all(&CHurtAnimation::new( + &entity_id, + entity.yaw.load(), + )) + .await } if config.swing {} } else { self.kick(TextComponent::text("Interacted with invalid entity id")) + .await } } } @@ -428,7 +449,7 @@ impl Player { dbg!("todo"); } }, - None => self.kick(TextComponent::text("Invalid action type")), + None => self.kick(TextComponent::text("Invalid action type")).await, } } pub async fn handle_player_action(&self, player_action: SPlayerAction) { @@ -447,9 +468,13 @@ impl Player { // TODO: currently this is always dirt replace it let entity = &self.living_entity.entity; let world = &entity.world; - world.broadcast_packet_all(&CWorldEvent::new(2001, &location, 11, false)); + world + .broadcast_packet_all(&CWorldEvent::new(2001, &location, 11, false)) + .await; // AIR - world.broadcast_packet_all(&CBlockUpdate::new(&location, 0.into())); + world + .broadcast_packet_all(&CBlockUpdate::new(&location, 0.into())) + .await; } } Status::CancelledDigging => { @@ -471,12 +496,17 @@ impl Player { // TODO: currently this is always dirt replace it let entity = &self.living_entity.entity; let world = &entity.world; - world.broadcast_packet_all(&CWorldEvent::new(2001, &location, 11, false)); + world + .broadcast_packet_all(&CWorldEvent::new(2001, &location, 11, false)) + .await; // AIR - world.broadcast_packet_all(&CBlockUpdate::new(&location, 0.into())); + world + .broadcast_packet_all(&CBlockUpdate::new(&location, 0.into())) + .await; // TODO: Send this every tick self.client - .send_packet(&CAcknowledgeBlockChange::new(player_action.sequence)); + .send_packet(&CAcknowledgeBlockChange::new(player_action.sequence)) + .await; } Status::DropItemStack => { dbg!("todo"); @@ -491,13 +521,14 @@ impl Player { dbg!("todo"); } }, - None => self.kick(TextComponent::text("Invalid status")), + None => self.kick(TextComponent::text("Invalid status")).await, } } - pub fn handle_play_ping_request(&self, request: SPlayPingRequest) { + pub async fn handle_play_ping_request(&self, request: SPlayPingRequest) { self.client - .send_packet(&CPingResponse::new(request.payload)); + .send_packet(&CPingResponse::new(request.payload)) + .await; } pub async fn handle_use_item_on(&self, use_item_on: SUseItemOn) { @@ -509,7 +540,7 @@ impl Player { } if let Some(face) = BlockFace::from_i32(use_item_on.face.0) { - if let Some(item) = self.inventory.lock().held_item() { + if let Some(item) = self.inventory.lock().await.held_item() { let minecraft_id = global_registry::find_minecraft_id( global_registry::ITEM_REGISTRY, item.item_id, @@ -518,20 +549,25 @@ impl Player { if let Ok(block_state_id) = BlockState::new(minecraft_id, None) { let entity = &self.living_entity.entity; let world = &entity.world; - world.broadcast_packet_all(&CBlockUpdate::new( - &location, - block_state_id.get_id_mojang_repr().into(), - )); - world.broadcast_packet_all(&CBlockUpdate::new( - &WorldPosition(location.0 + face.to_offset()), - block_state_id.get_id_mojang_repr().into(), - )); + world + .broadcast_packet_all(&CBlockUpdate::new( + &location, + block_state_id.get_id_mojang_repr().into(), + )) + .await; + world + .broadcast_packet_all(&CBlockUpdate::new( + &WorldPosition(location.0 + face.to_offset()), + block_state_id.get_id_mojang_repr().into(), + )) + .await; } } self.client - .send_packet(&CAcknowledgeBlockChange::new(use_item_on.sequence)); + .send_packet(&CAcknowledgeBlockChange::new(use_item_on.sequence)) + .await; } else { - self.kick(TextComponent::text("Invalid block face")) + self.kick(TextComponent::text("Invalid block face")).await } } @@ -540,42 +576,49 @@ impl Player { log::error!("An item was used(SUseItem), but the packet is not implemented yet"); } - pub fn handle_set_held_item(&self, held: SSetHeldItem) { + pub async fn handle_set_held_item(&self, held: SSetHeldItem) { let slot = held.slot; if !(0..=8).contains(&slot) { - self.kick(TextComponent::text("Invalid held slot")) + self.kick(TextComponent::text("Invalid held slot")).await; + return; } - self.inventory.lock().set_selected(slot as usize); + self.inventory.lock().await.set_selected(slot as usize); } - pub fn handle_set_creative_slot(&self, packet: SSetCreativeSlot) -> Result<(), InventoryError> { + pub async fn handle_set_creative_slot( + &self, + packet: SSetCreativeSlot, + ) -> Result<(), InventoryError> { if self.gamemode.load() != GameMode::Creative { return Err(InventoryError::PermissionError); } - self.inventory - .lock() - .set_slot(packet.slot as usize, packet.clicked_item.to_item(), false) + self.inventory.lock().await.set_slot( + packet.slot as usize, + packet.clicked_item.to_item(), + false, + ) } // TODO: // This function will in the future be used to keep track of if the client is in a valid state. // But this is not possible yet - pub fn handle_close_container(&self, server: &Server, packet: SCloseContainer) { + pub async fn handle_close_container(&self, server: &Server, packet: SCloseContainer) { // window_id 0 represents both 9x1 Generic AND inventory here self.inventory .lock() + .await .state_id .store(0, std::sync::atomic::Ordering::Relaxed); let open_container = self.open_container.load(); if let Some(id) = open_container { - let mut open_containers = server.open_containers.write(); + let mut open_containers = server.open_containers.write().await; if let Some(container) = open_containers.get_mut(&id) { container.remove_player(self.entity_id()) } self.open_container.store(None); } let Some(_window_type) = WindowType::from_u8(packet.window_id) else { - self.kick(TextComponent::text("Invalid window ID")); + self.kick(TextComponent::text("Invalid window ID")).await; return; }; } diff --git a/pumpkin/src/commands/cmd_echest.rs b/pumpkin/src/commands/cmd_echest.rs index e793ccb8a..0c6b0f77f 100644 --- a/pumpkin/src/commands/cmd_echest.rs +++ b/pumpkin/src/commands/cmd_echest.rs @@ -1,5 +1,3 @@ -use pumpkin_inventory::OpenContainer; - use crate::commands::tree::CommandTree; const NAMES: [&str; 2] = ["echest", "enderchest"]; @@ -11,20 +9,20 @@ pub fn init_command_tree<'a>() -> CommandTree<'a> { CommandTree::new(NAMES, DESCRIPTION).execute(&|sender, server, _| { if let Some(player) = sender.as_mut_player() { let entity_id = player.entity_id(); - player.open_container.store(Some(0)); - { - let mut open_containers = server.open_containers.write(); - match open_containers.get_mut(&0) { - Some(ender_chest) => { - ender_chest.add_player(entity_id); - } - None => { - let open_container = OpenContainer::empty(entity_id); - open_containers.insert(0, open_container); - } - } - } - player.open_container(server, "minecraft:generic_9x3"); + // player.open_container.store(Some(0)); + // { + // let mut open_containers = server.open_containers.write().await; + // match open_containers.get_mut(&0) { + // Some(ender_chest) => { + // ender_chest.add_player(entity_id); + // } + // None => { + // let open_container = OpenContainer::empty(entity_id); + // open_containers.insert(0, open_container); + // } + // } + // } + // player.open_container(server, "minecraft:generic_9x3"); } Ok(()) diff --git a/pumpkin/src/commands/cmd_help.rs b/pumpkin/src/commands/cmd_help.rs index a955f4c8c..f224f0854 100644 --- a/pumpkin/src/commands/cmd_help.rs +++ b/pumpkin/src/commands/cmd_help.rs @@ -3,7 +3,6 @@ use crate::commands::dispatcher::{CommandDispatcher, InvalidTreeError}; use crate::commands::tree::{Command, CommandTree, ConsumedArgs, RawArgs}; use crate::commands::tree_builder::argument; use crate::commands::CommandSender; -use pumpkin_core::text::TextComponent; const NAMES: [&str; 3] = ["help", "h", "?"]; @@ -38,12 +37,12 @@ pub fn init_command_tree<'a>() -> CommandTree<'a> { argument(ARG_COMMAND, consume_arg_command).execute(&|sender, server, args| { let tree = parse_arg_command(args, &server.command_dispatcher)?; - sender.send_message(TextComponent::text(&format!( - "{} - {} Usage: {}", - tree.names.join("/"), - tree.description, - tree - ))); + // sender.send_message(TextComponent::text(&format!( + // "{} - {} Usage: {}", + // tree.names.join("/"), + // tree.description, + // tree + // ))); Ok(()) }), @@ -57,12 +56,12 @@ pub fn init_command_tree<'a>() -> CommandTree<'a> { continue; }; - sender.send_message(TextComponent::text(&format!( - "{} - {} Usage: {}", - tree.names.join("/"), - tree.description, - tree - ))); + // sender.send_message(TextComponent::text(&format!( + // "{} - {} Usage: {}", + // tree.names.join("/"), + // tree.description, + // tree + // ))); } Ok(()) diff --git a/pumpkin/src/commands/cmd_pumpkin.rs b/pumpkin/src/commands/cmd_pumpkin.rs index 0d8d7565f..b13da29ff 100644 --- a/pumpkin/src/commands/cmd_pumpkin.rs +++ b/pumpkin/src/commands/cmd_pumpkin.rs @@ -1,7 +1,3 @@ -use crate::server::CURRENT_MC_VERSION; -use pumpkin_core::text::{color::NamedColor, TextComponent}; -use pumpkin_protocol::CURRENT_MC_PROTOCOL; - use crate::commands::tree::CommandTree; const NAMES: [&str; 1] = ["pumpkin"]; @@ -13,9 +9,9 @@ pub fn init_command_tree<'a>() -> CommandTree<'a> { let version = env!("CARGO_PKG_VERSION"); let description = env!("CARGO_PKG_DESCRIPTION"); - sender.send_message(TextComponent::text( - &format!("Pumpkin {version}, {description} (Minecraft {CURRENT_MC_VERSION}, Protocol {CURRENT_MC_PROTOCOL})") - ).color_named(NamedColor::Green)); + // sender.send_message(TextComponent::text( + // &format!("Pumpkin {version}, {description} (Minecraft {CURRENT_MC_VERSION}, Protocol {CURRENT_MC_PROTOCOL})") + // ).color_named(NamedColor::Green)); Ok(()) }) diff --git a/pumpkin/src/commands/dispatcher.rs b/pumpkin/src/commands/dispatcher.rs index a636e8144..282551253 100644 --- a/pumpkin/src/commands/dispatcher.rs +++ b/pumpkin/src/commands/dispatcher.rs @@ -1,5 +1,3 @@ -use pumpkin_core::text::TextComponent; - use crate::commands::dispatcher::InvalidTreeError::{ InvalidConsumptionError, InvalidRequirementError, }; @@ -28,9 +26,9 @@ pub struct CommandDispatcher<'a> { impl<'a> CommandDispatcher<'a> { pub fn handle_command(&self, sender: &mut CommandSender, server: &Server, cmd: &str) { if let Err(err) = self.dispatch(sender, server, cmd) { - sender.send_message( - TextComponent::text(&err).color_named(pumpkin_core::text::color::NamedColor::Red), - ) + // sender.send_message( + // TextComponent::text(&err).color_named(pumpkin_core::text::color::NamedColor::Red), + // ) } } diff --git a/pumpkin/src/commands/mod.rs b/pumpkin/src/commands/mod.rs index a0e10041f..3f7bce6c8 100644 --- a/pumpkin/src/commands/mod.rs +++ b/pumpkin/src/commands/mod.rs @@ -24,11 +24,11 @@ pub enum CommandSender<'a> { } impl<'a> CommandSender<'a> { - pub fn send_message(&mut self, text: TextComponent) { + pub async fn send_message(&mut self, text: TextComponent<'a>) { match self { // TODO: add color and stuff to console CommandSender::Console => log::info!("{}", text.to_pretty_console()), - CommandSender::Player(c) => c.send_system_message(text), + CommandSender::Player(c) => c.send_system_message(text).await, CommandSender::Rcon(s) => s.push(text.to_pretty_console()), } } diff --git a/pumpkin/src/entity/living.rs b/pumpkin/src/entity/living.rs index a6ae33ede..74b3e8b44 100644 --- a/pumpkin/src/entity/living.rs +++ b/pumpkin/src/entity/living.rs @@ -18,7 +18,7 @@ impl LivingEntity { } } - pub fn set_health(&self, health: f32) { + pub async fn set_health(&self, health: f32) { self.health.store(health); // tell everyone entities health changed self.entity @@ -26,21 +26,24 @@ impl LivingEntity { .broadcast_packet_all(&CSetEntityMetadata::new( self.entity.entity_id.into(), Metadata::new(9, 3.into(), health), - )); + )) + .await; } /// Kills the Entity /// /// This is similar to `kill` but Spawn Particles, Animation and plays death sound - pub fn kill(&self) { + pub async fn kill(&self) { // Spawns death smoke particles self.entity .world - .broadcast_packet_all(&CEntityStatus::new(self.entity.entity_id, 60)); + .broadcast_packet_all(&CEntityStatus::new(self.entity.entity_id, 60)) + .await; // Plays the death sound and death animation self.entity .world - .broadcast_packet_all(&CEntityStatus::new(self.entity.entity_id, 3)); - self.entity.remove(); + .broadcast_packet_all(&CEntityStatus::new(self.entity.entity_id, 3)) + .await; + self.entity.remove().await; } } diff --git a/pumpkin/src/entity/mod.rs b/pumpkin/src/entity/mod.rs index 9ffcbb94a..1deb672cd 100644 --- a/pumpkin/src/entity/mod.rs +++ b/pumpkin/src/entity/mod.rs @@ -122,8 +122,8 @@ impl Entity { } /// Removes the Entity from their current World - pub fn remove(&self) { - self.world.remove_entity(self); + pub async fn remove(&self) { + self.world.remove_entity(self).await; } /// Applies knockback to the entity, following vanilla Minecraft's mechanics. @@ -190,7 +190,7 @@ impl Entity { b &= !(1 << index); } let packet = CSetEntityMetadata::new(self.entity_id.into(), Metadata::new(0, 0.into(), b)); - self.world.broadcast_packet_all(&packet); + self.world.broadcast_packet_all(&packet).await; } pub async fn set_pose(&self, pose: EntityPose) { @@ -200,7 +200,7 @@ impl Entity { self.entity_id.into(), Metadata::new(6, 20.into(), (pose).into()), ); - self.world.broadcast_packet_all(&packet) + self.world.broadcast_packet_all(&packet).await } } diff --git a/pumpkin/src/entity/player.rs b/pumpkin/src/entity/player.rs index 865c8a58a..387e5c267 100644 --- a/pumpkin/src/entity/player.rs +++ b/pumpkin/src/entity/player.rs @@ -6,7 +6,6 @@ use std::sync::{ use crossbeam::atomic::AtomicCell; use num_derive::FromPrimitive; use num_traits::ToPrimitive; -use parking_lot::Mutex; use pumpkin_core::{ math::{boundingbox::BoundingBox, position::WorldPosition, vector3::Vector3}, text::TextComponent, @@ -28,6 +27,7 @@ use pumpkin_protocol::{ }, RawPacket, ServerPacket, VarInt, }; +use tokio::sync::Mutex; use pumpkin_protocol::server::play::{SCloseContainer, SKeepAlive}; use pumpkin_world::item::ItemStack; @@ -91,13 +91,13 @@ pub struct Player { } impl Player { - pub fn new( + pub async fn new( client: Arc, world: Arc, entity_id: EntityId, gamemode: GameMode, ) -> Self { - let gameprofile = client.gameprofile.lock().clone().map_or_else( + let gameprofile = client.gameprofile.lock().await.clone().map_or_else( || { log::error!("No gameprofile?. Impossible"); GameProfile { @@ -109,7 +109,7 @@ impl Player { }, |profile| profile, ); - let config = client.config.lock().clone().unwrap_or_default(); + let config = client.config.lock().await.clone().unwrap_or_default(); Self { living_entity: LivingEntity::new(Entity::new( entity_id, @@ -138,7 +138,7 @@ impl Player { /// Removes the Player out of the current World pub async fn remove(&self) { - self.living_entity.entity.world.remove_player(self); + self.living_entity.entity.world.remove_player(self).await; } pub const fn entity_id(&self) -> EntityId { @@ -146,7 +146,7 @@ impl Player { } /// Updates the current abilities the Player has - pub fn send_abilties_update(&mut self) { + pub async fn send_abilties_update(&mut self) { let mut b = 0i8; let abilities = &self.abilities; @@ -162,14 +162,16 @@ impl Player { if abilities.creative { b |= 8; } - self.client.send_packet(&CPlayerAbilities::new( - b, - abilities.fly_speed, - abilities.walk_speed_fov, - )); + self.client + .send_packet(&CPlayerAbilities::new( + b, + abilities.fly_speed, + abilities.walk_speed_fov, + )) + .await; } - pub fn teleport(&self, x: f64, y: f64, z: f64, yaw: f32, pitch: f32) { + pub async fn teleport(&self, x: f64, y: f64, z: f64, yaw: f32, pitch: f32) { // this is the ultra special magic code used to create the teleport id // This returns the old value let i = self @@ -183,16 +185,18 @@ impl Player { let entity = &self.living_entity.entity; entity.set_pos(x, y, z); entity.set_rotation(yaw, pitch); - *self.awaiting_teleport.lock() = Some((teleport_id.into(), Vector3::new(x, y, z))); - self.client.send_packet(&CSyncPlayerPosition::new( - x, - y, - z, - yaw, - pitch, - 0, - teleport_id.into(), - )); + *self.awaiting_teleport.lock().await = Some((teleport_id.into(), Vector3::new(x, y, z))); + self.client + .send_packet(&CSyncPlayerPosition::new( + x, + y, + z, + yaw, + pitch, + 0, + teleport_id.into(), + )) + .await; } pub fn block_interaction_range(&self) -> f64 { @@ -216,7 +220,7 @@ impl Player { } /// Kicks the Client with a reason depending on the connection state - pub fn kick(&self, reason: TextComponent) { + pub async fn kick<'a>(&self, reason: TextComponent<'a>) { assert!(!self .client .closed @@ -224,6 +228,7 @@ impl Player { self.client .try_send_packet(&CPlayDisconnect::new(&reason)) + .await .unwrap_or_else(|_| self.client.close()); log::info!( "Kicked {} for {}", @@ -233,15 +238,16 @@ impl Player { self.client.close() } - pub fn set_health(&self, health: f32, food: i32, food_saturation: f32) { - self.living_entity.set_health(health); + pub async fn set_health(&self, health: f32, food: i32, food_saturation: f32) { + self.living_entity.set_health(health).await; self.food.store(food, std::sync::atomic::Ordering::Relaxed); self.food_saturation.store(food_saturation); self.client - .send_packet(&CSetHealth::new(health, food.into(), food_saturation)); + .send_packet(&CSetHealth::new(health, food.into(), food_saturation)) + .await; } - pub fn set_gamemode(&self, gamemode: GameMode) { + pub async fn set_gamemode(&self, gamemode: GameMode) { // We could send the same gamemode without problems. But why waste bandwidth ? let current_gamemode = self.gamemode.load(); assert!( @@ -260,34 +266,39 @@ impl Player { uuid: self.gameprofile.id, actions: vec![PlayerAction::UpdateGameMode((gamemode as i32).into())], }], - )); - self.client.send_packet(&CGameEvent::new( - GameEvent::ChangeGameMode, - gamemode.to_f32().unwrap(), - )); + )) + .await; + self.client + .send_packet(&CGameEvent::new( + GameEvent::ChangeGameMode, + gamemode.to_f32().unwrap(), + )) + .await; } - pub fn send_system_message(&self, text: TextComponent) { + pub async fn send_system_message<'a>(&self, text: TextComponent<'a>) { self.client - .send_packet(&CSystemChatMessage::new(text, false)); + .send_packet(&CSystemChatMessage::new(text, false)) + .await; } } impl Player { pub async fn process_packets(&self, server: &Arc) { - let mut packets = self.client.client_packets_queue.lock(); + let mut packets = self.client.client_packets_queue.lock().await; while let Some(mut packet) = packets.pop_back() { match self.handle_play_packet(server, &mut packet).await { Ok(_) => {} Err(e) => { if e.is_kick() { if let Some(kick_reason) = e.client_kick_reason() { - self.kick(TextComponent::text(&kick_reason)) + self.kick(TextComponent::text(&kick_reason)).await } else { self.kick(TextComponent::text(&format!( "Error while reading incoming packet {}", e - ))); + ))) + .await; } } e.log(); @@ -304,7 +315,8 @@ impl Player { let bytebuf = &mut packet.bytebuf; match packet.id.0 { SConfirmTeleport::PACKET_ID => { - self.handle_confirm_teleport(SConfirmTeleport::read(bytebuf)?); + self.handle_confirm_teleport(SConfirmTeleport::read(bytebuf)?) + .await; Ok(()) } SChatCommand::PACKET_ID => { @@ -342,7 +354,8 @@ impl Player { Ok(()) } SClientInformationPlay::PACKET_ID => { - self.handle_client_information_play(SClientInformationPlay::read(bytebuf)?); + self.handle_client_information_play(SClientInformationPlay::read(bytebuf)?) + .await; Ok(()) } SInteract::PACKET_ID => { @@ -364,15 +377,18 @@ impl Player { Ok(()) } SSetHeldItem::PACKET_ID => { - self.handle_set_held_item(SSetHeldItem::read(bytebuf)?); + self.handle_set_held_item(SSetHeldItem::read(bytebuf)?) + .await; Ok(()) } SSetCreativeSlot::PACKET_ID => { - self.handle_set_creative_slot(SSetCreativeSlot::read(bytebuf)?)?; + self.handle_set_creative_slot(SSetCreativeSlot::read(bytebuf)?) + .await?; Ok(()) } SPlayPingRequest::PACKET_ID => { - self.handle_play_ping_request(SPlayPingRequest::read(bytebuf)?); + self.handle_play_ping_request(SPlayPingRequest::read(bytebuf)?) + .await; Ok(()) } SClickContainer::PACKET_ID => { @@ -381,7 +397,8 @@ impl Player { Ok(()) } SCloseContainer::PACKET_ID => { - self.handle_close_container(server, SCloseContainer::read(bytebuf)?); + self.handle_close_container(server, SCloseContainer::read(bytebuf)?) + .await; Ok(()) } SKeepAlive::PACKET_ID => { diff --git a/pumpkin/src/error.rs b/pumpkin/src/error.rs index 85cf4a34f..74fb5c0e7 100644 --- a/pumpkin/src/error.rs +++ b/pumpkin/src/error.rs @@ -3,7 +3,7 @@ use pumpkin_inventory::InventoryError; use pumpkin_protocol::bytebuf::DeserializerError; use std::fmt::Display; -pub trait PumpkinError: std::error::Error + Display { +pub trait PumpkinError: Send + std::error::Error + Display { fn is_kick(&self) -> bool; fn log(&self) { diff --git a/pumpkin/src/main.rs b/pumpkin/src/main.rs index 7b7be6fae..f79ffdbdc 100644 --- a/pumpkin/src/main.rs +++ b/pumpkin/src/main.rs @@ -17,15 +17,13 @@ compile_error!("Compiling for WASI targets is not supported!"); use log::LevelFilter; -use mio::net::TcpListener; -use mio::{Events, Interest, Poll, Token}; -use client::{interrupted, Client}; +use client::Client; use pumpkin_protocol::client::play::CKeepAlive; use pumpkin_protocol::ConnectionState; use server::Server; use std::collections::HashMap; -use std::io::{self, Read}; +use std::io::{self}; use std::time::Duration; // Setup some tokens to allow us to identify which event is for which socket. @@ -82,7 +80,8 @@ const fn convert_logger_filter(level: pumpkin_config::logging::LevelFilter) -> L } } -fn main() -> io::Result<()> { +#[tokio::main] +async fn main() -> io::Result<()> { use std::sync::Arc; use entity::player::Player; @@ -109,206 +108,150 @@ fn main() -> io::Result<()> { .unwrap(); // ensure rayon is built outside of tokio scope rayon::ThreadPoolBuilder::new().build_global().unwrap(); - rt.block_on(async { - let default_panic = std::panic::take_hook(); - std::panic::set_hook(Box::new(move |info| { - default_panic(info); - // TODO: Gracefully exit? - std::process::exit(1); - })); + let default_panic = std::panic::take_hook(); + std::panic::set_hook(Box::new(move |info| { + default_panic(info); + // TODO: Gracefully exit? + std::process::exit(1); + })); - const SERVER: Token = Token(0); - use std::time::Instant; + use std::time::Instant; - let time = Instant::now(); + let time = Instant::now(); - // Create a poll instance. - let mut poll = Poll::new()?; - // Create storage for events. - let mut events = Events::with_capacity(128); + // Setup the TCP server socket. + let addr = BASIC_CONFIG.server_address; + let listener = tokio::net::TcpListener::bind(addr) + .await + .expect("Failed to start TcpListener"); - // Setup the TCP server socket. - let addr = BASIC_CONFIG.server_address; - let mut listener = TcpListener::bind(addr)?; + let use_console = ADVANCED_CONFIG.commands.use_console; + let rcon = ADVANCED_CONFIG.rcon.clone(); - // Register the server with poll we can receive events for it. - poll.registry() - .register(&mut listener, SERVER, Interest::READABLE)?; + let players: Arc>>> = + Arc::new(tokio::sync::Mutex::new(HashMap::new())); - // Unique token for each incoming connection. - let mut unique_id = SERVER.0 + 1; + let server = Arc::new(Server::new()); + log::info!("Started Server took {}ms", time.elapsed().as_millis()); + log::info!("You now can connect to the server, Listening on {}", addr); - let use_console = ADVANCED_CONFIG.commands.use_console; - let rcon = ADVANCED_CONFIG.rcon.clone(); + if use_console { + let server = server.clone(); + tokio::spawn(async move { + let stdin = std::io::stdin(); + loop { + let mut out = String::new(); + stdin + .read_line(&mut out) + .expect("Failed to read console line"); - let mut clients: HashMap> = HashMap::new(); - let mut players: HashMap> = HashMap::new(); + if !out.is_empty() { + let dispatcher = server.command_dispatcher.clone(); + dispatcher.handle_command(&mut commands::CommandSender::Console, &server, &out); + } + } + }); + } + if rcon.enabled { + let server = server.clone(); + tokio::spawn(async move { + RCONServer::new(&rcon, &server).await.unwrap(); + }); + } + let mut unique_id = 0; + loop { + // Asynchronously wait for an inbound socket. + let (connection, address) = listener.accept().await?; - let server = Arc::new(Server::new()); - log::info!("Started Server took {}ms", time.elapsed().as_millis()); - log::info!("You now can connect to the server, Listening on {}", addr); + log::info!( + "Accepted connection from: {}", + scrub_address(&format!("{}", address)) + ); - if use_console { - let server = server.clone(); + if let Err(e) = connection.set_nodelay(true) { + log::warn!("failed to set TCP_NODELAY {e}"); + } + + unique_id += 1; + let id = unique_id; + + let keep_alive = tokio::sync::mpsc::channel(1024); + let client = Arc::new(Client::new(id, connection, addr, keep_alive.0.into())); + + { + let client = client.clone(); + let mut receiver = keep_alive.1; tokio::spawn(async move { - let stdin = std::io::stdin(); + let mut interval = tokio::time::interval(Duration::from_secs(1)); loop { - let mut out = String::new(); - stdin - .read_line(&mut out) - .expect("Failed to read console line"); - - if !out.is_empty() { - let dispatcher = server.command_dispatcher.clone(); - dispatcher.handle_command( - &mut commands::CommandSender::Console, - &server, - &out, - ); - } - } - }); - } - if rcon.enabled { - let server = server.clone(); - tokio::spawn(async move { - RCONServer::new(&rcon, &server).await.unwrap(); - }); - } - loop { - if let Err(err) = poll.poll(&mut events, None) { - if interrupted(&err) { - continue; - } - return Err(err); - } - - for event in events.iter() { - match event.token() { - s if s == SERVER => loop { - // Received an event for the TCP server socket, which - // indicates we can accept an connection. - let (mut connection, address) = match listener.accept() { - Ok((connection, address)) => (connection, address), - Err(e) if e.kind() == io::ErrorKind::WouldBlock => { - // If we get a `WouldBlock` error we know our - // listener has no more incoming connections queued, - // so we can return to polling and wait for some - // more. - break; - } - Err(e) => { - // If it was any other kind of error, something went - // wrong and we terminate with an error. - return Err(e); - } - }; - if let Err(e) = connection.set_nodelay(true) { - log::warn!("failed to set TCP_NODELAY {e}"); - } - - log::info!( - "Accepted connection from: {}", - scrub_address(&format!("{}", address)) - ); - - unique_id += 1; - let id = unique_id; - poll.registry().register( - &mut connection, - Token(id), - Interest::READABLE.add(Interest::WRITABLE), - )?; - let keep_alive = tokio::sync::mpsc::channel(1024); - let client = - Arc::new(Client::new(id, connection, address, keep_alive.0.into())); - + interval.tick().await; + let now = std::time::Instant::now(); + if client.connection_state.load() == ConnectionState::Play { + if now.duration_since(client.last_alive_received.load()) + >= Duration::from_secs(15) { - let client = client.clone(); - let mut receiver = keep_alive.1; - tokio::spawn(async move { - let mut interval = tokio::time::interval(Duration::from_secs(1)); - loop { - interval.tick().await; - let now = std::time::Instant::now(); - if client.connection_state.load() == ConnectionState::Play { - if now.duration_since(client.last_alive_received.load()) - >= Duration::from_secs(15) - { - dbg!("no keep alive"); - client.kick("No keep alive received"); - break; - } - let random = rand::random::(); - client.send_packet(&CKeepAlive { - keep_alive_id: random, - }); - if let Some(id) = receiver.recv().await { - if id == random { - client.last_alive_received.store(now); - } - } - } else { - client.last_alive_received.store(now); - } - } - }); + dbg!("no keep alive"); + client.kick("No keep alive received").await; + break; } - clients.insert(id, client); - }, - // Maybe received an event for a TCP connection. - token => { - // poll Player - if let Some(player) = players.get_mut(&token.0) { - player.client.poll(event).await; - let closed = player - .client - .closed - .load(std::sync::atomic::Ordering::Relaxed); - if !closed { - player.process_packets(&server).await; - } - if closed { - if let Some(player) = players.remove(&token.0) { - player.remove().await; - let connection = &mut player.client.connection.lock(); - poll.registry().deregister(connection.by_ref())?; - } - } - }; - - // Poll current Clients (non players) - let (done, make_player) = if let Some(client) = clients.get_mut(&token.0) { - client.poll(event).await; - let closed = client.closed.load(std::sync::atomic::Ordering::Relaxed); - if !closed { - client.process_packets(&server).await; - } - ( - closed, - client - .make_player - .load(std::sync::atomic::Ordering::Relaxed), - ) - } else { - (false, false) - }; - if done || make_player { - if let Some(client) = clients.remove(&token.0) { - if done { - let connection = &mut client.connection.lock(); - poll.registry().deregister(connection.by_ref())?; - } else if make_player { - let id = client.id; - let (player, world) = server.add_player(id, client).await; - players.insert(id, player.clone()); - world.spawn_player(&BASIC_CONFIG, player).await; - } + let random = rand::random::(); + client + .send_packet(&CKeepAlive { + keep_alive_id: random, + }) + .await; + if let Some(id) = receiver.recv().await { + if id == random { + client.last_alive_received.store(now); } } + } else { + client.last_alive_received.store(now); } } - } + }); } - }) + + let players = players.clone(); + let server = server.clone(); + tokio::spawn(async move { + let mut players = players.lock().await; // Move ownership of players + let server = &server; // Reference to server + // poll Player + if let Some(player) = players.get_mut(&id) { + while !player + .client + .closed + .load(std::sync::atomic::Ordering::Relaxed) + { + dbg!("a"); + player.process_packets(server).await; + player.client.poll().await; + } + if let Some(player) = players.remove(&id) { + dbg!("removed player"); + player.remove().await; + } + } else { + // Poll current Clients (non players) + while !client.closed.load(std::sync::atomic::Ordering::Relaxed) + && !client + .make_player + .load(std::sync::atomic::Ordering::Relaxed) + { + client.process_packets(server).await; + client.poll().await; + } + if client + .make_player + .load(std::sync::atomic::Ordering::Relaxed) + { + let id = client.id; + let (player, world) = server.add_player(id, client).await; + players.insert(id, player.clone()); + world.spawn_player(&BASIC_CONFIG, player).await; + } + } + }); + } } diff --git a/pumpkin/src/proxy/bungeecord.rs b/pumpkin/src/proxy/bungeecord.rs index 76a2e130a..c8c9d1cea 100644 --- a/pumpkin/src/proxy/bungeecord.rs +++ b/pumpkin/src/proxy/bungeecord.rs @@ -20,11 +20,11 @@ pub enum BungeeCordError { FailedMakeOfflineUUID, } -pub fn bungeecord_login( +pub async fn bungeecord_login( client: &Client, username: String, ) -> Result<(IpAddr, GameProfile), BungeeCordError> { - let server_address = client.server_address.lock(); + let server_address = client.server_address.lock().await; let data = server_address.split('\0').take(4).collect::>(); // Ip of player, only given if ip_forward on bungee is true @@ -32,7 +32,7 @@ pub fn bungeecord_login( Some(ip) => ip .parse() .map_err(|_| BungeeCordError::FailedParseAddress)?, - None => client.address.lock().ip(), + None => client.address.lock().await.ip(), }; // Uuid of player, only given if ip_forward on bungee is true diff --git a/pumpkin/src/proxy/velocity.rs b/pumpkin/src/proxy/velocity.rs index f1b4fe494..37099bddc 100644 --- a/pumpkin/src/proxy/velocity.rs +++ b/pumpkin/src/proxy/velocity.rs @@ -44,17 +44,19 @@ pub enum VelocityError { FailedReadProfileProperties, } -pub fn velocity_login(client: &Client) { +pub async fn velocity_login(client: &Client) { // TODO: validate packet transaction id from plugin response with this let velocity_message_id: i32 = rand::thread_rng().gen(); let mut buf = BytesMut::new(); buf.put_u8(MAX_SUPPORTED_FORWARDING_VERSION); - client.send_packet(&CLoginPluginRequest::new( - velocity_message_id.into(), - PLAYER_INFO_CHANNEL, - &buf, - )); + client + .send_packet(&CLoginPluginRequest::new( + velocity_message_id.into(), + PLAYER_INFO_CHANNEL, + &buf, + )) + .await; } pub fn check_integrity(data: (&[u8], &[u8]), secret: &str) -> bool { diff --git a/pumpkin/src/rcon/mod.rs b/pumpkin/src/rcon/mod.rs index 0feff8fa5..429312abb 100644 --- a/pumpkin/src/rcon/mod.rs +++ b/pumpkin/src/rcon/mod.rs @@ -1,16 +1,13 @@ use std::{ collections::HashMap, - io::{self, Read, Write}, + io::{self}, net::SocketAddr, }; -use mio::{ - net::{TcpListener, TcpStream}, - Events, Interest, Poll, Token, -}; use packet::{ClientboundPacket, Packet, PacketError, ServerboundPacket}; use pumpkin_config::{RCONConfig, ADVANCED_CONFIG}; use thiserror::Error; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; use crate::server::Server; @@ -26,87 +23,40 @@ pub enum RCONError { Io(io::Error), } -const SERVER: Token = Token(0); - pub struct RCONServer; impl RCONServer { pub async fn new(config: &RCONConfig, server: &Server) -> Result { assert!(config.enabled, "RCON is not enabled"); - let mut poll = Poll::new().unwrap(); - let mut listener = TcpListener::bind(config.address).unwrap(); - - poll.registry() - .register(&mut listener, SERVER, Interest::READABLE) - .unwrap(); - - let mut unique_id = SERVER.0 + 1; - - let mut events = Events::with_capacity(20); + let listener = tokio::net::TcpListener::bind(config.address).await.unwrap(); let mut connections: HashMap = HashMap::new(); let password = config.password.clone(); + let mut unique_id = 0; loop { - poll.poll(&mut events, None).unwrap(); + // Asynchronously wait for an inbound socket. + let (connection, address) = listener.accept().await?; - for event in events.iter() { - match event.token() { - SERVER => loop { - // Received an event for the TCP server socket, which - // indicates we can accept an connection. - let (mut connection, address) = match listener.accept() { - Ok((connection, address)) => (connection, address), - Err(e) if e.kind() == io::ErrorKind::WouldBlock => { - // If we get a `WouldBlock` error we know our - // listener has no more incoming connections queued, - // so we can return to polling and wait for some - // more. - break; - } - Err(e) => { - // If it was any other kind of error, something went - // wrong and we terminate with an error. - return Err(e); - } - }; - if config.max_connections != 0 - && connections.len() >= config.max_connections as usize - { - break; - } + if config.max_connections != 0 && connections.len() >= config.max_connections as usize { + continue; + } - unique_id += 1; - let token = unique_id; - poll.registry() - .register( - &mut connection, - Token(token), - Interest::READABLE.add(Interest::WRITABLE), - ) - .unwrap(); - connections.insert(token, RCONClient::new(connection, address)); - }, + unique_id += 1; + let token = unique_id; + connections.insert(token, RCONClient::new(connection, address)); - token => { - let done = if let Some(client) = connections.get_mut(&token.0) { - client.handle(server, &password).await - } else { - false - }; - if done { - if let Some(mut client) = connections.remove(&token.0) { - let config = &ADVANCED_CONFIG.rcon; - if config.logging.log_quit { - log::info!( - "RCON ({}): Client closed connection", - client.address - ); - } - poll.registry().deregister(&mut client.connection)?; - } - } + let done = if let Some(client) = connections.get_mut(&token) { + client.handle(server, &password).await + } else { + false + }; + if done { + if let Some(client) = connections.remove(&token) { + let config = &ADVANCED_CONFIG.rcon; + if config.logging.log_quit { + log::info!("RCON ({}): Client closed connection", client.address); } } } @@ -115,7 +65,7 @@ impl RCONServer { } pub struct RCONClient { - connection: TcpStream, + connection: tokio::net::TcpStream, address: SocketAddr, logged_in: bool, incoming: Vec, @@ -123,7 +73,7 @@ pub struct RCONClient { } impl RCONClient { - pub const fn new(connection: TcpStream, address: SocketAddr) -> Self { + pub const fn new(connection: tokio::net::TcpStream, address: SocketAddr) -> Self { Self { connection, address, @@ -136,7 +86,7 @@ impl RCONClient { pub async fn handle(&mut self, server: &Server, password: &str) -> bool { if !self.closed { loop { - match self.read_bytes() { + match self.read_bytes().await { // Stream closed, so we can't reply, so we just close everything. Ok(true) => return true, Ok(false) => {} @@ -205,9 +155,9 @@ impl RCONClient { } } - fn read_bytes(&mut self) -> io::Result { + async fn read_bytes(&mut self) -> io::Result { let mut buf = [0; 1460]; - let n = self.connection.read(&mut buf)?; + let n = self.connection.read(&mut buf).await?; if n == 0 { return Ok(true); } @@ -224,6 +174,7 @@ impl RCONClient { let buf = packet.write_buf(id, body); self.connection .write(&buf) + .await .map_err(PacketError::FailedSend)?; Ok(()) } diff --git a/pumpkin/src/server/mod.rs b/pumpkin/src/server/mod.rs index 6deee69e2..87d90aaeb 100644 --- a/pumpkin/src/server/mod.rs +++ b/pumpkin/src/server/mod.rs @@ -1,6 +1,5 @@ use connection_cache::{CachedBranding, CachedStatus}; use key_store::KeyStore; -use parking_lot::{Mutex, RwLock}; use pumpkin_config::BASIC_CONFIG; use pumpkin_core::GameMode; use pumpkin_entity::EntityId; @@ -19,6 +18,8 @@ use std::{ }, time::Duration, }; +use tokio::sync::Mutex; +use tokio::sync::RwLock; use crate::client::EncryptionError; use crate::{ @@ -100,17 +101,17 @@ impl Server { // TODO: select default from config let world = &self.worlds[0]; - let player = Arc::new(Player::new(client, world.clone(), entity_id, gamemode)); - world.add_player(id, player.clone()); + let player = Arc::new(Player::new(client, world.clone(), entity_id, gamemode).await); + world.add_player(id, player.clone()).await; (player, world.clone()) } - pub fn try_get_container( + pub async fn try_get_container( &self, player_id: EntityId, container_id: u64, ) -> Option>>> { - let open_containers = self.open_containers.read(); + let open_containers = self.open_containers.read().await; open_containers .get(&container_id)? .try_open(player_id) @@ -118,19 +119,19 @@ impl Server { } /// Sends a Packet to all Players in all worlds - pub fn broadcast_packet_all

(&self, packet: &P) + pub async fn broadcast_packet_all

(&self, packet: &P) where P: ClientPacket, { for world in &self.worlds { - world.broadcast_packet_all(packet) + world.broadcast_packet_all(packet).await } } /// Searches every world for a player by name - pub fn get_player_by_name(&self, name: &str) -> Option> { + pub async fn get_player_by_name(&self, name: &str) -> Option> { for world in self.worlds.iter() { - if let Some(player) = world.get_player_by_name(name) { + if let Some(player) = world.get_player_by_name(name).await { return Some(player); } } diff --git a/pumpkin/src/world/mod.rs b/pumpkin/src/world/mod.rs index 394aad40c..5bf7e7062 100644 --- a/pumpkin/src/world/mod.rs +++ b/pumpkin/src/world/mod.rs @@ -7,7 +7,6 @@ use crate::{ entity::{player::Player, Entity}, }; use num_traits::ToPrimitive; -use parking_lot::Mutex; use pumpkin_config::BasicConfiguration; use pumpkin_core::math::vector2::Vector2; use pumpkin_entity::{entity_type::EntityType, EntityId}; @@ -20,6 +19,7 @@ use pumpkin_protocol::{ }; use pumpkin_world::level::Level; use tokio::sync::mpsc; +use tokio::sync::Mutex; /// Represents a Minecraft world, containing entities, players, and the underlying level data. /// @@ -51,13 +51,13 @@ impl World { /// Sends the specified packet to every player currently logged in to the server. /// /// **Note:** This function acquires a lock on the `current_players` map, ensuring thread safety. - pub fn broadcast_packet_all

(&self, packet: &P) + pub async fn broadcast_packet_all

(&self, packet: &P) where P: ClientPacket, { - let current_players = self.current_players.lock(); + let current_players = self.current_players.lock().await; for player in current_players.values() { - player.client.send_packet(packet); + player.client.send_packet(packet).await; } } @@ -66,13 +66,13 @@ impl World { /// Sends the specified packet to every player currently logged in to the server, excluding the players listed in the `except` parameter. /// /// **Note:** This function acquires a lock on the `current_players` map, ensuring thread safety. - pub fn broadcast_packet_expect

(&self, except: &[usize], packet: &P) + pub async fn broadcast_packet_expect

(&self, except: &[usize], packet: &P) where P: ClientPacket, { - let current_players = self.current_players.lock(); + let current_players = self.current_players.lock().await; for (_, player) in current_players.iter().filter(|c| !except.contains(c.0)) { - player.client.send_packet(packet); + player.client.send_packet(packet).await; } } @@ -83,33 +83,37 @@ impl World { log::debug!("spawning player, entity id {}", entity_id); // login packet for our new player - player.client.send_packet(&CLogin::new( - entity_id, - base_config.hardcore, - &["minecraft:overworld"], - base_config.max_players.into(), - base_config.view_distance.into(), // TODO: view distance - base_config.simulation_distance.into(), // TODO: sim view dinstance - false, - false, - false, - 0.into(), - "minecraft:overworld", - 0, // seed - gamemode.to_u8().unwrap(), - base_config.default_gamemode.to_i8().unwrap(), - false, - false, - None, - 0.into(), - false, - )); + player + .client + .send_packet(&CLogin::new( + entity_id, + base_config.hardcore, + &["minecraft:overworld"], + base_config.max_players.into(), + base_config.view_distance.into(), // TODO: view distance + base_config.simulation_distance.into(), // TODO: sim view dinstance + false, + false, + false, + 0.into(), + "minecraft:overworld", + 0, // seed + gamemode.to_u8().unwrap(), + base_config.default_gamemode.to_i8().unwrap(), + false, + false, + None, + 0.into(), + false, + )) + .await; dbg!("sending abilities"); // player abilities // TODO: this is for debug purpose, remove later player .client - .send_packet(&CPlayerAbilities::new(0x02, 0.4, 0.1)); + .send_packet(&CPlayerAbilities::new(0x02, 0.4, 0.1)) + .await; // teleport let x = 10.0; @@ -117,7 +121,7 @@ impl World { let z = 10.0; let yaw = 10.0; let pitch = 10.0; - player.teleport(x, y, z, 10.0, 10.0); + player.teleport(x, y, z, 10.0, 10.0).await; let gameprofile = &player.gameprofile; // first send info update to our new player, So he can see his Skin // also send his info to everyone else @@ -133,12 +137,13 @@ impl World { PlayerAction::UpdateListed(true), ], }], - )); + )) + .await; // here we send all the infos of already joined players let mut entries = Vec::new(); { - let current_players = self.current_players.lock(); + let current_players = self.current_players.lock().await; for (_, playerr) in current_players .iter() .filter(|(c, _)| **c != player.client.id) @@ -157,7 +162,8 @@ impl World { } player .client - .send_packet(&CPlayerInfoUpdate::new(0x01 | 0x08, &entries)); + .send_packet(&CPlayerInfoUpdate::new(0x01 | 0x08, &entries)) + .await; } let gameprofile = &player.gameprofile; @@ -181,43 +187,54 @@ impl World { 0.0, 0.0, ), - ); + ) + .await; // spawn players for our client let token = player.client.id; - for (_, existing_player) in self.current_players.lock().iter().filter(|c| c.0 != &token) { + for (_, existing_player) in self + .current_players + .lock() + .await + .iter() + .filter(|c| c.0 != &token) + { let entity = &existing_player.living_entity.entity; let pos = entity.pos.load(); let gameprofile = &existing_player.gameprofile; - player.client.send_packet(&CSpawnEntity::new( - existing_player.entity_id().into(), - gameprofile.id, - (EntityType::Player as i32).into(), - pos.x, - pos.y, - pos.z, - entity.yaw.load(), - entity.pitch.load(), - entity.head_yaw.load(), - 0.into(), - 0.0, - 0.0, - 0.0, - )) + player + .client + .send_packet(&CSpawnEntity::new( + existing_player.entity_id().into(), + gameprofile.id, + (EntityType::Player as i32).into(), + pos.x, + pos.y, + pos.z, + entity.yaw.load(), + entity.pitch.load(), + entity.head_yaw.load(), + 0.into(), + 0.0, + 0.0, + 0.0, + )) + .await } // entity meta data // set skin parts - if let Some(config) = player.client.config.lock().as_ref() { + if let Some(config) = player.client.config.lock().await.as_ref() { let packet = CSetEntityMetadata::new( entity_id.into(), Metadata::new(17, VarInt(0), config.skin_parts), ); - self.broadcast_packet_all(&packet) + self.broadcast_packet_all(&packet).await } // Start waiting for level chunks, Sets the "Loading Terrain" screen player .client - .send_packet(&CGameEvent::new(GameEvent::StartWaitingChunks, 0.0)); + .send_packet(&CGameEvent::new(GameEvent::StartWaitingChunks, 0.0)) + .await; // Spawn in initial chunks player_chunker::player_join(self, player.clone()).await; @@ -230,7 +247,10 @@ impl World { let level = self.level.clone(); let closed = client.closed.load(std::sync::atomic::Ordering::Relaxed); let chunks = Arc::new(chunks); - tokio::task::spawn_blocking(move || level.lock().fetch_chunks(&chunks, sender, closed)); + tokio::spawn(async move { + let level = level.lock().await; + level.fetch_chunks(&chunks, sender, closed) + }); while let Some(chunk_data) = chunk_receiver.recv().await { // dbg!(chunk_pos); @@ -248,15 +268,15 @@ impl World { ); } if !client.closed.load(std::sync::atomic::Ordering::Relaxed) { - client.send_packet(&CChunkData(&chunk_data)); + client.send_packet(&CChunkData(&chunk_data)).await; } } dbg!("DONE CHUNKS", inst.elapsed()); } /// Gets a Player by entity id - pub fn get_player_by_entityid(&self, id: EntityId) -> Option> { - for player in self.current_players.lock().values() { + pub async fn get_player_by_entityid(&self, id: EntityId) -> Option> { + for player in self.current_players.lock().await.values() { if player.entity_id() == id { return Some(player.clone()); } @@ -265,8 +285,8 @@ impl World { } /// Gets a Player by name - pub fn get_player_by_name(&self, name: &str) -> Option> { - for player in self.current_players.lock().values() { + pub async fn get_player_by_name(&self, name: &str) -> Option> { + for player in self.current_players.lock().await.values() { if player.gameprofile.name == name { return Some(player.clone()); } @@ -274,24 +294,27 @@ impl World { None } - pub fn add_player(&self, id: usize, player: Arc) { - self.current_players.lock().insert(id, player); + pub async fn add_player(&self, id: usize, player: Arc) { + self.current_players.lock().await.insert(id, player); } - pub fn remove_player(&self, player: &Player) { + pub async fn remove_player(&self, player: &Player) { self.current_players .lock() + .await .remove(&player.client.id) .unwrap(); let uuid = player.gameprofile.id; self.broadcast_packet_expect( &[player.client.id], &CRemovePlayerInfo::new(1.into(), &[uuid]), - ); - self.remove_entity(&player.living_entity.entity); + ) + .await; + self.remove_entity(&player.living_entity.entity).await; } - pub fn remove_entity(&self, entity: &Entity) { + pub async fn remove_entity(&self, entity: &Entity) { self.broadcast_packet_all(&CRemoveEntities::new(&[entity.entity_id.into()])) + .await } } diff --git a/pumpkin/src/world/player_chunker.rs b/pumpkin/src/world/player_chunker.rs index 8cb5ea257..7d9fe1f4a 100644 --- a/pumpkin/src/world/player_chunker.rs +++ b/pumpkin/src/world/player_chunker.rs @@ -4,17 +4,18 @@ use pumpkin_config::BASIC_CONFIG; use pumpkin_core::math::{ get_section_cord, position::WorldPosition, vector2::Vector2, vector3::Vector3, }; -use pumpkin_protocol::client::play::{CCenterChunk, CUnloadChunk}; +use pumpkin_protocol::client::play::CCenterChunk; use pumpkin_world::cylindrical_chunk_iterator::Cylindrical; use crate::entity::{player::Player, Entity}; use super::World; -fn get_view_distance(player: &Player) -> i8 { +async fn get_view_distance(player: &Player) -> i8 { player .config .lock() + .await .view_distance .clamp(2, BASIC_CONFIG.view_distance as i8) } @@ -24,11 +25,14 @@ pub async fn player_join(world: &World, player: Arc) { player.watched_section.store(new_watched); let watched_section = new_watched; let chunk_pos = player.living_entity.entity.chunk_pos.load(); - player.client.send_packet(&CCenterChunk { - chunk_x: chunk_pos.x.into(), - chunk_z: chunk_pos.z.into(), - }); - let view_distance = get_view_distance(&player) as i32; + player + .client + .send_packet(&CCenterChunk { + chunk_x: chunk_pos.x.into(), + chunk_z: chunk_pos.z.into(), + }) + .await; + let view_distance = get_view_distance(&player).await as i32; dbg!(view_distance); let old_cylindrical = Cylindrical::new( Vector2::new(watched_section.x, watched_section.z), @@ -43,9 +47,9 @@ pub async fn player_join(world: &World, player: Arc) { loading_chunks.push(chunk_pos); }, |chunk_pos| { - player - .client - .send_packet(&CUnloadChunk::new(chunk_pos.x, chunk_pos.z)); + // player + // .client + // .send_packet(&CUnloadChunk::new(chunk_pos.x, chunk_pos.z)); }, true, ); @@ -61,12 +65,15 @@ pub async fn update_position(entity: &Entity, player: &Player) { let new_watched = chunk_section_from_pos(&entity.block_pos.load()); if current_watched != new_watched { let chunk_pos = entity.chunk_pos.load(); - player.client.send_packet(&CCenterChunk { - chunk_x: chunk_pos.x.into(), - chunk_z: chunk_pos.z.into(), - }); + player + .client + .send_packet(&CCenterChunk { + chunk_x: chunk_pos.x.into(), + chunk_z: chunk_pos.z.into(), + }) + .await; - let view_distance = get_view_distance(player) as i32; + let view_distance = get_view_distance(player).await as i32; let old_cylindrical = Cylindrical::new( Vector2::new(current_watched.x, current_watched.z), view_distance, @@ -82,9 +89,9 @@ pub async fn update_position(entity: &Entity, player: &Player) { loading_chunks.push(chunk_pos); }, |chunk_pos| { - player - .client - .send_packet(&CUnloadChunk::new(chunk_pos.x, chunk_pos.z)); + // player + // .client + // .send_packet(&CUnloadChunk::new(chunk_pos.x, chunk_pos.z)); }, false, );