Pumpkin-MC/Pumpkin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add tokio

Browse Source
This commit is contained in:
Snowiiii 2024-10-17 18:59:51 +02:00
parent c12fc46e69
commit f74f77a1f0
25 changed files with 812 additions and 806 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pumpkin-inventory/Cargo.toml
View File

@ -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

18
pumpkin-inventory/src/drag_handler.rs
View File

@ -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<HashMap<u64, Arc<Mutex<Drag>>>>);
@ -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<T: Container>(
pub async fn apply_drag<T: Container>(
&self,
maybe_carried_item: &mut Option<ItemStack>,
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)?

2
pumpkin-inventory/src/open_container.rs
View File

@ -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<i32>,

4
pumpkin-protocol/src/lib.rs
View File

@ -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")]

4
pumpkin/Cargo.toml
View File

@ -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

105
pumpkin/src/client/client_packet.rs
View File

@ -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<GameProfile, AuthError> {
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(
&registry.registry_id,
&registry.registry_entries,
));
))
.await;
}
// We are done with configuring
dbg!("finish config");
self.send_packet(&CFinishConfig::new());
self.send_packet(&CFinishConfig::new()).await;
}
pub async fn handle_config_acknowledged(&self, _config_acknowledged: SAcknowledgeFinishConfig) {

185
pumpkin/src/client/container.rs
View File

@ -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<dyn Container>>) {
let mut inventory = self.inventory.lock();
pub async fn set_container_content(&self, container: Option<&mut Box<dyn Container>>) {
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<T: WindowPropertyTrait>(
pub async fn set_container_property<T: WindowPropertyTrait>(
&mut self,
window_property: WindowProperty<T>,
) {
let (id, value) = window_property.into_tuple();
self.client.send_packet(&CSetContainerProperty::new(
self.inventory.lock().total_opened_containers,
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) => match click.slot {
container_click::Slot::Normal(slot) => {
self.number_button_pressed(opened_container.as_deref_mut(), key_click, slot)
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),
}
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<dyn Container>>,
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<dyn Container>>,
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<dyn Container>>,
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<dyn Container>>,
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<dyn Container>>,
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<dyn Container>>,
@ -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<Arc<Self>> {
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<Arc<Mutex<Box<dyn Container>>>> {
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<Arc<tokio::sync::Mutex<Box<dyn Container>>>> {
// self.open_container .load().map_or_else(|| None, |id| server.try_get_container(self.entity_id(), id).await)
None
}
}

197
pumpkin/src/client/mod.rs
View File

@ -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<Mutex<TcpStream>>,
pub connection_reader: Arc<Mutex<tokio::net::tcp::OwnedReadHalf>>,
pub connection_writer: Arc<Mutex<tokio::net::tcp::OwnedWriteHalf>>,
/// The client's IP address.
pub address: Mutex<SocketAddr>,
/// 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<tokio::sync::mpsc::Sender<i64>>,
) -> 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<CompressionInfo>) {
self.dec.lock().set_compression(compression.is_some());
self.enc.lock().set_compression(compression);
pub async fn set_compression(&self, compression: Option<CompressionInfo>) {
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<P: ClientPacket>(&self, packet: &P) {
pub async fn send_packet<P: ClientPacket>(&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<P: ClientPacket>(&self, packet: &P) -> Result<(), PacketError> {
pub async fn try_send_packet<P: ClientPacket>(&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<Server>) {
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<Server>,
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];
pub async fn poll(&self) {
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.
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();
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(),
}
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
}

143
pumpkin/src/client/player_packet.rs
View File

@ -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,7 +105,8 @@ impl Player {
// return;
// }
// send new position to all other players
world.broadcast_packet_expect(
world
.broadcast_packet_expect(
&[self.client.id],
&CUpdateEntityPos::new(
entity_id.into(),
@ -113,7 +115,8 @@ impl Player {
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,7 +173,8 @@ impl Player {
// }
// send new position to all other players
world.broadcast_packet_expect(
world
.broadcast_packet_expect(
&[self.client.id],
&CUpdateEntityPosRot::new(
entity_id.into(),
@ -181,17 +185,20 @@ impl Player {
pitch as u8,
position_rotation.ground,
),
);
world.broadcast_packet_expect(
)
.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(
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,7 +331,8 @@ impl Player {
let entity = &self.living_entity.entity;
let world = &entity.world;
world.broadcast_packet_all(&CPlayerChatMessage::new(
world
.broadcast_packet_all(&CPlayerChatMessage::new(
gameprofile.id,
1.into(),
chat_message.signature.as_deref(),
@ -331,6 +346,7 @@ impl Player {
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(
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(
world
.broadcast_packet_all(&CBlockUpdate::new(
&location,
block_state_id.get_id_mojang_repr().into(),
));
world.broadcast_packet_all(&CBlockUpdate::new(
))
.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;
};
}

30
pumpkin/src/commands/cmd_echest.rs
View File

@ -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(())

25
pumpkin/src/commands/cmd_help.rs
View File

@ -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(())

10
pumpkin/src/commands/cmd_pumpkin.rs
View File

@ -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(())
})

8
pumpkin/src/commands/dispatcher.rs
View File

@ -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),
// )
}
}

4
pumpkin/src/commands/mod.rs
View File

@ -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()),
}
}

15
pumpkin/src/entity/living.rs
View File

@ -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;
}
}

8
pumpkin/src/entity/mod.rs
View File

@ -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
}
}

79
pumpkin/src/entity/player.rs
View File

@ -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<Client>,
world: Arc<World>,
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(
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,8 +185,9 @@ 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(
*self.awaiting_teleport.lock().await = Some((teleport_id.into(), Vector3::new(x, y, z)));
self.client
.send_packet(&CSyncPlayerPosition::new(
x,
y,
z,
@ -192,7 +195,8 @@ impl Player {
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(
))
.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<Server>) {
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 => {

2
pumpkin/src/error.rs
View File

@ -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) {

163
pumpkin/src/main.rs
View File

@ -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,7 +108,6 @@ 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);
@ -117,32 +115,21 @@ fn main() -> io::Result<()> {
std::process::exit(1);
}));
const SERVER: Token = Token(0);
use std::time::Instant;
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 mut listener = TcpListener::bind(addr)?;
// Register the server with poll we can receive events for it.
poll.registry()
.register(&mut listener, SERVER, Interest::READABLE)?;
// Unique token for each incoming connection.
let mut unique_id = SERVER.0 + 1;
let listener = tokio::net::TcpListener::bind(addr)
.await
.expect("Failed to start TcpListener");
let use_console = ADVANCED_CONFIG.commands.use_console;
let rcon = ADVANCED_CONFIG.rcon.clone();
let mut clients: HashMap<usize, Arc<Client>> = HashMap::new();
let mut players: HashMap<usize, Arc<Player>> = HashMap::new();
let players: Arc<tokio::sync::Mutex<HashMap<usize, Arc<Player>>>> =
Arc::new(tokio::sync::Mutex::new(HashMap::new()));
let server = Arc::new(Server::new());
log::info!("Started Server took {}ms", time.elapsed().as_millis());
@ -160,11 +147,7 @@ fn main() -> io::Result<()> {
if !out.is_empty() {
let dispatcher = server.command_dispatcher.clone();
dispatcher.handle_command(
&mut commands::CommandSender::Console,
&server,
&out,
);
dispatcher.handle_command(&mut commands::CommandSender::Console, &server, &out);
}
}
});
@ -175,53 +158,25 @@ fn main() -> io::Result<()> {
RCONServer::new(&rcon, &server).await.unwrap();
});
}
let mut unique_id = 0;
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}");
}
// Asynchronously wait for an inbound socket.
let (connection, address) = listener.accept().await?;
log::info!(
"Accepted connection from: {}",
scrub_address(&format!("{}", address))
);
if let Err(e) = connection.set_nodelay(true) {
log::warn!("failed to set TCP_NODELAY {e}");
}
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()));
let client = Arc::new(Client::new(id, connection, addr, keep_alive.0.into()));
{
let client = client.clone();
@ -236,13 +191,15 @@ fn main() -> io::Result<()> {
>= Duration::from_secs(15)
{
dbg!("no keep alive");
client.kick("No keep alive received");
client.kick("No keep alive received").await;
break;
}
let random = rand::random::<i64>();
client.send_packet(&CKeepAlive {
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);
@ -254,61 +211,47 @@ fn main() -> io::Result<()> {
}
});
}
clients.insert(id, client);
},
// Maybe received an event for a TCP connection.
token => {
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(&token.0) {
player.client.poll(event).await;
let closed = player
if let Some(player) = players.get_mut(&id) {
while !player
.client
.closed
.load(std::sync::atomic::Ordering::Relaxed);
if !closed {
player.process_packets(&server).await;
.load(std::sync::atomic::Ordering::Relaxed)
{
dbg!("a");
player.process_packets(server).await;
player.client.poll().await;
}
if closed {
if let Some(player) = players.remove(&token.0) {
if let Some(player) = players.remove(&id) {
dbg!("removed player");
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 {
// 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;
}
}
});
}
}
}
}
}
})
}

6
pumpkin/src/proxy/bungeecord.rs
View File

@ -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::<Vec<_>>();
// 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

8
pumpkin/src/proxy/velocity.rs
View File

@ -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(
client
.send_packet(&CLoginPluginRequest::new(
velocity_message_id.into(),
PLAYER_INFO_CHANNEL,
&buf,
));
))
.await;
}
pub fn check_integrity(data: (&[u8], &[u8]), secret: &str) -> bool {

83
pumpkin/src/rcon/mod.rs
View File

@ -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<Self, io::Error> {
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<usize, RCONClient> = 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));
},
token => {
let done = if let Some(client) = connections.get_mut(&token.0) {
let done = if let Some(client) = connections.get_mut(&token) {
client.handle(server, &password).await
} else {
false
};
if done {
if let Some(mut client) = connections.remove(&token.0) {
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
);
}
poll.registry().deregister(&mut client.connection)?;
}
}
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<u8>,
@ -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<bool> {
async fn read_bytes(&mut self) -> io::Result<bool> {
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(())
}

19
pumpkin/src/server/mod.rs
View File

@ -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<Arc<Mutex<Box<dyn Container>>>> {
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<P>(&self, packet: &P)
pub async fn broadcast_packet_all<P>(&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<Arc<Player>> {
pub async fn get_player_by_name(&self, name: &str) -> Option<Arc<Player>> {
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);
}
}

87
pumpkin/src/world/mod.rs
View File

@ -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<P>(&self, packet: &P)
pub async fn broadcast_packet_all<P>(&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<P>(&self, except: &[usize], packet: &P)
pub async fn broadcast_packet_expect<P>(&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,7 +83,9 @@ impl World {
log::debug!("spawning player, entity id {}", entity_id);
// login packet for our new player
player.client.send_packet(&CLogin::new(
player
.client
.send_packet(&CLogin::new(
entity_id,
base_config.hardcore,
&["minecraft:overworld"],
@ -103,13 +105,15 @@ impl World {
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,14 +187,23 @@ 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(
player
.client
.send_packet(&CSpawnEntity::new(
existing_player.entity_id().into(),
gameprofile.id,
(EntityType::Player as i32).into(),
@ -203,21 +218,23 @@ impl World {
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<Arc<Player>> {
for player in self.current_players.lock().values() {
pub async fn get_player_by_entityid(&self, id: EntityId) -> Option<Arc<Player>> {
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<Arc<Player>> {
for player in self.current_players.lock().values() {
pub async fn get_player_by_name(&self, name: &str) -> Option<Arc<Player>> {
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<Player>) {
self.current_players.lock().insert(id, player);
pub async fn add_player(&self, id: usize, player: Arc<Player>) {
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
}
}

35
pumpkin/src/world/player_chunker.rs
View File

@ -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>) {
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 {
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;
})
.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<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));
},
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 {
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,
);