diff --git a/.codecov.yml b/.codecov.yml index a02264cc0..3d915b19a 100644 --- a/.codecov.yml +++ b/.codecov.yml @@ -1,6 +1,6 @@ codecov: notify: - require_ci_to_pass: no + require_ci_to_pass: yes ignore: - "blocks" \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 90d58b51c..3486f5a1c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -399,7 +399,7 @@ dependencies = [ name = "feather_client_core" version = "0.1.0" dependencies = [ - "feather_core 0.1.0", + "feather_core 0.3.0", ] [[package]] @@ -421,7 +421,7 @@ dependencies = [ [[package]] name = "feather_core" -version = "0.1.0" +version = "0.3.0" dependencies = [ "byteorder 1.3.2 (registry+https://github.com/rust-lang/crates.io-index)", "bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)", @@ -444,27 +444,29 @@ dependencies = [ [[package]] name = "feather_server" -version = "0.1.0" +version = "0.3.0" dependencies = [ "bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)", "crossbeam 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)", "derive-new 0.5.6 (registry+https://github.com/rust-lang/crates.io-index)", "feather_blocks 0.1.0", - "feather_core 0.1.0", + "feather_core 0.3.0", "hashbrown 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.4.7 (registry+https://github.com/rust-lang/crates.io-index)", "mio 0.6.19 (registry+https://github.com/rust-lang/crates.io-index)", - "mio-extras 2.0.5 (registry+https://github.com/rust-lang/crates.io-index)", + "mio-extras 2.0.5 (git+https://github.com/caelunshun/mio-extras?rev=e5dca5d)", "mockers 0.21.0 (registry+https://github.com/rust-lang/crates.io-index)", "mockers_derive 0.21.0 (registry+https://github.com/rust-lang/crates.io-index)", "mojang-api 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)", "multimap 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)", "openssl 0.10.23 (registry+https://github.com/rust-lang/crates.io-index)", "rand 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)", + "rayon 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)", "rnbt 0.1.0 (git+https://github.com/caelunshun/rnbt)", "serde 1.0.94 (registry+https://github.com/rust-lang/crates.io-index)", "serde_derive 1.0.94 (registry+https://github.com/rust-lang/crates.io-index)", "serde_json 1.0.40 (registry+https://github.com/rust-lang/crates.io-index)", + "shrev 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", "simple_logger 1.3.0 (registry+https://github.com/rust-lang/crates.io-index)", "specs 0.15.0 (registry+https://github.com/rust-lang/crates.io-index)", "toml 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)", @@ -862,8 +864,9 @@ dependencies = [ [[package]] name = "mio-extras" version = "2.0.5" -source = "registry+https://github.com/rust-lang/crates.io-index" +source = "git+https://github.com/caelunshun/mio-extras?rev=e5dca5d#e5dca5d385b6e111810330a315b1fb2ae9bff244" dependencies = [ + "crossbeam 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)", "lazycell 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.4.7 (registry+https://github.com/rust-lang/crates.io-index)", "mio 0.6.19 (registry+https://github.com/rust-lang/crates.io-index)", @@ -2076,7 +2079,7 @@ dependencies = [ "checksum miniz_oxide 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "b6c3756d66cf286314d5f7ebe74886188a9a92f5eee68b06f31ac2b4f314c99d" "checksum miniz_oxide_c_api 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "5b78ca5446dd9fe0dab00e058731b6b08a8c1d2b9cdb8efb10876e24e9ae2494" "checksum mio 0.6.19 (registry+https://github.com/rust-lang/crates.io-index)" = "83f51996a3ed004ef184e16818edc51fadffe8e7ca68be67f9dee67d84d0ff23" -"checksum mio-extras 2.0.5 (registry+https://github.com/rust-lang/crates.io-index)" = "46e73a04c2fa6250b8d802134d56d554a9ec2922bf977777c805ea5def61ce40" +"checksum mio-extras 2.0.5 (git+https://github.com/caelunshun/mio-extras?rev=e5dca5d)" = "" "checksum miow 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "8c1f2f3b1cf331de6896aabf6e9d55dca90356cc9960cca7eaaf408a355ae919" "checksum mockers 0.21.0 (registry+https://github.com/rust-lang/crates.io-index)" = "8bf36d14570661eebd299640fff1fae157c6d7da34bcbf4e4eba1c3b47b46ec9" "checksum mockers_derive 0.21.0 (registry+https://github.com/rust-lang/crates.io-index)" = "a53c3de46bf8e9cbe2b80e5086bbc8659bdf508c3719d50c2f366c9368d5726c" diff --git a/core/Cargo.toml b/core/Cargo.toml index 41be8d8fb..a0461074a 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "feather_core" -version = "0.1.0" +version = "0.3.0" authors = ["caelunshun "] edition = "2018" diff --git a/core/src/network/mctypes.rs b/core/src/network/mctypes.rs index 707d4d7a1..bdebc290c 100644 --- a/core/src/network/mctypes.rs +++ b/core/src/network/mctypes.rs @@ -1,4 +1,4 @@ -use crate::bytebuf::{BufMutAlloc, ByteBuf}; +use crate::bytebuf::{BufMutAlloc, BufResulted, ByteBuf}; use crate::prelude::*; use crate::world::BlockPosition; use bytes::Buf; @@ -110,7 +110,7 @@ impl McTypeRead for T { if self.remaining() == 0 { return Err(()); } - let read = self.get_u8(); + let read = self.read_u8()?; let value = read & 0b01111111; result |= (value as i32) << (7 * num_read); @@ -149,7 +149,7 @@ impl McTypeRead for T { if self.remaining() < 8 { return Err(()); } - let val = self.get_u64_be(); + let val = self.read_i64_be()?; let x = val >> 38; let y = (val >> 26) & 0xFFF; let z = val << 38 >> 38; @@ -158,7 +158,7 @@ impl McTypeRead for T { } fn read_bool(&mut self) -> Result { - let byte = self.get_u8(); + let byte = self.read_u8()?; match byte { 0 => Ok(false), 1 => Ok(true), diff --git a/core/src/network/packet/implementation.rs b/core/src/network/packet/implementation.rs index 4c96cfa25..417d4c1fe 100644 --- a/core/src/network/packet/implementation.rs +++ b/core/src/network/packet/implementation.rs @@ -553,9 +553,7 @@ pub struct DisconnectLogin { #[derive(Default, AsAny, new, Clone)] pub struct EncryptionRequest { pub server_id: String, - pub public_key_len: VarInt, pub public_key: Vec, - pub verify_token_len: VarInt, pub verify_token: Vec, } @@ -567,10 +565,10 @@ impl Packet for EncryptionRequest { fn write_to(&self, mut buf: &mut ByteBuf) { buf.write_string(self.server_id.as_str()); - buf.write_var_int(self.public_key_len); + buf.write_var_int(self.public_key.len() as i32); buf.write(&self.public_key); - buf.write_var_int(self.verify_token_len); + buf.write_var_int(self.verify_token.len() as i32); buf.write(&self.verify_token); } diff --git a/core/src/network/packet/mod.rs b/core/src/network/packet/mod.rs index db54bcd2b..fae482fad 100644 --- a/core/src/network/packet/mod.rs +++ b/core/src/network/packet/mod.rs @@ -41,61 +41,61 @@ pub enum PacketType { // Serverbound // Handshake - Handshake, + Handshake = 1, // Login - LoginStart, - EncryptionResponse, - LoginPluginResponse, + LoginStart = 2, + EncryptionResponse = 3, + LoginPluginResponse = 4, // Play - TeleportConfirm, - QueryBlockNBT, - ChatMessageServerbound, - ClientStatus, - ClientSettings, - TabCompleteServerbound, - ConfirmTransactionServerbound, - EnchantItem, - ClickWindow, - CloseWindowServerbound, - PluginMessageServerbound, - EditBook, - QueryEntityNBT, - UseEntity, - KeepAliveServerbound, - Player, - PlayerPosition, - PlayerPositionAndLookServerbound, - PlayerLook, - VehicleMoveServerbound, - SteerBoat, - PickItem, - CraftRecipeRequest, - PlayerAbilitiesServerbound, - PlayerDigging, - EntityAction, - SteerVehicle, - RecipeBookData, - NameItem, - ResourcePackStatus, - AdvancementTab, - SelectTrade, - SetBeaconEffect, - HeldItemChangeServerbound, - UpdateCommandBlock, - UpdateCommandBlockMinecart, - CreativeInventoryAction, - UpdateStructureBlock, - UpdateSign, - AnimationServerbound, - Spectate, - PlayerBlockPlacement, - UseItem, + TeleportConfirm = 5, + QueryBlockNBT = 6, + ChatMessageServerbound = 7, + ClientStatus = 8, + ClientSettings = 9, + TabCompleteServerbound = 10, + ConfirmTransactionServerbound = 11, + EnchantItem = 12, + ClickWindow = 13, + CloseWindowServerbound = 14, + PluginMessageServerbound = 15, + EditBook = 16, + QueryEntityNBT = 17, + UseEntity = 18, + KeepAliveServerbound = 19, + Player = 20, + PlayerPosition = 21, + PlayerPositionAndLookServerbound = 22, + PlayerLook = 23, + VehicleMoveServerbound = 24, + SteerBoat = 25, + PickItem = 26, + CraftRecipeRequest = 27, + PlayerAbilitiesServerbound = 28, + PlayerDigging = 29, + EntityAction = 30, + SteerVehicle = 31, + RecipeBookData = 32, + NameItem = 33, + ResourcePackStatus = 34, + AdvancementTab = 35, + SelectTrade = 36, + SetBeaconEffect = 37, + HeldItemChangeServerbound = 38, + UpdateCommandBlock = 39, + UpdateCommandBlockMinecart = 40, + CreativeInventoryAction = 41, + UpdateStructureBlock = 42, + UpdateSign = 43, + AnimationServerbound = 44, + Spectate = 45, + PlayerBlockPlacement = 46, + UseItem = 47, // Status - Request, - Ping, + Request = 48, + Ping = 49, // Clientbound @@ -103,103 +103,103 @@ pub enum PacketType { // (none) // Login - DisconnectLogin, - EncryptionRequest, - LoginSuccess, - SetCompression, - LoginPluginRequest, + DisconnectLogin = 50, + EncryptionRequest = 51, + LoginSuccess = 52, + SetCompression = 53, + LoginPluginRequest = 54, // Play - SpawnObject, - SpawnExperienceOrb, - SpawnGlobalOrb, - SpawnGlobalEntity, - SpawnMob, - SpawnPainting, - SpawnPlayer, - AnimationClientbound, - Statistics, - BlockBreakAnimation, - UpdateBlockEntity, - BlockAction, - BlockChange, - BossBar, - ServerDifficulty, - ChatMessageClientbound, - MultiBlockChange, - TabCompleteClientbound, - DeclareCommands, - ConfirmTransactionClientbound, - CloseWindowClientbound, - OpenWindow, - WindowItems, - WindowProperty, - SetSlot, - SetCooldown, - PluginMessageClientbound, - NamedSoundEffect, - DisconnectPlay, - EntityStatus, - NBTQueryResponse, - Explosion, - UnloadChunk, - ChangeGameState, - KeepAliveClientbound, - ChunkData, - Effect, - Particle, - JoinGame, - MapData, - Entity, - EntityRelativeMove, - EntityLookAndRelativeMove, - EntityLook, - VehicleMoveClientbound, - OpenSignEditor, - CraftRecipeResponse, - PlayerAbilitiesClientbound, - CombatEvent, - PlayerInfo, - FacePlayer, - PlayerPositionAndLookClientbound, - UseBed, - UnlockRecipes, - DestroyEntities, - RemoveEntityEffect, - ResourcePackSend, - Respawn, - EntityHeadLook, - SelectAdvancementTab, - WorldBorder, - Camera, - HeldItemChangeClientbound, - DisplayScoreboard, - EntityMetadata, - AttachEntity, - EntityVelocity, - EntityEquipment, - SetExperience, - UpdateHealth, - ScoreboardObjective, - SetPassengers, - Teams, - UpdateScore, - SpawnPosition, - TimeUpdate, - StopSound, - SoundEffect, - PlayerListHeaderAndFooter, - CollectItem, - EntityTeleport, - Advancements, - EntityProperties, - EntityEffect, - DeclareRecipes, - Tags, + SpawnObject = 55, + SpawnExperienceOrb = 56, + SpawnGlobalOrb = 57, + SpawnGlobalEntity = 58, + SpawnMob = 59, + SpawnPainting = 60, + SpawnPlayer = 61, + AnimationClientbound = 62, + Statistics = 63, + BlockBreakAnimation = 64, + UpdateBlockEntity = 65, + BlockAction = 66, + BlockChange = 67, + BossBar = 68, + ServerDifficulty = 69, + ChatMessageClientbound = 70, + MultiBlockChange = 71, + TabCompleteClientbound = 72, + DeclareCommands = 73, + ConfirmTransactionClientbound = 74, + CloseWindowClientbound = 75, + OpenWindow = 76, + WindowItems = 77, + WindowProperty = 78, + SetSlot = 79, + SetCooldown = 80, + PluginMessageClientbound = 81, + NamedSoundEffect = 82, + DisconnectPlay = 83, + EntityStatus = 84, + NBTQueryResponse = 85, + Explosion = 86, + UnloadChunk = 87, + ChangeGameState = 88, + KeepAliveClientbound = 89, + ChunkData = 90, + Effect = 91, + Particle = 92, + JoinGame = 93, + MapData = 94, + Entity = 95, + EntityRelativeMove = 96, + EntityLookAndRelativeMove = 97, + EntityLook = 98, + VehicleMoveClientbound = 99, + OpenSignEditor = 100, + CraftRecipeResponse = 101, + PlayerAbilitiesClientbound = 102, + CombatEvent = 103, + PlayerInfo = 104, + FacePlayer = 105, + PlayerPositionAndLookClientbound = 106, + UseBed = 107, + UnlockRecipes = 108, + DestroyEntities = 109, + RemoveEntityEffect = 110, + ResourcePackSend = 111, + Respawn = 112, + EntityHeadLook = 113, + SelectAdvancementTab = 114, + WorldBorder = 115, + Camera = 116, + HeldItemChangeClientbound = 117, + DisplayScoreboard = 118, + EntityMetadata = 119, + AttachEntity = 120, + EntityVelocity = 121, + EntityEquipment = 122, + SetExperience = 123, + UpdateHealth = 124, + ScoreboardObjective = 125, + SetPassengers = 126, + Teams = 127, + UpdateScore = 128, + SpawnPosition = 129, + TimeUpdate = 130, + StopSound = 131, + SoundEffect = 132, + PlayerListHeaderAndFooter = 133, + CollectItem = 134, + EntityTeleport = 135, + Advancements = 136, + EntityProperties = 137, + EntityEffect = 138, + DeclareRecipes = 139, + Tags = 140, // Status - Response, - Pong, + Response = 141, + Pong = 142, } lazy_static! { @@ -437,6 +437,11 @@ lazy_static! { PacketType::Pong, ); + m.insert( + PacketId(0x1B, PacketDirection::Clientbound, PacketStage::Play), + PacketType::DisconnectPlay, + ); + m.insert( PacketId(0x21, PacketDirection::Clientbound, PacketStage::Play), PacketType::KeepAliveClientbound, @@ -527,6 +532,12 @@ impl PacketType { pub fn get_implementation(&self) -> Box { implementation::IMPL_MAP.get(self).unwrap().build() } + + /// Returns a unique ID, allocated + /// consecutively for each packet type. + pub fn ordinal(&self) -> usize { + *self as usize + } } /// Certain packets have the same ID as diff --git a/core/src/world/mod.rs b/core/src/world/mod.rs index 6ca2b0524..febe7fd18 100644 --- a/core/src/world/mod.rs +++ b/core/src/world/mod.rs @@ -65,7 +65,7 @@ impl ChunkMap { /// Retrieves the chunk at the specified location. /// If the chunk is not loaded, `None` will be returned. - pub fn chunk_at(&mut self, pos: ChunkPosition) -> Option<&Chunk> { + pub fn chunk_at(&self, pos: ChunkPosition) -> Option<&Chunk> { if let Some(chunk) = self.chunk_map.get(&pos) { return Some(chunk); } @@ -76,7 +76,7 @@ impl ChunkMap { /// Retrieves the block at the specified /// location. If the chunk in which the block /// exists is not laoded, `None` is returned. - pub fn block_at(&mut self, pos: BlockPosition) -> Option { + pub fn block_at(&self, pos: BlockPosition) -> Option { let chunk_pos = pos.chunk_pos(); if let Some(chunk) = self.chunk_at(chunk_pos) { @@ -91,6 +91,10 @@ impl ChunkMap { /// If the chunk in which the position resides /// does not exist, `Err` is returned. In all /// other cases, `Ok` is returned. + /// + /// Note that on the server side, calling this function + /// does not broadcast the update in any way. As such, + /// the according function should be called instead. pub fn set_block_at(&mut self, pos: BlockPosition, block: Block) -> Result<(), ()> { let chunk_pos = pos.chunk_pos(); @@ -102,13 +106,24 @@ impl ChunkMap { Err(()) } } + + /// Sets the chunk at the given location. + pub fn set_chunk_at(&mut self, pos: ChunkPosition, chunk: Chunk) { + self.chunk_map.insert(pos, chunk); + } +} + +impl Default for ChunkMap { + fn default() -> Self { + Self::new() + } } fn chunk_relative_pos(block_pos: BlockPosition) -> (usize, usize, usize) { ( - (block_pos.x % 16) as usize, + block_pos.x as usize & 0xf, block_pos.y as usize, - (block_pos.z % 16) as usize, + block_pos.z as usize & 0xf, ) } diff --git a/server/Cargo.toml b/server/Cargo.toml index 06961083b..9d28b2c1d 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "feather_server" -version = "0.1.0" +version = "0.3.0" authors = ["caelunshun "] edition = "2018" @@ -8,7 +8,7 @@ edition = "2018" feather_core = { path = "../core" } feather_blocks = { path = "../blocks/lib" } mio = "0.6.19" -mio-extras = "2.0.5" +mio-extras = { git = "https://github.com/caelunshun/mio-extras", rev = "e5dca5d" } crossbeam = "0.7.1" log = "0.4.6" simple_logger = "1.3.0" @@ -26,6 +26,8 @@ mojang-api = "0.1.0" multimap = "0.5.0" rnbt = { git = "https://github.com/caelunshun/rnbt", version = "0.1.0" } specs = "0.15.0" +rayon = "1.1.0" +shrev = "1.1.1" [dev-dependencies] mockers = "0.21.0" diff --git a/server/config/feather.toml b/server/config/feather.toml index 32a3ae501..4d575eefc 100644 --- a/server/config/feather.toml +++ b/server/config/feather.toml @@ -12,7 +12,7 @@ proxy_mode = "none" [server] online_mode = true -motd = " &lMesa called Jar Jar Binks!\n Mesa your humble servant!" +motd = "A Feather server" max_players = 16 default_gamemode = "survival" difficulty = "none" diff --git a/server/src/chunkclient.rs b/server/src/chunkclient.rs new file mode 100644 index 000000000..a523637ee --- /dev/null +++ b/server/src/chunkclient.rs @@ -0,0 +1,142 @@ +//! Module for interacting with the chunk worker thread +//! from the server threads. +use crossbeam::channel::{Receiver, Sender}; +use shrev::EventChannel; +use specs::{Read, System, World, Write}; + +use feather_core::world::{ChunkMap, ChunkPosition}; + +use crate::chunkworker; + +pub struct ChunkWorkerHandle { + sender: Sender, + receiver: Receiver, +} + +impl Default for ChunkWorkerHandle { + fn default() -> Self { + let (sender, receiver) = chunkworker::start("world"); + Self { sender, receiver } + } +} + +/// Event which is triggered when a chunk is loaded. +#[derive(Debug, Clone, Copy)] +pub struct ChunkLoadEvent { + pub pos: ChunkPosition, +} + +/// System for receiving loaded chunks from the chunk worker thread. +pub struct ChunkLoadSystem; + +impl<'a> System<'a> for ChunkLoadSystem { + type SystemData = ( + Write<'a, ChunkMap>, + Write<'a, EventChannel>, + Read<'a, ChunkWorkerHandle>, + ); + + fn run(&mut self, data: Self::SystemData) { + let (mut chunk_map, mut events, handle) = data; + + while let Ok((pos, result)) = handle.receiver.try_recv() { + match result { + Ok(chunk) => { + chunk_map.set_chunk_at(pos, chunk); + + // Trigger event + let event = ChunkLoadEvent { pos }; + events.single_write(event); + + trace!("Loaded chunk at {:?}", pos); + } + Err(err) => { + // TODO generate chunk if it didn't exist + warn!("Failed to load chunk at {:?}: {}", pos, err); + } + } + } + } + + fn setup(&mut self, world: &mut World) { + use specs::prelude::SystemData; + + info!("Starting chunk worker thread"); + let handle = chunkworker::start("world"); + world.insert(handle); + + Self::SystemData::setup(world); + } +} + +/// Asynchronously loads the chunk at the given position. +/// At some point in time after this function is called, +/// the chunk will appear in the chunk map. +/// +/// In the event that the requested chunk does not exist +/// in the world save, it will be generated asynchronously. +pub fn load_chunk(handle: &ChunkWorkerHandle, pos: ChunkPosition) { + // Send request to chunk worker thread + handle + .sender + .send(chunkworker::Request::LoadChunk(pos)) + .unwrap(); +} + +#[cfg(test)] +mod tests { + use specs::{RunNow, World, WorldExt}; + + use feather_core::world::chunk::Chunk; + use feather_core::world::ChunkPosition; + + use super::*; + + #[test] + fn test_chunk_system() { + let (send1, _recv1) = crossbeam::channel::unbounded(); + let (send2, recv2) = crossbeam::channel::unbounded(); + let handle = ChunkWorkerHandle { + sender: send1, + receiver: recv2, + }; + + let chunk_map = ChunkMap::new(); + let pos = ChunkPosition::new(0, 0); + send2.send((pos, Ok(Chunk::new(pos)))).unwrap(); + + let event_channel = EventChannel::::new(); + + let mut system = ChunkLoadSystem; + let mut world = World::new(); + world.insert(chunk_map); + world.insert(handle); + world.insert(event_channel); + + system.run_now(&world); + + // Confirm that chunk was loaded + let chunk_map = world.read_resource::(); + let chunk = chunk_map.chunk_at(pos); + + assert!(chunk.is_some()); + assert!(chunk.unwrap().position() == pos); + } + + #[test] + fn test_load_chunk() { + let (send1, recv1) = crossbeam::channel::unbounded(); + let (_send2, recv2) = crossbeam::channel::unbounded(); + let handle = ChunkWorkerHandle { + sender: send1, + receiver: recv2, + }; + + let pos = ChunkPosition::new(0, 0); + + load_chunk(&handle, pos); + + let recv = recv1.try_recv().unwrap(); + assert_eq!(recv, chunkworker::Request::LoadChunk(pos)); + } +} diff --git a/server/src/chunkworker.rs b/server/src/chunkworker.rs index ebf43bb58..ba0833ffe 100644 --- a/server/src/chunkworker.rs +++ b/server/src/chunkworker.rs @@ -16,7 +16,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; pub type Reply = (ChunkPosition, Result); -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum Request { LoadChunk(ChunkPosition), ShutDown, diff --git a/server/src/config.rs b/server/src/config.rs index ebe3973a5..3203fc6ca 100644 --- a/server/src/config.rs +++ b/server/src/config.rs @@ -1,6 +1,6 @@ use std::fs::read_to_string; -#[derive(Deserialize, Debug)] +#[derive(Deserialize, Debug, Clone)] pub struct Config { pub io: IO, pub proxy: Proxy, @@ -9,16 +9,45 @@ pub struct Config { pub log: Log, } -#[derive(Deserialize, Debug)] +impl Default for Config { + fn default() -> Self { + Self { + io: IO { + compression_threshold: 256, + io_worker_threads: 4, + }, + proxy: Proxy {}, + server: Server { + online_mode: true, + motd: "A Feather server".to_string(), + max_players: 256, + view_distance: 6, + address: "126.0.0.1".to_string(), + port: 25565, + }, + gameplay: Gameplay { + monster_spawning: true, + animal_spawning: true, + pvp: true, + nerf_spawner_mobs: false, + }, + log: Log { + level: "debug".to_string(), + }, + } + } +} + +#[derive(Deserialize, Debug, Clone)] pub struct IO { pub compression_threshold: i32, pub io_worker_threads: u16, } -#[derive(Deserialize, Debug)] +#[derive(Deserialize, Debug, Clone)] pub struct Proxy {} -#[derive(Deserialize, Debug)] +#[derive(Deserialize, Debug, Clone)] pub struct Server { pub online_mode: bool, pub motd: String, @@ -28,7 +57,7 @@ pub struct Server { pub port: u16, } -#[derive(Deserialize, Debug)] +#[derive(Deserialize, Debug, Clone)] pub struct Gameplay { pub monster_spawning: bool, pub animal_spawning: bool, @@ -36,7 +65,7 @@ pub struct Gameplay { pub nerf_spawner_mobs: bool, } -#[derive(Deserialize, Debug)] +#[derive(Deserialize, Debug, Clone)] pub struct Log { pub level: String, } @@ -54,7 +83,7 @@ pub fn load(input: String) -> Result { Ok(config) } -#[derive(Serialize, Deserialize, Debug)] +#[derive(Deserialize, Debug, Clone)] pub enum ProxyMode { None, Bungee, diff --git a/server/src/entity.rs b/server/src/entity.rs new file mode 100644 index 000000000..5e938a684 --- /dev/null +++ b/server/src/entity.rs @@ -0,0 +1,96 @@ +//! Provides several useful components, including `EntityComponent` +//! and `PlayerComponent`. In the future, will also +//! provide entity-specific components and systems. + +use specs::storage::BTreeStorage; +use specs::{Component, Entities, Entity, ReadStorage, VecStorage}; +use uuid::Uuid; + +use feather_core::network::packet::implementation::{ + EntityHeadLook, EntityLook, EntityLookAndRelativeMove, EntityRelativeMove, +}; +use feather_core::world::Position; +use feather_core::Gamemode; + +use crate::network::{send_packet_to_all_players, NetworkComponent}; + +pub struct PlayerComponent { + pub profile_properties: Vec, + pub gamemode: Gamemode, +} + +impl Component for PlayerComponent { + type Storage = BTreeStorage; +} + +pub struct EntityComponent { + pub uuid: Uuid, + pub display_name: String, + pub position: Position, + pub on_ground: bool, +} + +impl Component for EntityComponent { + type Storage = VecStorage; +} + +/// Broadcasts to all joined players that an entity has moved. +pub fn broadcast_entity_movement( + entity: Entity, + old_pos: Position, + new_pos: Position, + has_moved: bool, + has_looked: bool, + netcomps: &ReadStorage, + pcomps: &ReadStorage, + entities: &Entities, +) { + assert!(has_moved || has_looked); + + if has_moved { + let (rx, ry, rz) = calculate_relative_move(old_pos, new_pos); + + if has_looked { + let packet = EntityLookAndRelativeMove::new( + entity.id() as i32, + rx, + ry, + rz, + degrees_to_stops(new_pos.yaw), + degrees_to_stops(new_pos.pitch), + true, + ); + send_packet_to_all_players(netcomps, pcomps, entities, packet, Some(entity)); + } else { + let packet = EntityRelativeMove::new(entity.id() as i32, rx, ry, rz, true); + send_packet_to_all_players(netcomps, pcomps, entities, packet, Some(entity)); + } + } else { + let packet = EntityLook::new( + entity.id() as i32, + degrees_to_stops(new_pos.yaw), + degrees_to_stops(new_pos.pitch), + true, + ); + send_packet_to_all_players(netcomps, pcomps, entities, packet, Some(entity)); + } + + // Entity Head Look also needs to be sent if the entity turned its head + if has_looked { + let packet = EntityHeadLook::new(entity.id() as i32, degrees_to_stops(new_pos.yaw)); + send_packet_to_all_players(netcomps, pcomps, entities, packet, Some(entity)); + } +} + +/// Calculates the relative move fields +/// as used in the Entity Relative Move packets. +pub fn calculate_relative_move(old: Position, current: Position) -> (i16, i16, i16) { + let x = ((current.x * 32.0 - old.x * 32.0) * 128.0) as i16; + let y = ((current.y * 32.0 - old.y * 32.0) * 128.0) as i16; + let z = ((current.z * 32.0 - old.z * 32.0) * 128.0) as i16; + (x, y, z) +} + +pub fn degrees_to_stops(degs: f32) -> u8 { + ((degs / 360.0) * 256.0) as u8 +} diff --git a/server/src/genindex.rs b/server/src/genindex.rs deleted file mode 100644 index c65b7ea78..000000000 --- a/server/src/genindex.rs +++ /dev/null @@ -1,193 +0,0 @@ -use std::ops::{Index, IndexMut}; - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct GenerationalIndex { - index: usize, - generation: u32, -} - -impl GenerationalIndex { - pub fn index(&self) -> usize { - self.index - } -} - -struct AllocatorEntry { - is_live: bool, - generation: u32, -} - -pub struct GenerationalIndexAllocator { - entries: Vec, - free: Vec, -} - -impl GenerationalIndexAllocator { - pub fn new() -> Self { - Self { - entries: vec![], - free: vec![], - } - } - pub fn allocate(&mut self) -> GenerationalIndex { - if self.free.is_empty() { - self.entries.push(AllocatorEntry { - is_live: true, - generation: 0, - }); - - GenerationalIndex { - index: self.entries.len() - 1, - generation: 0, - } - } else { - let index = self.free.pop().unwrap(); - let entry = &mut self.entries[index]; - - entry.is_live = true; - entry.generation += 1; - - GenerationalIndex { - index, - generation: entry.generation, - } - } - } - - pub fn deallocate(&mut self, genindex: GenerationalIndex) { - let index = genindex.index; - - assert_eq!(self.entries[index].generation, genindex.generation); - - self.free.push(index); - self.entries[index].is_live = false; - } -} - -struct ArrayEntry { - value: T, - generation: u32, -} - -pub struct GenerationalArray(Vec>>); - -impl GenerationalArray { - pub fn new() -> Self { - Self(vec![]) - } - - pub fn set(&mut self, index: GenerationalIndex, value: T) { - while self.0.len() <= index.index() { - self.0.push(None); - } - - self.0[index.index()] = Some(ArrayEntry { - value, - generation: index.generation, - }); - } - - pub fn remove(&mut self, index: GenerationalIndex) -> Option { - let entry = &mut self.0[index.index()]; - if let Some(entry) = entry { - if entry.generation == index.generation { - let entry = self.0.get_mut(index.index()).unwrap().take().unwrap(); - return Some(entry.value); - } - } - - None - } - - pub fn get(&self, index: GenerationalIndex) -> Option<&T> { - let entry = self.0.get(index.index()); - - if entry.is_none() { - return None; - } - - let entry = entry.unwrap(); - if let Some(entry) = entry { - if entry.generation != index.generation { - return None; - } - - Some(&entry.value) - } else { - None - } - } - - pub fn get_mut(&mut self, index: GenerationalIndex) -> Option<&mut T> { - let entry = self.0.get_mut(index.index()); - - if entry.is_none() { - return None; - } - - let entry = entry.unwrap(); - if let Some(entry) = entry { - if entry.generation != index.generation { - return None; - } - - Some(&mut entry.value) - } else { - None - } - } -} - -impl Index for GenerationalArray { - type Output = T; - - fn index(&self, index: GenerationalIndex) -> &Self::Output { - self.get(index).unwrap() - } -} - -impl IndexMut for GenerationalArray { - fn index_mut(&mut self, index: GenerationalIndex) -> &mut Self::Output { - self.get_mut(index).unwrap() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_allocator() { - let mut allocator = GenerationalIndexAllocator::new(); - - let index = allocator.allocate(); - assert_eq!(index.generation, 0); - assert_eq!(index.index(), 0); - - let index2 = allocator.allocate(); - assert_eq!(index2.generation, 0); - assert_eq!(index2.index(), 1); - - allocator.deallocate(index); - let index3 = allocator.allocate(); - assert_eq!(index.index(), index3.index()); - assert_eq!(index3.generation, 1); - } - - #[test] - fn test_array() { - let mut arr = GenerationalArray::new(); - let mut alloc = GenerationalIndexAllocator::new(); - - let index1 = alloc.allocate(); - arr.set(index1, "test"); - assert_eq!(arr.get(index1), Some(&"test")); - - alloc.deallocate(index1); - - let index2 = alloc.allocate(); - arr.set(index2, "test2"); - assert_eq!(arr.get(index2), Some(&"test2")); - assert_eq!(arr.get(index1), None); - } -} diff --git a/server/src/initialhandler.rs b/server/src/initialhandler.rs deleted file mode 100644 index fee38c8e3..000000000 --- a/server/src/initialhandler.rs +++ /dev/null @@ -1,425 +0,0 @@ -use crate::network::{ - broadcast_player_join, enable_compression_for_player, enable_encryption_for_player, - get_player_initialization_packets, send_packet_to_player, -}; -use crate::prelude::*; -use crate::{load_chunk, remove_player, Entity, EntityComponent, PlayerComponent, State}; -use feather_core::network::packet::{implementation::*, Packet, PacketType}; -use openssl::pkey::Private; -use openssl::rsa::{Padding, Rsa}; -use std::fmt::Formatter; -use std::str::FromStr; - -use super::{PROTOCOL_VERSION, SERVER_VERSION}; -use mojang_api::ServerAuthResponse; - -const RSA_BITS: u32 = 1024; // Yes, very secure - -const VERIFY_TOKEN_LEN: usize = 4; - -#[derive(Debug, Eq, PartialEq)] -pub enum Stage { - AwaitHandshake, - - AwaitRequest, - AwaitPing, - - AwaitLoginStart, - AwaitEncryptionResponse, - AwaitChunkLoad, -} - -pub struct InitialHandlerComponent { - pub stage: Stage, - rsa_key: Option>, - verify_token: [u8; VERIFY_TOKEN_LEN], - /// Sent in Login Start - username: Option, -} - -impl InitialHandlerComponent { - pub fn new() -> Self { - Self { - stage: Stage::AwaitHandshake, - rsa_key: None, - verify_token: rand::random(), - username: None, - } - } -} - -pub enum Error { - InvalidPacket(PacketType), - MalformedData(String), - InvalidProtocolVersion(u32, u32), - AuthenticationFailed, -} - -impl std::fmt::Display for Error { - fn fmt(&self, f: &mut Formatter) -> Result<(), std::fmt::Error> { - match self { - Error::InvalidPacket(ty) => f - .write_str(&format!("Sent invalid packet type {:?}", ty)) - .unwrap(), - Error::MalformedData(details) => f - .write_str(&format!("Sent invalid data: {}", details)) - .unwrap(), - Error::InvalidProtocolVersion(server, client) => f - .write_str(&format!( - "Protocol versions do not match: client is on {}; server is on {}", - client, server - )) - .unwrap(), - Error::AuthenticationFailed => f.write_str("Authentication failed").unwrap(), - }; - - Ok(()) - } -} - -pub fn handle_packet(state: &mut State, player: Entity, packet: Box) -> Result<(), Error> { - match packet.ty() { - PacketType::Handshake => { - handle_handshake(state, player, cast_packet::(&packet))? - } - PacketType::Request => handle_request(state, player, cast_packet::(&packet))?, - PacketType::Ping => handle_ping(state, player, cast_packet::(&packet))?, - PacketType::LoginStart => { - handle_login_start(state, player, cast_packet::(&packet))? - } - PacketType::EncryptionResponse => { - handle_encryption_response(state, player, cast_packet::(&packet))? - } - _ => return Err(Error::InvalidPacket(packet.ty())), - } - Ok(()) -} - -fn handle_handshake(state: &mut State, player: Entity, packet: &Handshake) -> Result<(), Error> { - let ih = state.ih_components.get_mut(player).unwrap(); - - if ih.stage != Stage::AwaitHandshake { - return Err(Error::InvalidPacket(PacketType::Handshake)); - } - - if packet.protocol_version != PROTOCOL_VERSION && packet.next_state != HandshakeState::Status { - return Err(Error::InvalidProtocolVersion( - PROTOCOL_VERSION, - packet.protocol_version, - )); - } - - match packet.next_state { - HandshakeState::Login => ih.stage = Stage::AwaitLoginStart, - HandshakeState::Status => ih.stage = Stage::AwaitRequest, - } - - Ok(()) -} - -fn handle_request(state: &mut State, player: Entity, _packet: &Request) -> Result<(), Error> { - let ih = state.ih_components.get_mut(player).unwrap(); - - if ih.stage != Stage::AwaitRequest { - return Err(Error::InvalidPacket(PacketType::Request)); - } - - let payload = json!({ - "version": { - "name": SERVER_VERSION, - "protocol": PROTOCOL_VERSION, - }, - "players": { - "max": state.config.server.max_players, - "online": state.joined_players.len(), - }, - "description": state.config.server.motd, - }); - - let response = Response::new(payload.to_string()); - - ih.stage = Stage::AwaitPing; - - send_packet_to_player(state, player, response); - - Ok(()) -} - -fn handle_ping(state: &mut State, player: Entity, packet: &Ping) -> Result<(), Error> { - let ih = state.ih_components.get_mut(player).unwrap(); - - if ih.stage != Stage::AwaitPing { - return Err(Error::InvalidPacket(PacketType::Ping)); - } - - let pong = Pong::new(packet.payload); - send_packet_to_player(state, player, pong); - - remove_player(state, player); - debug!("Status handling success"); - - Ok(()) -} - -fn handle_login_start(state: &mut State, player: Entity, packet: &LoginStart) -> Result<(), Error> { - state.ih_components[player].username = Some(packet.username.clone()); - - let ih = state.ih_components.get(player).unwrap(); - - if ih.stage != Stage::AwaitLoginStart { - return Err(Error::InvalidPacket(PacketType::Ping)); - } - - // If in online mode, enable encryption - if state.config.server.online_mode { - let rsa_key = Rsa::generate(RSA_BITS).unwrap(); - - let key_bytes = rsa_key.public_key_to_der().unwrap(); - - let mut verify_token = vec![]; - verify_token.extend_from_slice(&ih.verify_token); - - let encryption_request = EncryptionRequest::new( - "".to_string(), // Server ID - always empty nowadays - key_bytes.len() as i32, - key_bytes, - VERIFY_TOKEN_LEN as i32, - verify_token, - ); - - state.ih_components[player].rsa_key = Some(rsa_key); - state.ih_components[player].stage = Stage::AwaitEncryptionResponse; - - send_packet_to_player(state, player, encryption_request); - } else { - // Login completed - initialize entity and player components - // This would otherwise be done in `handle_encryption_response` - let player_comp = PlayerComponent { - profile_properties: vec![], - gamemode: Gamemode::Creative, - }; - state.player_components.set(player, player_comp); - - let entity_comp = EntityComponent { - position: Position::new(0.0, 0.0, 0.0, 0.0, 0.0), - uuid: Uuid::new_v4(), - display_name: ih.username.as_ref().unwrap().clone(), - on_ground: true, - }; - state.entity_components.set(player, entity_comp); - - finish(state, player); - } - - Ok(()) -} - -fn handle_encryption_response( - state: &mut State, - player: Entity, - packet: &EncryptionResponse, -) -> Result<(), Error> { - let ih = state.ih_components.get(player).unwrap(); - - if ih.stage != Stage::AwaitEncryptionResponse { - return Err(Error::InvalidPacket(PacketType::Ping)); - } - - let rsa = ih.rsa_key.as_ref().unwrap(); - - let secret: [u8; 16] = { - let mut buf = vec![0u8; rsa.size() as usize]; - - if let Ok(amnt) = rsa.private_decrypt(&packet.secret, &mut buf, Padding::PKCS1) { - if amnt != 16 { - return Err(Error::MalformedData(format!( - "Invalid shared secret length {}", - amnt - ))); - } - - let mut res = [0u8; 16]; - for (i, val) in buf[..16].iter().enumerate() { - res[i] = *val; - } - res - } else { - return Err(Error::MalformedData( - "Error decrypting shared secret".to_string(), - )); - } - }; - - let verify_token: [u8; VERIFY_TOKEN_LEN] = { - let mut buf = vec![0u8; rsa.size() as usize]; - - if let Ok(amnt) = rsa.private_decrypt(&packet.verify_token, &mut buf, Padding::PKCS1) { - if amnt != VERIFY_TOKEN_LEN { - return Err(Error::MalformedData(format!( - "Invalid verify token length {}", - amnt - ))); - } - - let mut res = [0u8; VERIFY_TOKEN_LEN]; - for (i, val) in buf[..VERIFY_TOKEN_LEN].iter().enumerate() { - res[i] = *val; - } - res - } else { - return Err(Error::MalformedData( - "Error decrypting verify token".to_string(), - )); - } - }; - - if !compare_verify_tokens(ih.verify_token.clone(), verify_token) { - return Err(Error::MalformedData( - "Verify tokens do not match".to_string(), - )); - } - - // Authenticate - let auth_res = authenticate( - secret.clone(), - &ih.rsa_key.as_ref().unwrap().public_key_to_der().unwrap(), - ih.username.as_ref().unwrap(), - ); - if let Ok(res) = auth_res { - let player_comp = PlayerComponent { - profile_properties: res.properties, - gamemode: Gamemode::Creative, - }; - state.player_components.set(player, player_comp); - - let entity_comp = EntityComponent { - position: Position::new(0.0, 64.0, 0.0, 0.0, 0.0), - uuid: Uuid::from_str(&res.id).unwrap(), - display_name: ih.username.as_ref().unwrap().clone(), - on_ground: true, - }; - state.entity_components.set(player, entity_comp); - debug!("Authentication successful"); - } else { - return Err(Error::AuthenticationFailed); - } - - enable_encryption_for_player(state, player, secret); - - finish(state, player); - - Ok(()) -} - -fn finish(state: &mut State, player: Entity) { - // Enable compression if needed - let threshold = state.config.io.compression_threshold; - if threshold > 0 { - let set_compression = SetCompression::new(threshold); - send_packet_to_player(state, player, set_compression); - - enable_compression_for_player(state, player, threshold as usize); - } - - let entity_comp = state.entity_components.get(player).unwrap(); - - let login_success = LoginSuccess::new( - entity_comp.uuid.to_hyphenated_ref().to_string(), - entity_comp.display_name.to_string(), - ); - send_packet_to_player(state, player, login_success); - - join_game(state, player); -} - -fn join_game(state: &mut State, player: Entity) { - let join_game = JoinGame::new( - player.index() as i32, - Gamemode::Creative.get_id(), - Dimension::Overwold.get_id(), - Difficulty::Medium.get_id(), - 0, - "default".to_string(), - false, - ); - send_packet_to_player(state, player, join_game); - - // Send chunk data. If a chunk in the view distance - // hasn't been loaded, a load request will be queued. - let view_distance = state.config.server.view_distance as i32; - for x in -view_distance..view_distance + 1 { - for y in -view_distance..view_distance + 1 { - let pos = ChunkPosition::new(x, y); - - if let Some(chunk) = state.chunk_map.chunk_at(pos) { - let chunk_data = ChunkData::new(chunk.clone()); - send_packet_to_player(state, player, chunk_data); - } else { - // Queue chunk for loading. - load_chunk(state, pos); - // Make sure that the chunk is sent once loaded - state.network_components[player].chunks_to_send.push(pos); - } - } - } - - if state.network_components[player].chunks_to_send.is_empty() { - complete_join_game(state, player); - } else { - // If not all chunks have been sent, we need to wait - // until they're loaded and sent before spawning the player. - // The network system will call `complete_join_game` upon - // sending all chunks. - state.ih_components[player].stage = Stage::AwaitChunkLoad; - } -} - -pub fn complete_join_game(state: &mut State, player: Entity) { - // Send spawn position + player position - // TODO proper persistence - - let spawn_position = SpawnPosition::new(BlockPosition::new(0, 64, 0)); - send_packet_to_player(state, player, spawn_position); - - let pos_and_look = PlayerPositionAndLookClientbound::new(0.0, 64.0, 0.0, 0.0, 0.0, 0, 0); - send_packet_to_player(state, player, pos_and_look); - - // Send other players on the server - for other_player in &state.joined_players { - let (player_info, spawn_player) = get_player_initialization_packets(state, *other_player); - send_packet_to_player(state, player, player_info); - send_packet_to_player(state, player, spawn_player); - } - - state.joined_players.push(player); - - state.ih_components.remove(player); - - broadcast_player_join(state, player); - - info!("A player joined the game"); -} - -/// Authenticates a client. -fn authenticate(secret: [u8; 16], pubkey: &[u8], username: &str) -> Result { - let server_hash = mojang_api::server_hash("", secret, pubkey); - let res = mojang_api::server_auth(username, &server_hash); - - res.map_err(|_| ()) -} - -fn compare_verify_tokens(x: [u8; VERIFY_TOKEN_LEN], y: [u8; VERIFY_TOKEN_LEN]) -> bool { - for i in 0..VERIFY_TOKEN_LEN { - if x[i] != y[i] { - return false; - } - } - - true -} - -pub fn disconnect_login(state: &mut State, player: Entity, reason: &str) { - let packet = DisconnectLogin::new(json!({ "text": reason }).to_string()); - - send_packet_to_player(state, player, packet); -} diff --git a/server/src/io/initialhandler.rs b/server/src/io/initialhandler.rs new file mode 100644 index 000000000..79e2fd570 --- /dev/null +++ b/server/src/io/initialhandler.rs @@ -0,0 +1,615 @@ +//! The initial handler is responsible for +//! handling new connections and getting +//! through the login sequence. After login +//! is completed, control is handed over to the server +//! thread, which is responsible for sending chunks/inventory/ +//! players and then spawning the player. +//! +//! The initial handler is also responsible for handling +//! server list pings. To do this, it shares an `Arc` +//! representing the player count with the server. +//! +//! The initial handler runs on the IO worker thread. +//! This is done to ensure minimal latency in packet handling, +//! speeding up the login process and making the latency calculation in +//! the server list ping as low as possible. + +use std::str::FromStr; +use std::sync::atomic::Ordering; +use std::sync::Arc; + +use openssl::pkey::Private; +use openssl::rsa::{Padding, Rsa}; +use uuid::Uuid; + +use feather_core::network::cast_packet; +use feather_core::network::packet::implementation::{ + DisconnectLogin, EncryptionRequest, EncryptionResponse, Handshake, HandshakeState, LoginStart, + LoginSuccess, Ping, Pong, Request, Response, SetCompression, +}; +use feather_core::network::packet::{Packet, PacketType}; + +use crate::config::Config; +use crate::{PlayerCount, PROTOCOL_VERSION, SERVER_VERSION}; + +/// The key used for symmetric encryption. +pub type Key = [u8; 16]; +/// The verify token used to ensure that encryption +/// is working correctly. +type VerifyToken = [u8; 4]; + +/// The number of bits used for the RSA key. +const RSA_KEY_BITS: u32 = 1024; +/// The number of bytes in the shared secret +const SHARED_SECRET_LEN: usize = 128 / 8; + +/// An action for the worker thread to execute +/// after `InitialHandler::handle_packet` is called. +pub enum Action { + EnableCompression(i32), + EnableEncryption(Key), + SendPacket(Box), + Disconnect, + JoinGame(JoinResult), +} + +/// The type returned for when a player has completed the login process. +#[derive(Clone, Debug)] +pub struct JoinResult { + pub username: String, + pub uuid: Uuid, + pub props: Vec, +} + +/// An initial handler for a connection. +/// +/// When a packet is received from the client this initial +/// handler is registered with, `handle_packet` should be called. +/// This function runs all the necessary code to handle the +/// login sequence or the server list ping. +/// +/// The initial handler is able to communicate with the worker +/// implementation by exposing the `actions_to_execute` method, +/// which returns a vector of actions for the worker to execute. +/// These may include, for example, enabling encryption or sending +/// a packet. +pub struct InitialHandler { + /// A queue of actions to perform. When `actions_to_execute` + /// is called, the queue is flushed. + action_queue: Vec, + + /// If set to a value, indicates that encryption + /// should be enabled with the given key. + key: Option, + /// If set to a value, indicates that compression + /// should be enabled with the given threshold. + compression_threshold: Option, + + /// The 1024-bit RSA key used for key exchange + /// with this client. + rsa: Rsa, + /// The verify token generated for this exchange. + verify_token: VerifyToken, + + /// The server's configuration. + config: Arc, + /// The server's player count. + player_count: Arc, + + /// The username of the player, sent + /// in Login Start. + username: Option, + + /// The player info, set to `Some` once + /// the initial handler is finished and + /// the player should join. + info: Option, + + /// The stage of this initial handler. + stage: Stage, +} + +impl InitialHandler { + pub fn new(config: Arc, player_count: Arc) -> Self { + Self { + action_queue: vec![], + + key: None, + compression_threshold: None, + + rsa: Rsa::generate(RSA_KEY_BITS).unwrap(), + verify_token: rand::random(), + + config, + player_count, + + username: None, + + info: None, + + stage: Stage::AwaitHandshake, + } + } + + /// Notifies this initial handler of a packet + /// received from the client. After calling this + /// function, `action_queue` should be called + /// and the actions should be executed in order. + pub fn handle_packet(&mut self, packet: Box) { + if self.stage == Stage::Finished { + panic!("Called InitialHandler::handle_packet() after completion"); + } + + if let Err(e) = _handle_packet(self, packet) { + // Disconnect + disconnect_login(self, &format!("{}", e)); + info!( + "Player {} disconnected: {}", + self.username.as_ref().unwrap_or(&"unknown".to_string()), + e + ); + } + } + + /// Returns a vector of actions to perform. + pub fn actions_to_execute(&mut self) -> Vec { + let mut new_vec = vec![]; + std::mem::swap(&mut new_vec, &mut self.action_queue); + + new_vec + } +} + +/// Handles a packet, returning `Err` if the player +/// should be disconnected. +fn _handle_packet(ih: &mut InitialHandler, packet: Box) -> Result<(), Error> { + // Find packet type and forward to correct function + match packet.ty() { + PacketType::Handshake => handle_handshake(ih, cast_packet::(&packet))?, + PacketType::Request => handle_request(ih, cast_packet::(&packet))?, + PacketType::Ping => handle_ping(ih, cast_packet::(&packet))?, + PacketType::LoginStart => handle_login_start(ih, cast_packet::(&packet))?, + PacketType::EncryptionResponse => { + handle_encryption_response(ih, cast_packet::(&packet))? + } + ty => return Err(Error::InvalidPacket(ty, ih.stage)), + } + + Ok(()) +} + +fn handle_handshake(ih: &mut InitialHandler, packet: &Handshake) -> Result<(), Error> { + check_stage(ih, Stage::AwaitHandshake, packet.ty())?; + + ih.stage = match packet.next_state { + HandshakeState::Status => Stage::AwaitRequest, + HandshakeState::Login => { + // While status requests can use differing + // protocol versions, a client + // needs to have a matching protocol version + // to log in. + if packet.protocol_version != PROTOCOL_VERSION { + return Err(Error::InvalidProtocol(packet.protocol_version)); + } + + Stage::AwaitLoginStart + } + }; + + Ok(()) +} + +fn handle_request(ih: &mut InitialHandler, packet: &Request) -> Result<(), Error> { + check_stage(ih, Stage::AwaitRequest, packet.ty())?; + + // Send response packet + let json = json!({ + "version": { + "name": SERVER_VERSION, + "protocol": PROTOCOL_VERSION, + }, + "players": { + "max": ih.config.server.max_players, + "online": ih.player_count.0.load(Ordering::SeqCst), + }, + "description": { + "text": ih.config.server.motd, + } + }); + + let response = Response::new(json.to_string()); + send_packet(ih, response); + + ih.stage = Stage::AwaitPing; + + Ok(()) +} + +fn handle_ping(ih: &mut InitialHandler, packet: &Ping) -> Result<(), Error> { + check_stage(ih, Stage::AwaitPing, packet.ty())?; + + let pong = Pong::new(packet.payload); + send_packet(ih, pong); + + // After sending pong, we should disconnect. + ih.action_queue.push(Action::Disconnect); + ih.stage = Stage::Finished; + + Ok(()) +} + +fn handle_login_start(ih: &mut InitialHandler, packet: &LoginStart) -> Result<(), Error> { + check_stage(ih, Stage::AwaitLoginStart, packet.ty())?; + + ih.username = Some(packet.username.clone()); + + // If in online mode, encryption needs to be enabled, + // and authentication needs to be performed. + // If not in online mode, the login sequence is + // already finished, so we can call `finish` after + // setting the player's info. + if ih.config.server.online_mode { + // Start enabling encryption + let encryption_request = EncryptionRequest::new( + "".to_string(), // Server ID - always empty + ih.rsa.public_key_to_der().unwrap(), + ih.verify_token.to_vec(), + ); + send_packet(ih, encryption_request); + + ih.stage = Stage::AwaitEncryptionResponse; + } else { + // Finished - set info and join + ih.info = Some(JoinResult { + username: ih.username.clone().unwrap(), + uuid: Uuid::new_v4(), + props: vec![], + }); + finish(ih); + } + + Ok(()) +} + +fn handle_encryption_response( + ih: &mut InitialHandler, + packet: &EncryptionResponse, +) -> Result<(), Error> { + check_stage(ih, Stage::AwaitEncryptionResponse, packet.ty())?; + + // Decrypt verify token + shared secret + let shared_secret = decrypt_using_rsa(&packet.secret, &ih.rsa)?; + if shared_secret.len() != SHARED_SECRET_LEN { + return Err(Error::BadSecretLength); + } + + let verify_token = decrypt_using_rsa(&packet.verify_token, &ih.rsa)?; + if verify_token.len() != ih.verify_token.len() { + return Err(Error::VerifyTokenMismatch); + } + + // Check that verify token matches + if verify_token.as_slice() != &ih.verify_token { + return Err(Error::VerifyTokenMismatch); + } + + // Enable encryption + let mut key = [0u8; SHARED_SECRET_LEN]; + for (i, x) in shared_secret[..SHARED_SECRET_LEN].into_iter().enumerate() { + key[i] = *x; + } + + ih.key = Some(key); + ih.action_queue + .push(Action::EnableEncryption(ih.key.clone().unwrap())); + + // Perform authentication + let auth_result = mojang_api::server_auth( + ih.username.as_ref().unwrap(), + &mojang_api::server_hash( + "", + ih.key.clone().unwrap(), + ih.rsa.public_key_to_der().unwrap().as_slice(), + ), + ); + + match auth_result { + Ok(auth) => { + let info = JoinResult { + username: auth.name, + uuid: Uuid::from_str(&auth.id).unwrap(), + props: auth.properties, + }; + ih.info = Some(info); + } + Err(_) => return Err(Error::AuthenticationFailed), + } + + finish(ih); + + Ok(()) +} + +fn decrypt_using_rsa(data: &[u8], key: &Rsa) -> Result, Error> { + let mut buf = vec![0u8; key.size() as usize]; + + let len = key + .private_decrypt(data, &mut buf, Padding::PKCS1) + .map_err(|_| Error::BadEncryption)?; + + buf.truncate(len); + + Ok(buf) +} + +/// Terminates the login process, sending Set Compression (if necessary) +/// and Login Success. +/// +/// Before calling this function, it is expected that: +/// * `info` is set to a valid value +/// * Encryption has been enabled, if necessary +/// * All other login processes have already run +fn finish(ih: &mut InitialHandler) { + assert!(ih.info.is_some()); + + // Enable compression if necessary + let compression_threshold = ih.config.io.compression_threshold; + if compression_threshold > 0 { + enable_compression(ih, compression_threshold); + } + + let info = ih.info.as_ref().unwrap(); + + // Send Login Success + let login_success = LoginSuccess::new( + info.uuid.to_hyphenated_ref().to_string(), + info.username.clone(), + ); + send_packet(ih, login_success); + ih.action_queue + .push(Action::JoinGame(ih.info.clone().unwrap())); +} + +/// Enables compression, sending the Set Compression +/// packet. +fn enable_compression(ih: &mut InitialHandler, threshold: i32) { + ih.compression_threshold = Some(threshold); + send_packet(ih, SetCompression::new(threshold)); + ih.action_queue.push(Action::EnableCompression(threshold)); +} + +/// Checks that the initial handler stage matches +/// the expected stage, returning `Err` with a proper +/// error message if not. +fn check_stage(ih: &InitialHandler, expected: Stage, packet_ty: PacketType) -> Result<(), Error> { + if ih.stage != expected { + Err(Error::InvalidPacket(packet_ty, ih.stage)) + } else { + Ok(()) + } +} + +/// Disconnects the initial handler, sending +/// a disconnect packet containing the reason. +fn disconnect_login(ih: &mut InitialHandler, reason: &str) { + let json = json!({ + "text": reason, + }) + .to_string(); + + let packet = DisconnectLogin::new(json); + send_packet(ih, packet); + + ih.action_queue.push(Action::Disconnect); +} + +/// Adds a packet to the internal packet queue. +fn send_packet(ih: &mut InitialHandler, packet: P) { + ih.action_queue.push(Action::SendPacket(Box::new(packet))); +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Error { + InvalidPacket(PacketType, Stage), + InvalidProtocol(u32), + BadEncryption, + VerifyTokenMismatch, + BadSecretLength, + AuthenticationFailed, +} + +impl std::fmt::Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> { + match self { + Error::InvalidPacket(ty, stage) => write!( + f, + "Sent invalid packet {:?} at initial handler stage {:?}", + ty, stage + )?, + Error::InvalidProtocol(protocol) => write!( + f, + "Invalid protocol version {} - this server is on {}", + *protocol, PROTOCOL_VERSION + )?, + Error::BadEncryption => write!(f, "Failed to decrypt value")?, + Error::VerifyTokenMismatch => write!(f, "Verify token does not match")?, + Error::BadSecretLength => write!(f, "Invalid shared secret length")?, + Error::AuthenticationFailed => write!(f, "Authentication failed")?, + } + + Ok(()) + } +} + +/// The stage of an initial handler. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Stage { + AwaitHandshake, + AwaitRequest, + AwaitPing, + AwaitLoginStart, + AwaitEncryptionResponse, + Finished, +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::AtomicUsize; + + use feather_core::network::cast_packet; + use feather_core::network::packet::implementation::{ + Handshake, HandshakeState, LoginSuccess, Ping, Pong, Request, Response, SetCompression, + }; + use feather_core::network::packet::PacketType; + + use crate::PROTOCOL_VERSION; + + use super::*; + + #[test] + fn test_initial_handler_new() { + let mut ih = ih(); + + assert!(ih.actions_to_execute().is_empty()); + } + + #[test] + fn test_status_ping() { + let player_count = 24; + let mut ih = ih_with_player_count(player_count); + + let handshake = Handshake::new( + PROTOCOL_VERSION, + "".to_string(), // Unused - server address + 25565, + HandshakeState::Status, + ); + ih.handle_packet(Box::new(handshake)); + + // Confirm that no packets were sent and the player wasn't disconnected + assert!(ih.actions_to_execute().is_empty()); + + let request = Request::new(); + ih.handle_packet(Box::new(request)); + + let actions = ih.actions_to_execute(); + + // Confirm that correct response was received + assert_eq!(actions.len(), 1); + + let _response = actions.first().unwrap(); + match _response { + Action::SendPacket(_response) => { + assert_eq!(_response.ty(), PacketType::Response); + + let response = cast_packet::(&_response); + let _: serde_json::Value = serde_json::from_str(&response.json_response).unwrap(); + } + _ => panic!(), + } + + // Send ping + let payload = 39842; + let ping = Ping::new(payload); + ih.handle_packet(Box::new(ping)); + + let mut actions = ih.actions_to_execute(); + + assert_eq!(actions.len(), 2); + let _pong = actions.remove(0); + match _pong { + Action::SendPacket(_pong) => { + assert_eq!(_pong.ty(), PacketType::Pong); + let pong = cast_packet::(&_pong); + assert_eq!(pong.payload, payload); + } + _ => panic!(), + } + + let disconnect = actions.remove(0); + match disconnect { + Action::Disconnect => (), + _ => panic!(), + } + } + + #[test] + fn test_login_sequence() { + let mut config = Config::default(); + config.server.online_mode = false; + let mut ih = ih_with_config(config.clone()); + + let handshake = Handshake::new( + PROTOCOL_VERSION, + "".to_string(), // Unused - server address + 25565, + HandshakeState::Login, + ); + ih.handle_packet(Box::new(handshake)); + + assert!(ih.actions_to_execute().is_empty()); + + let username = "test"; + let login_start = LoginStart::new(username.to_string()); + ih.handle_packet(Box::new(login_start)); + + let mut actions = ih.actions_to_execute(); + assert_eq!(actions.len(), 4); + + let _set_compression = actions.remove(0); + + match _set_compression { + Action::SendPacket(_set_compression) => { + assert_eq!(_set_compression.ty(), PacketType::SetCompression); + + let set_compression = cast_packet::(&_set_compression); + assert_eq!(set_compression.threshold, config.io.compression_threshold); + } + _ => panic!(), + } + + let enable_compression = actions.remove(0); + match enable_compression { + Action::EnableCompression(threshold) => { + assert_eq!(threshold, config.io.compression_threshold); + } + _ => panic!(), + } + + let _login_success = actions.remove(0); + + match _login_success { + Action::SendPacket(_login_success) => { + assert_eq!(_login_success.ty(), PacketType::LoginSuccess); + + let login_success = cast_packet::(&_login_success); + assert_eq!(login_success.username, username.to_string()); + } + _ => panic!(), + } + + let join = actions.remove(0); + match join { + Action::JoinGame(_) => (), + _ => panic!(), + } + } + + fn ih() -> InitialHandler { + InitialHandler::new( + Arc::new(Config::default()), + Arc::new(PlayerCount(AtomicUsize::new(0))), + ) + } + + fn ih_with_player_count(count: usize) -> InitialHandler { + InitialHandler::new( + Arc::new(Config::default()), + Arc::new(PlayerCount(AtomicUsize::new(count))), + ) + } + + fn ih_with_config(config: Config) -> InitialHandler { + InitialHandler::new(Arc::new(config), Arc::new(PlayerCount(AtomicUsize::new(0)))) + } +} diff --git a/server/src/io/mod.rs b/server/src/io/mod.rs index 5a343bcfc..0700a6c0c 100644 --- a/server/src/io/mod.rs +++ b/server/src/io/mod.rs @@ -1,8 +1,13 @@ +use crate::config::Config; +use crate::PlayerCount; use feather_core::network::packet::Packet; use mio_extras::channel::{channel, Receiver, Sender}; use std::net::SocketAddr; +use std::sync::Arc; use std::thread; +use uuid::Uuid; +mod initialhandler; mod listener; mod worker; @@ -14,9 +19,6 @@ pub enum ServerToWorkerMessage { NotifyPacketReceived(Box), NotifyDisconnect, Disconnect, - - EnableCompression(usize), - EnableEncryption([u8; 16]), } pub enum ServerToListenerMessage { @@ -32,6 +34,9 @@ pub enum ListenerToWorkerMessage { pub struct NewClientInfo { pub ip: SocketAddr, + pub username: String, + pub profile: Vec, + pub uuid: Uuid, pub sender: Sender, pub receiver: Receiver, @@ -45,7 +50,12 @@ pub struct NetworkIoManager { impl NetworkIoManager { /// Starts a new IO event loop with the specified number /// of worker threads. - pub fn start(addr: SocketAddr, num_worker_threads: u16) -> Self { + pub fn start( + addr: SocketAddr, + num_worker_threads: u16, + config: Arc, + player_count: Arc, + ) -> Self { info!( "Starting IO event loop on {} with {} worker threads", addr, num_worker_threads @@ -55,8 +65,10 @@ impl NetworkIoManager { for _ in 0..num_worker_threads { let (send1, recv1) = channel(); let (send2, recv2) = channel(); + let player_count = Arc::clone(&player_count); + let config = Arc::clone(&config); - thread::spawn(move || worker::start(recv1, send2)); + thread::spawn(move || worker::start(recv1, send2, config, player_count)); workers.push((send1, recv2)); } @@ -78,3 +90,9 @@ impl NetworkIoManager { info!("Shut down IO event loop"); } } + +impl Default for NetworkIoManager { + fn default() -> Self { + panic!("Nope, don't call default() on the IO manager. That won't work."); + } +} diff --git a/server/src/io/worker.rs b/server/src/io/worker.rs index 5a8a8fa26..7aaa2762a 100644 --- a/server/src/io/worker.rs +++ b/server/src/io/worker.rs @@ -1,13 +1,21 @@ -use super::*; +use std::io::Read; +use std::io::Write; +use std::sync::Arc; + use bytes::BufMut; -use feather_core::bytebuf::{BufMutAlloc, ByteBuf}; -use feather_core::network::packet::PacketDirection; -use feather_core::network::serialize::ConnectionIOManager; use hashbrown::HashMap; use mio::Event; use mio::{net::TcpStream, Events, Poll, PollOpt, Ready, Token}; -use std::io::Read; -use std::io::Write; + +use feather_core::bytebuf::{BufMutAlloc, ByteBuf}; +use feather_core::network::packet::PacketDirection; +use feather_core::network::serialize::ConnectionIOManager; + +use crate::config::Config; +use crate::io::initialhandler::{Action, InitialHandler}; +use crate::PlayerCount; + +use super::*; // The token used to listen on the channel receiving messages from the listener thread const LISTENER_TOKEN: Token = Token(0); @@ -22,6 +30,9 @@ struct Worker { clients: HashMap, pending_disconnects: Vec, + + config: Arc, + player_count: Arc, } struct ClientHandle { @@ -30,19 +41,26 @@ struct ClientHandle { write_buffer: Option, - receiver: Receiver, - sender: Sender, + receiver: Option>, + sender: Option>, stream_token: Token, server_to_worker_token: Token, manager: ConnectionIOManager, + + initial_handler: Option, } /// Starts an IO worker on the current thread, /// blocking indefinitely until a `ShutDown` message /// is received from the listener. -pub fn start(receiver: Receiver, sender: Sender) { +pub fn start( + receiver: Receiver, + sender: Sender, + config: Arc, + player_count: Arc, +) { trace!("Starting IO worker thread"); let poll = Poll::new().unwrap(); @@ -56,6 +74,8 @@ pub fn start(receiver: Receiver, sender: Sender send_packet(worker, client_id, packet), - ServerToWorkerMessage::EnableCompression(threshold) => worker - .clients - .get_mut(&client_id) - .unwrap() - .manager - .enable_compression(threshold), - ServerToWorkerMessage::EnableEncryption(key) => worker - .clients - .get_mut(&client_id) - .unwrap() - .manager - .enable_encryption(key), _ => panic!("Invalid message received from server thread"), } } @@ -212,17 +204,23 @@ fn disconnect_client(worker: &mut Worker, client_id: Client) { return; } - worker.poll.deregister(&client.receiver).unwrap(); worker.poll.deregister(&client.stream).unwrap(); - let _ = client.sender.send(ServerToWorkerMessage::NotifyDisconnect); + if let Some(sender) = client.sender.as_ref() { + worker + .poll + .deregister(client.receiver.as_ref().unwrap()) + .unwrap(); + + let _ = sender.send(ServerToWorkerMessage::NotifyDisconnect); + } debug!("Disconnecting client {}", client_id.0); worker.clients.remove(&client_id); } -fn send_packet(worker: &mut Worker, client_id: Client, packet: Box) { +fn send_packet(worker: &mut Worker, client_id: Client, packet: Box) { let client = worker.clients.get_mut(&client_id).unwrap(); let manager = &mut client.manager; @@ -274,6 +272,10 @@ fn read_from_stream(worker: &mut Worker, token: Token) -> Result<(), ()> { .take_pending_packets() { handle_packet(worker, client_id, packet); + if !worker.clients.contains_key(&client_id) { + // Client was disconnected + return Ok(()); + } } Ok(()) @@ -307,11 +309,70 @@ fn write_to_client(worker: &mut Worker, client_id: Client) -> Result<(), ()> { Ok(()) } -fn handle_packet(worker: &mut Worker, client_id: Client, packet: Box) { +fn handle_packet(worker: &mut Worker, client_id: Client, packet: Box) { let client = worker.clients.get_mut(&client_id).unwrap(); - let msg = ServerToWorkerMessage::NotifyPacketReceived(packet); - client.sender.send(msg).unwrap(); + let mut action_queue = vec![]; + + if let Some(ih) = client.initial_handler.as_mut() { + // Forward packet to the initial handler. + ih.handle_packet(packet); + + action_queue = ih.actions_to_execute(); + } else { + // Forward packet to the server. + let msg = ServerToWorkerMessage::NotifyPacketReceived(packet); + client.sender.as_ref().unwrap().send(msg).unwrap(); + } + + for action in action_queue { + let client = worker.clients.get_mut(&client_id).unwrap(); + + match action { + Action::Disconnect => { + disconnect_client(worker, client_id); + return; + } + Action::EnableCompression(threshold) => { + client.manager.enable_compression(threshold as usize); + } + Action::JoinGame(info) => { + let (send1, recv1) = channel(); + let (send2, recv2) = channel(); + + let player_info = NewClientInfo { + ip: client.addr.clone(), + username: info.username, + profile: info.props, + uuid: info.uuid, + sender: send1, + receiver: recv2, + }; + + worker + .sender + .send(ListenerToWorkerMessage::NewClient(player_info)) + .unwrap(); + + client.initial_handler = None; + + client.sender = Some(send2); + client.receiver = Some(recv1); + + worker + .poll + .register( + client.receiver.as_ref().unwrap(), + client.server_to_worker_token, + Ready::readable(), + PollOpt::edge(), + ) + .unwrap(); + } + Action::EnableEncryption(key) => client.manager.enable_encryption(key), + Action::SendPacket(packet) => send_packet(worker, client_id, packet), + } + } } fn get_stream_token(client_id: Client) -> Token { diff --git a/server/src/joinhandler.rs b/server/src/joinhandler.rs new file mode 100644 index 000000000..4121836e5 --- /dev/null +++ b/server/src/joinhandler.rs @@ -0,0 +1,177 @@ +//! The join handler, in contrast to the initial handler, +//! takes over after the login sequence has completed. +//! It's responsible for asyncrhonously loading the player's +//! data (inventory, chunks, etc.) and then sending the necessary +//! packets to join the player. After completion, the component is +//! removed. + +use std::sync::atomic::Ordering; +use std::sync::Arc; + +use shrev::EventChannel; +use specs::{ + Component, Entities, Entity, HashMapStorage, Join, LazyUpdate, Read, ReadStorage, System, + Write, WriteStorage, +}; + +use feather_core::network::packet::implementation::{ + JoinGame, PlayerPositionAndLookClientbound, SpawnPosition, +}; +use feather_core::world::{BlockPosition, ChunkMap, ChunkPosition, Position}; +use feather_core::{Difficulty, Dimension, Gamemode}; + +use crate::chunkclient::ChunkWorkerHandle; +use crate::config::Config; +use crate::network::NetworkComponent; +use crate::player::ChunkPendingComponent; +use crate::PlayerCount; + +/// For now, we use a fixed spawn position. +/// In the future, the spawn position should +/// be loaded asynchronously from the world save. +pub const SPAWN_POSITION: Position = Position { + x: 0.0, + y: 64.0, + z: 0.0, + pitch: 0.0, + yaw: 0.0, +}; + +/// See `SPAWN_POSITION` +const COMPASS_SPAWN_POSITION: BlockPosition = BlockPosition { x: 0, y: 64, z: 0 }; + +pub struct JoinHandlerComponent { + stage: Stage, +} + +impl JoinHandlerComponent { + pub fn new() -> Self { + Self { + stage: Stage::Initial, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum Stage { + Initial, + AwaitChunkSends, +} + +impl Component for JoinHandlerComponent { + type Storage = HashMapStorage; +} + +/// Event which is triggered when a player +/// completes the join process (i.e. when +/// all chunks have been sent). +pub struct PlayerJoinEvent { + pub player: Entity, +} + +/// System for join handling. +pub struct JoinHandlerSystem; + +impl<'a> System<'a> for JoinHandlerSystem { + type SystemData = ( + WriteStorage<'a, JoinHandlerComponent>, + ReadStorage<'a, NetworkComponent>, + ReadStorage<'a, ChunkPendingComponent>, + Write<'a, EventChannel>, + Read<'a, ChunkWorkerHandle>, + Entities<'a>, + Read<'a, LazyUpdate>, + Read<'a, Arc>, + Read<'a, Arc>, + Read<'a, ChunkMap>, + ); + + fn run(&mut self, data: Self::SystemData) { + let ( + mut joincomps, + netcomps, + pending_chunks, + mut join_events, + worker_handle, + entities, + lazy, + config, + player_count, + chunk_map, + ) = data; + + let mut to_remove = vec![]; + + for (player, net, join_handler, pending_chunks) in + (&entities, &netcomps, &mut joincomps, &pending_chunks).join() + { + match join_handler.stage { + Stage::Initial => { + // Send Join Game, then queue chunks for loading + sending. + let join_game = JoinGame::new( + player.id() as i32, + Gamemode::Creative.get_id(), + Dimension::Overwold.get_id(), + Difficulty::Medium.get_id(), + 0, // Max players - not used + "default".to_string(), // Level type + false, // Reduced debug info + ); + crate::network::send_packet_to_player(net, join_game); + + // Queue chunks + let view_distance = config.server.view_distance as i32; + for x in -view_distance..=view_distance { + for y in -view_distance..=view_distance { + let pos = ChunkPosition::new(x, y); + crate::player::send_chunk_to_player( + pos, + net, + player, + &chunk_map, + &worker_handle, + &lazy, + ); + } + } + + // Increment player count + player_count.0.fetch_add(1, Ordering::SeqCst); + + join_handler.stage = Stage::AwaitChunkSends; + } + Stage::AwaitChunkSends => { + // If 0 chunks have yet to be sent, join the player by sending spawn position. + // See https://wiki.vg/Protocol_FAQ + if pending_chunks.len() != 0 { + continue; + } + let spawn_position = SpawnPosition::new(COMPASS_SPAWN_POSITION); + crate::network::send_packet_to_player(net, spawn_position); + + let position_and_look = PlayerPositionAndLookClientbound::new( + SPAWN_POSITION.x, + SPAWN_POSITION.y, + SPAWN_POSITION.z, + SPAWN_POSITION.yaw, + SPAWN_POSITION.pitch, + 0, // Flags - unused by us + 0, // Teleport ID - unused by us + ); + crate::network::send_packet_to_player(net, position_and_look); + + // Trigger event + let event = PlayerJoinEvent { player }; + join_events.single_write(event); + + // We're finished here. + to_remove.push(player); + } + } + } + + to_remove.into_iter().for_each(|player| { + joincomps.remove(player); + }); + } +} diff --git a/server/src/main.rs b/server/src/main.rs index df1ab9406..1d0ac91d7 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -6,174 +6,159 @@ extern crate serde_derive; extern crate serde_json; use std::alloc::System; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use specs::{Dispatcher, DispatcherBuilder, Entity, LazyUpdate, World, WorldExt}; + +use feather_core::network::packet::implementation::DisconnectPlay; +use prelude::*; + +use crate::entity::EntityComponent; +use crate::network::send_packet_to_player; +use crate::player::PlayerDisconnectEvent; +use shrev::EventChannel; #[global_allocator] static ALLOC: System = System; +pub mod chunkclient; pub mod chunkworker; pub mod config; -pub mod genindex; -pub mod initialhandler; +pub mod entity; pub mod io; +pub mod joinhandler; pub mod network; +pub mod player; pub mod prelude; - -use crate::genindex::{GenerationalArray, GenerationalIndex, GenerationalIndexAllocator}; -use crate::initialhandler::InitialHandlerComponent; -use crate::io::NetworkIoManager; -use crate::network::NetworkComponent; -use multimap::MultiMap; -use prelude::*; -use std::thread::sleep; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +#[cfg(test)] +pub mod testframework; +pub mod worldupdate; pub const TPS: u64 = 20; pub const PROTOCOL_VERSION: u32 = 404; pub const SERVER_VERSION: &'static str = "Feather 1.13.2"; +pub const TICK_TIME: u64 = 1000 / TPS; -type EntityMap = GenerationalArray; -type Entity = GenerationalIndex; - -pub struct State { - pub chunk_map: ChunkMap, - pub config: Config, - - pub allocator: GenerationalIndexAllocator, - pub io_manager: NetworkIoManager, - - pub network_components: EntityMap, - pub ih_components: EntityMap, - pub entity_components: EntityMap, - pub player_components: EntityMap, - - /// Difference between players and joined_players: - /// `players` includes all clients connected - /// to the server, including those in the login - /// or status phase, while `joined_players` only - /// includes those who have joined the server (whose - /// state is Play). - pub players: Vec, - pub joined_players: Vec, - - /// Each entity which requires its chunks and chunks - /// around it to be loaded can insert an entry into - /// this map. For example, a player will have - /// an entry for every chunk within the view distance. - /// When a loaded chunk has no more holders, it will - /// be queued for unloading. - pub chunk_holders: MultiMap, - - pub chunk_worker_handle: ChunkWorkerHandle, - - pub running: bool, - pub tick_count: u64, -} - -/// A handle for interacting -/// with the chunk worker IO thread. -pub struct ChunkWorkerHandle { - /// Channel used to send chunk requests - sender: crossbeam::channel::Sender, - /// Channel used to receive replies from the worker thread - receiver: crossbeam::channel::Receiver, -} - -pub struct PlayerComponent { - pub profile_properties: Vec, - pub gamemode: Gamemode, -} - -pub struct EntityComponent { - pub uuid: Uuid, - pub display_name: String, - pub position: Position, - pub on_ground: bool, -} +#[derive(Default, Debug)] +pub struct PlayerCount(AtomicUsize); +#[derive(Default, Debug)] +pub struct TickCount(u64); fn main() { - let config = config::load_from_file("feather.toml") - .expect("Failed to load configuration. Please ensure that the file feather.toml exists and is correct."); + let config = Arc::new(config::load_from_file("feather.toml") + .expect("Failed to load configuration. Please ensure that the file feather.toml exists and is correct.")); init_log(&config); info!("Starting Feather; please wait..."); - let io_manager = io::NetworkIoManager::start( - format!("127.0.0.1:{}", config.server.port).parse().unwrap(), - config.io.io_worker_threads, - ); + let player_count = Arc::new(PlayerCount(AtomicUsize::new(0))); - let mut state = init_state(config, io_manager); + let io_manager = init_io_manager(Arc::clone(&config), Arc::clone(&player_count)); - info!("Initialized server state"); + let (mut world, mut dispatcher) = init_world(config, player_count, io_manager); - run_loop(&mut state); + info!("Initialized world"); - state.io_manager.stop(); - state - .chunk_worker_handle - .sender - .send(chunkworker::Request::ShutDown) - .unwrap(); + info!("Server started"); + run_loop(&mut world, &mut dispatcher); } -fn run_loop(state: &mut State) { - while state.running { - // For optimal latency, the chunk worker system - // should run before all other systems - chunk_worker_system(state); - network::network_system(state); +/// Runs the server loop, blocking until the server +/// is shut down. +fn run_loop(world: &mut World, dispatcher: &mut Dispatcher) { + loop { + let start_time = current_time_in_millis(); - state.tick_count += 1; - sleep(Duration::from_millis(1000 / 20)); // TODO - proper game loop - } -} + dispatcher.dispatch(world); + world.maintain(); + + // Increment tick count + let mut tick_count = world.write_resource::(); + tick_count.0 += 1; -/// System for emptying queue of loaded chunks. -fn chunk_worker_system(state: &mut State) { - // Receive all replies from chunk worker thread and load into world - while let Ok((pos, result)) = state.chunk_worker_handle.receiver.try_recv() { - match result { - Ok(chunk) => { - state.chunk_map.inner_mut().insert(pos, chunk); - debug!("Loaded chunk at {:?}", pos); - } - Err(e) => { - // TODO generate new chunk if it wasn't found - warn!("Error occurred while loading chunk at {:?}: {:?}", pos, e); - } + // Sleep correct amount + let end_time = current_time_in_millis(); + let elapsed = end_time - start_time; + if elapsed > TICK_TIME { + continue; // Behind - start next tick immediately + } + + // Sleep in 1ms increments until we've slept enough + let mut sleep_time = (TICK_TIME - elapsed) as i64; + let mut last_sleep_time = current_time_in_millis(); + while sleep_time > 0 { + std::thread::sleep(Duration::from_millis(1)); + sleep_time -= (current_time_in_millis() - last_sleep_time) as i64; + last_sleep_time = current_time_in_millis(); } } } -fn init_state(config: Config, io_manager: NetworkIoManager) -> State { - info!("Starting chunk worker thread"); - let (tx, rx) = chunkworker::start("world"); - State { - chunk_map: ChunkMap::new(), +/// Starts the IO threads. +fn init_io_manager(config: Arc, player_count: Arc) -> io::NetworkIoManager { + let ioman = io::NetworkIoManager::start( + format!("127.0.0.1:{}", config.server.port).parse().unwrap(), + config.io.io_worker_threads, config, + player_count, + ); + ioman +} - allocator: GenerationalIndexAllocator::new(), - io_manager, - - network_components: EntityMap::new(), - ih_components: EntityMap::new(), - entity_components: EntityMap::new(), - player_components: EntityMap::new(), - - players: vec![], - joined_players: vec![], - - chunk_holders: MultiMap::new(), - - chunk_worker_handle: ChunkWorkerHandle { - sender: tx, - receiver: rx, - }, - - running: true, - tick_count: 0, - } +/// Initializes the Specs world. +fn init_world<'a, 'b>( + config: Arc, + player_count: Arc, + ioman: io::NetworkIoManager, +) -> (World, Dispatcher<'a, 'b>) { + let mut world = World::new(); + world.insert(config); + world.insert(player_count); + world.insert(ioman); + world.insert(TickCount::default()); + + let mut dispatcher = DispatcherBuilder::new() + .with(chunkclient::ChunkLoadSystem, "chunk_load", &[]) + .with(network::NetworkSystem, "network", &[]) + .with( + worldupdate::PlayerDiggingSystem, + "player_digging", + &["network"], + ) + .with( + player::PlayerMovementSystem, + "player_movement", + &["network"], + ) + .with( + player::ChunkSendSystem::new(), + "chunk_send", + &["chunk_load"], + ) + .with( + joinhandler::JoinHandlerSystem, + "join_handler", + &["chunk_send"], + ) + .with(player::PlayerInitSystem::new(), "player_init", &["network"]) + .with( + player::JoinBroadcastSystem::new(), + "join_broadcast", + &["join_handler", "player_init"], + ) + .with( + player::DisconnectBroadcastSystem::new(), + "disconnect_broadcast", + &[], + ) + .build(); + + dispatcher.setup(&mut world); + + (world, dispatcher) } fn init_log(config: &Config) { @@ -189,53 +174,77 @@ fn init_log(config: &Config) { simple_logger::init_with_level(level).unwrap(); } -pub fn add_entity(state: &mut State) -> Entity { - let e = state.allocator.allocate(); - e +/// Retrieves the current time in seconds +/// since the UNIX epoch. +pub fn current_time_in_secs() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() } -pub fn add_player(state: &mut State) -> Entity { - let e = add_entity(state); - state.players.push(e); - e +/// Retrieves the current time in milleseconds +/// since the UNIX epoch. +pub fn current_time_in_millis() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as u64 } -pub fn remove_entity(state: &mut State, entity: Entity) { - state.allocator.deallocate(entity); -} +/// Disconnects the given player, removing them from the world. +/// This operation is performed lazily. +pub fn disconnect_player(player: Entity, reason: String, lazy: &LazyUpdate) { + lazy.exec_mut(move |world| { + let json = json!({ + "text": reason, + }); -pub fn remove_player(state: &mut State, entity: Entity) { - // If player was joined, broadcast disconnect - if state.joined_players.contains(&entity) { - network::broadcast_player_leave(state, entity); - } + let packet = DisconnectPlay::new(json.to_string()); + send_packet_to_player(world.read_component().get(player).unwrap(), packet); - network::handle_player_remove(state, entity); + disconnect_player_without_packet(player, world, reason); + }) +} - remove_entity(state, entity); +/// Disconnects a player without sending Disconnect Play. +/// This should be used when the client disconnects. +pub fn disconnect_player_without_packet(player: Entity, world: &mut World, reason: String) { + { + let ecomps = world.write_component::(); + let ecomp = ecomps.get(player).unwrap(); + + info!("Disconnecting player {}: {}", ecomp.display_name, reason); + + // Decrement player count + let player_count = world.fetch_mut::>(); + player_count.0.fetch_sub(1, Ordering::SeqCst); + + // Trigger disconnect event + let event = PlayerDisconnectEvent { + player, + uuid: ecomp.uuid.clone(), + reason, + }; + world + .fetch_mut::>() + .single_write(event); + } - state.joined_players.retain(|e| *e != entity); - state.players.retain(|e| *e != entity); + world.delete_entity(player).unwrap(); } -/// Asynchronously loads the chunk at the given position. -/// At some point in time after this function is called, -/// the chunk will appear in the chunk map. -/// -/// In the event that the requested chunk does not exist -/// in the world save, it will be generated asynchronously. -pub fn load_chunk(state: &mut State, pos: ChunkPosition) { - // Send request to chunk worker thread - state - .chunk_worker_handle - .sender - .send(chunkworker::Request::LoadChunk(pos)) - .unwrap(); -} +#[cfg(test)] +mod tests { + use super::*; -pub fn current_time_in_secs() -> u64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_secs() + #[test] + fn test_init_world() { + let config = Arc::new(Config::default()); + let player_count = Arc::new(PlayerCount(AtomicUsize::new(0))); + let ioman = init_io_manager(Arc::clone(&config), Arc::clone(&player_count)); + + let (world, mut dispatcher) = init_world(config, player_count, ioman); + dispatcher.dispatch(&world); + } } diff --git a/server/src/network.rs b/server/src/network.rs index 50f3fb94a..19c378b19 100644 --- a/server/src/network.rs +++ b/server/src/network.rs @@ -1,15 +1,83 @@ -use crate::initialhandler::InitialHandlerComponent; -use crate::io::{ServerToListenerMessage, ServerToWorkerMessage}; -use crate::prelude::*; -use crate::{add_player, initialhandler as ih, remove_player, Entity, State}; -use feather_blocks::Block; -use feather_core::entitymeta::{EntityMetadata, MetaEntry}; -use feather_core::network::packet::{implementation::*, Packet, PacketType}; +use std::sync::Mutex; + use mio_extras::channel::{Receiver, Sender}; +use shrev::EventChannel; +use specs::{ + Component, DenseVecStorage, Entities, Entity, Join, LazyUpdate, Read, ReadStorage, System, + WorldExt, Write, WriteStorage, +}; + +use feather_core::network::packet::{implementation::*, Packet, PacketType}; + +use crate::entity::PlayerComponent; +use crate::io::{NetworkIoManager, ServerToListenerMessage, ServerToWorkerMessage}; +use crate::joinhandler::JoinHandlerComponent; +use crate::prelude::*; +use crate::{disconnect_player_without_packet, TickCount}; //const MAX_KEEP_ALIVE_TIME: u64 = 30; //const HEAD_OFFSET: f64 = 1.62; // Offset from feet pos to head pos +/// A component which contains the received packets +/// for this tick. +pub struct PacketQueue { + queue: Mutex)>>>, +} + +impl PacketQueue { + /// Returns the packets queued for handling + /// of the given type, draining the queue of this + /// type of packet. + pub fn for_packet(&self, ty: PacketType) -> Vec<(Entity, Box)> { + let mut queue = self.queue.lock().unwrap(); + + let ordinal = ty.ordinal(); + if ordinal >= queue.len() { + self.expand(&mut queue, ordinal); + } + + let mut result = vec![]; + std::mem::swap(&mut result, queue.get_mut(ordinal).unwrap()); + + result + } + + /// Expands the internal vector to allow for additional packet types. + fn expand( + &self, + queue: &mut std::sync::MutexGuard)>>>, + to: usize, + ) { + if to < queue.len() { + return; + } + + for _ in queue.len()..(to + 1) { + queue.push(Vec::new()); + } + } + + /// Adds a packet to the queue. + pub fn add_for_packet(&self, player: Entity, packet: Box) { + let mut queue = self.queue.lock().unwrap(); + + let ordinal = packet.ty().ordinal(); + if ordinal >= queue.len() { + self.expand(&mut queue, ordinal); + } + + queue[ordinal].push((player, packet)); + } +} + +impl Default for PacketQueue { + fn default() -> Self { + Self { + queue: Mutex::new(vec![]), + } + } +} + pub struct NetworkComponent { sender: Sender, receiver: Receiver, @@ -33,426 +101,146 @@ impl NetworkComponent { } } -pub fn network_system(state: &mut State) { - handle_connections(state); - - send_keep_alives(state); - - poll_for_new_players(state); -} - -fn handle_connections(state: &mut State) { - let mut players_to_remove = vec![]; +impl Component for NetworkComponent { + type Storage = DenseVecStorage; +} + +/// The network system, responsible for +/// receiving and buffering packets received +/// from players. Received packets +/// are added to a queue (`PacketQueue`) so that +/// other systems can handle them. +pub struct NetworkSystem; + +/// Event which is triggered when a player joins +/// but before the join handler is completed. +pub struct PlayerPreJoinEvent { + pub player: Entity, + pub username: String, + pub uuid: Uuid, + pub profile_properties: Vec, +} + +impl<'a> System<'a> for NetworkSystem { + type SystemData = ( + WriteStorage<'a, NetworkComponent>, + ReadStorage<'a, PlayerComponent>, + Write<'a, EventChannel>, + Write<'a, PacketQueue>, + Read<'a, NetworkIoManager>, + Entities<'a>, + Read<'a, TickCount>, + Read<'a, LazyUpdate>, + ); - for player in state.players.clone() { - while let Ok(msg) = state.network_components[player].receiver.try_recv() { + fn run(&mut self, data: Self::SystemData) { + let ( + mut netcomps, + pcomps, + mut join_events, + packet_queue, + ioman, + entities, + tick_count, + lazy, + ) = data; + // Poll for new connections + while let Ok(msg) = ioman.receiver.try_recv() { match msg { - ServerToWorkerMessage::NotifyPacketReceived(packet) => { - if let Some(ih) = state.ih_components.get(player) { - // Skip if initial handler is awaiting chunks - if ih.stage != ih::Stage::AwaitChunkLoad { - if let Err(e) = ih::handle_packet(state, player, packet) { - info!("Disconnecting player: {}", e); - ih::disconnect_login(state, player, &e.to_string()); - remove_player(state, player); - } - } - } else { - handle_player_packet(state, player, packet); - } + ServerToListenerMessage::NewClient(info) => { + // New connection - handle it + info!("Accepting connection from {}", info.ip); + let netcomp = NetworkComponent::new(info.sender, info.receiver); + + // Create entity + let new_entity = entities.create(); + netcomps.insert(new_entity, netcomp).unwrap(); + + // Create join handler + let join_handler = JoinHandlerComponent::new(); + lazy.exec_mut(move |world| { + world + .write_component::() + .insert(new_entity, join_handler) + .unwrap(); + }); + + // Queue event + let event = PlayerPreJoinEvent { + player: new_entity, + username: info.username.clone(), + uuid: info.uuid.clone(), + profile_properties: info.profile.clone(), + }; + join_events.single_write(event); } - ServerToWorkerMessage::NotifyDisconnect => { - players_to_remove.push(player); - } - _ => panic!("Invalid message received from worker thread"), + _ => panic!("Network system received invalid message from IO listener"), } } - // Send all pending chunks which have been loaded - let mut to_remove = vec![]; - for (i, chunk_pos) in state.network_components[player] - .chunks_to_send - .clone() - .iter() - .enumerate() - { - if let Some(chunk) = state.chunk_map.chunk_at(*chunk_pos) { - let packet = ChunkData::new(chunk.clone()); - send_packet_to_player(state, player, packet); - to_remove.push(i); - } else { + // Receive packets + disconnects from players + for (player, netcomp) in (&entities, &netcomps).join() { + while let Ok(msg) = netcomp.receiver.try_recv() { + match msg { + ServerToWorkerMessage::NotifyPacketReceived(packet) => { + packet_queue.add_for_packet(player, packet); + } + ServerToWorkerMessage::NotifyDisconnect => { + // TODO broadcast disconnect + lazy.exec_mut(move |world| { + disconnect_player_without_packet( + player, + world, + "Client disconnected".to_string(), + ) + }); + } + _ => panic!("Network system received invalid message from IO worker}"), + } } } - let mut count = 0; - for i in to_remove { - state.network_components[player] - .chunks_to_send - .remove(i - count); - count += 1; - } - // If the player has yet to receive their spawn position, - // send it if all their chunks have been loaded - if state.network_components[player].chunks_to_send.is_empty() - && state.ih_components.get(player).is_some() - && state.ih_components[player].stage == ih::Stage::AwaitChunkLoad - { - ih::complete_join_game(state, player); - } - } - - for _player in players_to_remove { - remove_player(state, _player); - } -} - -fn handle_player_packet(state: &mut State, player: Entity, packet: Box) { - match packet.ty() { - PacketType::PlayerPositionAndLookServerbound => handle_player_pos_and_look( - state, - player, - cast_packet::(&packet), - ), - PacketType::PlayerPosition => { - handle_player_pos(state, player, cast_packet::(&packet)); - } - PacketType::PlayerLook => { - handle_player_look(state, player, cast_packet::(&packet)); - } - PacketType::PlayerDigging => { - handle_player_digging(state, player, cast_packet::(&packet)); + // Send keepalives every second. The dependency on the player + // component is required because keepalives should + // only be sent to players who have joined (completed + // the login process). + // TODO check that player hasn't timed out + if tick_count.0 % TPS == 0 { + for (netcomp, _) in (&netcomps, &pcomps).join() { + send_packet_to_player(netcomp, KeepAliveClientbound::new(0)); + } } - _ => (), // TODO } } -// TODO proper validation of new position - -fn handle_player_pos_and_look( - state: &mut State, - player: Entity, - packet: &PlayerPositionAndLookServerbound, +/// Sends a packet to all (joined) players on the server, excluding +/// `neq`, if it exists. +pub fn send_packet_to_all_players( + net_comps: &ReadStorage, + player_comps: &ReadStorage, + entities: &Entities, + packet: P, + neq: Option, ) { - let ecomp = &state.entity_components[player]; - let old_pos = ecomp.position; - - let new_pos = Position::new(packet.x, packet.feet_y, packet.z, packet.pitch, packet.yaw); - - broadcast_entity_movement(state, player, old_pos, new_pos, true, true); - - state.entity_components[player].position = new_pos; -} - -fn handle_player_pos(state: &mut State, player: Entity, packet: &PlayerPosition) { - let ecomp = &state.entity_components[player]; - let old_pos = ecomp.position; - - let new_pos = Position::new( - packet.x, - packet.feet_y, - packet.z, - old_pos.pitch, - old_pos.yaw, - ); - - broadcast_entity_movement(state, player, old_pos, new_pos, true, false); - - state.entity_components[player].position = new_pos; -} - -fn handle_player_look(state: &mut State, player: Entity, packet: &PlayerLook) { - let ecomp = &state.entity_components[player]; - let old_pos = ecomp.position; - - let new_pos = Position::new(old_pos.x, old_pos.y, old_pos.z, packet.pitch, packet.yaw); - - broadcast_entity_movement(state, player, old_pos, new_pos, false, true); - - state.entity_components[player].position = new_pos; -} - -fn handle_player_digging(state: &mut State, player: Entity, packet: &PlayerDigging) { - match packet.status { - PlayerDiggingStatus::FinishedDigging => { - if state - .chunk_map - .set_block_at(packet.location, Block::Air) - .is_err() - { - // TODO kick player - } - broadcast_block_update(state, packet.location); - } - PlayerDiggingStatus::StartedDigging => { - let pcomp = &state.player_components[player]; - if pcomp.gamemode == Gamemode::Creative { - // Break block instantly - TODO not with sword in hand - if state - .chunk_map - .set_block_at(packet.location, Block::Air) - .is_err() - { - // TODO kick player - warn!("Client sent invalid Player Digging packet"); - return; - } - broadcast_block_update(state, packet.location); + for (entity, net, _) in (entities, net_comps, player_comps).join() { + if let Some(e) = neq.as_ref() { + if *e == entity { + continue; // Exclude this entity } } - _ => (), // TODO - } -} - -fn send_keep_alives(state: &mut State) { - if state.tick_count % TPS != 0 { - return; // Only run once per second - } - for player in state.joined_players.clone() { - let keep_alive = KeepAliveClientbound::new(0); - send_packet_to_player(state, player, keep_alive); + send_packet_to_player(net, packet.clone()); } } -fn poll_for_new_players(state: &mut State) { - while let Ok(msg) = state.io_manager.receiver.try_recv() { - match msg { - ServerToListenerMessage::NewClient(info) => { - debug!("Server registering player"); - let player = add_player(state); - let ih = InitialHandlerComponent::new(); - state.ih_components.set(player, ih); - - let netc = NetworkComponent::new(info.sender, info.receiver); - state.network_components.set(player, netc); - } - _ => panic!("Invalid message received from listener thread"), - } - } -} - -pub fn send_packet_to_player(state: &State, player: Entity, packet: P) { - let comp = &state.network_components[player]; +/// Sends a packet to the given player. +pub fn send_packet_to_player(comp: &NetworkComponent, packet: P) { let _ = comp .sender .send(ServerToWorkerMessage::SendPacket(Box::new(packet))); } -pub fn send_packet_boxed_to_player(state: &State, player: Entity, packet: Box) { - let comp = &state.network_components[player]; +/// Sends a packet to the given player. +pub fn send_packet_boxed_to_player(comp: &NetworkComponent, packet: Box) { let _ = comp.sender.send(ServerToWorkerMessage::SendPacket(packet)); } - -pub fn enable_compression_for_player(state: &State, player: Entity, threshold: usize) { - let comp = &state.network_components[player]; - let _ = comp - .sender - .send(ServerToWorkerMessage::EnableCompression(threshold)); -} - -pub fn enable_encryption_for_player(state: &State, player: Entity, key: [u8; 16]) { - let comp = &state.network_components[player]; - let _ = comp - .sender - .send(ServerToWorkerMessage::EnableEncryption(key)); -} - -pub fn handle_player_remove(state: &mut State, player: Entity) { - let comp = &state.network_components[player]; - let _ = comp.sender.send(ServerToWorkerMessage::Disconnect); -} - -/// Broadcasts to all clients that the specified player -/// has joined the game. This should be called -/// whenever a player joins. -/// -/// This function is currently called by the initial handler. -pub fn broadcast_player_join(state: &mut State, player: Entity) { - let (player_info, spawn_player) = get_player_initialization_packets(state, player); - - for p in &state.joined_players { - send_packet_to_player(state, *p, player_info.clone()); - if *p != player { - send_packet_to_player(state, *p, spawn_player.clone()); - } - } -} - -/// Returns the player info and spawn player packets -/// for the given player. -pub fn get_player_initialization_packets( - state: &State, - player: Entity, -) -> (PlayerInfo, SpawnPlayer) { - let entity_comp = state.entity_components.get(player).unwrap(); - let player_comp = state.player_components.get(player).unwrap(); - - let display_name = json!({ - "text": entity_comp.display_name - }) - .to_string(); - - let mut props = vec![]; - for prop in player_comp.profile_properties.iter() { - props.push(( - prop.name.clone(), - prop.value.clone(), - prop.signature.clone(), - )); - } - - let action = PlayerInfoAction::AddPlayer( - entity_comp.display_name.clone(), - props, - Gamemode::Creative, - 50, - display_name, - ); - let player_info = PlayerInfo::new(action, entity_comp.uuid.clone()); - - let metadata = EntityMetadata::new().with(&[ - (0, MetaEntry::Byte(0)), - (1, MetaEntry::VarInt(300)), - (2, MetaEntry::OptChat(None)), - (3, MetaEntry::Boolean(false)), - (4, MetaEntry::Boolean(false)), - (5, MetaEntry::Boolean(false)), - (6, MetaEntry::Byte(0)), - (7, MetaEntry::Float(1.0)), - (8, MetaEntry::VarInt(0)), - (9, MetaEntry::Boolean(false)), - (10, MetaEntry::VarInt(0)), - (11, MetaEntry::Float(0.0)), - (12, MetaEntry::VarInt(0)), - (13, MetaEntry::Byte(0)), - (14, MetaEntry::Byte(1)), - // TODO NBT - ]); - - let spawn_player = SpawnPlayer::new( - player.index() as i32, - entity_comp.uuid.clone(), - entity_comp.position.x, - entity_comp.position.y, - entity_comp.position.z, - degrees_to_stops(entity_comp.position.pitch), - degrees_to_stops(entity_comp.position.yaw), - metadata, - ); - - (player_info, spawn_player) -} - -/// Broadcasts to all joined players that the -/// given player has left the server. This should -/// remove the player from the tablist. -pub fn broadcast_player_leave(state: &mut State, player: Entity) { - let ecomp = &state.entity_components[player]; - - let player_info = PlayerInfo::new(PlayerInfoAction::RemovePlayer, ecomp.uuid.clone()); - send_packet_to_all_players(state, player_info, Some(player)); - - let destroy_entities = DestroyEntities::new(vec![player.index() as i32]); - send_packet_to_all_players(state, destroy_entities, Some(player)); -} - -/// Notifies all players within range -/// that an entity has moved. This -/// entity can be a player. -/// -/// The `has_moved` and `has_moved` fields indicate -/// whether the entity has moved its position -/// or changed its pitch/yaw. These values -/// are used to determine which packet to send: -/// for example, if an entity has only moved and not looked, -/// an Entity Relative Move packet is sent -/// rather than an Entity Look and Relative Move. -/// -/// In the event that the entity has moved -/// more than 8 blocks, an Entity Teleport packet -/// is sent instead. -pub fn broadcast_entity_movement( - state: &mut State, - entity: Entity, - old_pos: Position, - new_pos: Position, - has_moved: bool, - has_looked: bool, -) { - let ecomp = &state.entity_components[entity]; - - let dist = new_pos.distance(old_pos).abs(); - - if dist <= 8.0 { - if has_moved && has_looked { - // Entity Look and Relative Move - let (dx, dy, dz) = calculate_relative_move(old_pos, new_pos); - let packet = EntityLookAndRelativeMove::new( - entity.index() as i32, - dx, - dy, - dz, - degrees_to_stops(new_pos.yaw), - degrees_to_stops(new_pos.pitch), - ecomp.on_ground, - ); - send_packet_to_all_players(state, packet, Some(entity)); - } else if has_moved { - // Entity Relative Move - let (dx, dy, dz) = calculate_relative_move(old_pos, new_pos); - let packet = - EntityRelativeMove::new(entity.index() as i32, dx, dy, dz, ecomp.on_ground); - send_packet_to_all_players(state, packet, Some(entity)); - } else if has_looked { - // Entity Look - let packet = EntityLook::new( - entity.index() as i32, - degrees_to_stops(new_pos.yaw), - degrees_to_stops(new_pos.pitch), - ecomp.on_ground, - ); - send_packet_to_all_players(state, packet, Some(entity)); - } - } else { - // TODO - } - - // Send Entity Head Look for head yaw - if has_looked { - let packet = EntityHeadLook::new(entity.index() as i32, degrees_to_stops(new_pos.yaw)); - send_packet_to_all_players(state, packet, Some(entity)); - } -} - -pub fn broadcast_block_update(state: &mut State, pos: BlockPosition) { - // TODO only send for players in range - let block = state.chunk_map.block_at(pos).unwrap(); - let packet = BlockChange::new(pos, block.block_state_id() as i32); - - send_packet_to_all_players(state, packet, None); -} - -fn send_packet_to_all_players( - state: &State, - packet: P, - neq: Option, -) { - for player in &state.joined_players { - if neq.is_none() || *player != neq.unwrap() { - send_packet_to_player(state, *player, packet.clone()); - } - } -} - -pub fn degrees_to_stops(degs: f32) -> u8 { - ((degs / 360.0) * 256.) as u8 -} - -/// Calculates the relative move fields -/// as used in the Entity Relative Move packets. -pub fn calculate_relative_move(old: Position, current: Position) -> (i16, i16, i16) { - let x = ((current.x * 32.0 - old.x * 32.0) * 128.0) as i16; - let y = ((current.y * 32.0 - old.y * 32.0) * 128.0) as i16; - let z = ((current.z * 32.0 - old.z * 32.0) * 128.0) as i16; - (x, y, z) -} diff --git a/server/src/player.rs b/server/src/player.rs new file mode 100644 index 000000000..4c010644c --- /dev/null +++ b/server/src/player.rs @@ -0,0 +1,479 @@ +//! This module provides systems and components +//! relating to players, including player movement +//! and inventory handling. + +use std::ops::{Deref, DerefMut}; + +use hashbrown::HashSet; +use rayon::prelude::*; +use shrev::EventChannel; +use specs::storage::BTreeStorage; +use specs::{ + Component, Entities, Entity, Join, LazyUpdate, ParJoin, Read, ReadStorage, ReaderId, System, + World, WorldExt, WriteStorage, +}; + +use feather_core::network::cast_packet; +use feather_core::network::packet::implementation::{ + ChunkData, DestroyEntities, PlayerInfo, PlayerInfoAction, PlayerLook, PlayerPosition, + PlayerPositionAndLookServerbound, SpawnPlayer, +}; +use feather_core::network::packet::{Packet, PacketType}; +use feather_core::world::chunk::Chunk; +use feather_core::world::{ChunkMap, ChunkPosition, Position}; +use feather_core::Gamemode; + +use crate::chunkclient::{load_chunk, ChunkLoadEvent, ChunkWorkerHandle}; +use crate::entity::{broadcast_entity_movement, EntityComponent, PlayerComponent}; +use crate::joinhandler::{PlayerJoinEvent, SPAWN_POSITION}; +use crate::network::{send_packet_to_player, NetworkComponent, PacketQueue, PlayerPreJoinEvent}; +use feather_core::entitymeta::{EntityMetadata, MetaEntry}; +use uuid::Uuid; + +/// Event which is called when a player disconnected. +pub struct PlayerDisconnectEvent { + pub player: Entity, + pub reason: String, + pub uuid: Uuid, +} + +/// System for handling player movement +/// packets. +pub struct PlayerMovementSystem; + +impl<'a> System<'a> for PlayerMovementSystem { + type SystemData = ( + WriteStorage<'a, EntityComponent>, + ReadStorage<'a, PlayerComponent>, + Read<'a, PacketQueue>, + ReadStorage<'a, NetworkComponent>, + Entities<'a>, + Read<'a, LazyUpdate>, + ); + + fn run(&mut self, data: Self::SystemData) { + let (mut ecomps, pcomps, packet_queue, netcomps, entities, _) = data; + + // Take movement packets + let mut packets = vec![]; + packets.append(&mut packet_queue.for_packet(PacketType::PlayerPosition)); + packets.append(&mut packet_queue.for_packet(PacketType::PlayerPositionAndLookServerbound)); + packets.append(&mut packet_queue.for_packet(PacketType::PlayerLook)); + + // Handle movement packets + for (player, packet) in packets { + let ecomp = ecomps.get(player).unwrap(); + + // Get position using packet and old position + let (new_pos, has_moved, has_looked) = new_pos_from_packet(ecomp.position, packet); + + // Broadcast position update + broadcast_entity_movement( + player, + ecomp.position, + new_pos, + has_moved, + has_looked, + &netcomps, + &pcomps, + &entities, + ); + + // Set new position + ecomps.get_mut(player).unwrap().position = new_pos; + } + } +} + +fn new_pos_from_packet(old_pos: Position, packet: Box) -> (Position, bool, bool) { + let mut has_looked = false; + let mut has_moved = false; + + let pos = match packet.ty() { + PacketType::PlayerPosition => { + has_moved = true; + let packet = cast_packet::(&packet); + + Position::new( + packet.x, + packet.feet_y, + packet.z, + old_pos.pitch, + old_pos.yaw, + ) + } + PacketType::PlayerLook => { + has_looked = true; + let packet = cast_packet::(&packet); + + Position::new(old_pos.x, old_pos.y, old_pos.z, packet.pitch, packet.yaw) + } + PacketType::PlayerPositionAndLookServerbound => { + has_moved = true; + has_looked = true; + let packet = cast_packet::(&packet); + + Position::new(packet.x, packet.feet_y, packet.z, packet.pitch, packet.yaw) + } + _ => panic!(), + }; + + (pos, has_moved, has_looked) +} + +/// Component storing what chunks are pending +/// to send to a player. +#[derive(Clone, Debug)] +pub struct ChunkPendingComponent { + pub pending: HashSet, +} + +impl Deref for ChunkPendingComponent { + type Target = HashSet; + + fn deref(&self) -> &Self::Target { + &self.pending + } +} + +impl DerefMut for ChunkPendingComponent { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.pending + } +} + +impl Component for ChunkPendingComponent { + type Storage = BTreeStorage; +} + +/// System for initializing the necessary components +/// when a player joins. +pub struct PlayerInitSystem { + join_event_reader: Option>, +} + +impl PlayerInitSystem { + pub fn new() -> Self { + Self { + join_event_reader: None, + } + } +} + +impl<'a> System<'a> for PlayerInitSystem { + type SystemData = ( + Read<'a, EventChannel>, + WriteStorage<'a, PlayerComponent>, + WriteStorage<'a, EntityComponent>, + WriteStorage<'a, ChunkPendingComponent>, + ); + + fn run(&mut self, data: Self::SystemData) { + let (events, mut player_comps, mut entity_comps, mut chunk_pending_comps) = data; + + // Run through events + for event in events.read(&mut self.join_event_reader.as_mut().unwrap()) { + let player_comp = PlayerComponent { + profile_properties: event.profile_properties.clone(), + gamemode: Gamemode::Creative, + }; + player_comps.insert(event.player, player_comp).unwrap(); + + let entity_comp = EntityComponent { + uuid: event.uuid.clone(), + display_name: event.username.clone(), + position: SPAWN_POSITION, + on_ground: true, + }; + entity_comps.insert(event.player, entity_comp).unwrap(); + + let chunk_pending_comp = ChunkPendingComponent { + pending: HashSet::new(), + }; + chunk_pending_comps + .insert(event.player, chunk_pending_comp) + .unwrap(); + } + } + + fn setup(&mut self, world: &mut World) { + use specs::SystemData; + Self::SystemData::setup(world); + + self.join_event_reader = Some( + world + .fetch_mut::>() + .register_reader(), + ); + } +} + +/// System for sending chunks to players once they're loaded. +/// +/// This system listens to `ChunkLoadEvent`s. +pub struct ChunkSendSystem { + load_event_reader: Option>, +} + +impl ChunkSendSystem { + pub fn new() -> Self { + Self { + load_event_reader: None, + } + } +} + +impl<'a> System<'a> for ChunkSendSystem { + type SystemData = ( + WriteStorage<'a, ChunkPendingComponent>, + ReadStorage<'a, NetworkComponent>, + Read<'a, ChunkMap>, + Read<'a, EventChannel>, + ); + + fn run(&mut self, data: Self::SystemData) { + let (mut pendings, netcomps, chunk_map, load_events) = data; + + for event in load_events.read(&mut self.load_event_reader.as_mut().unwrap()) { + // TODO perhaps this is slightly inefficient? + (&netcomps, &mut pendings) + .par_join() + .for_each(|(net, pending)| { + if pending.contains(&event.pos) { + // It's safe to unwrap the chunk value now, + // because we know it's been loaded. + let chunk = chunk_map.chunk_at(event.pos).unwrap(); + send_chunk_data(chunk, net); + + pending.remove(&event.pos); + } + }); + } + } + + fn setup(&mut self, world: &mut World) { + use specs::SystemData; + Self::SystemData::setup(world); + self.load_event_reader = Some( + world + .fetch_mut::>() + .register_reader(), + ); + } +} + +/// System for broadcasting when a player joins +/// the game. Also spawns other players to +/// the player's client. +pub struct JoinBroadcastSystem { + reader: Option>, +} + +impl JoinBroadcastSystem { + pub fn new() -> Self { + Self { reader: None } + } +} + +impl<'a> System<'a> for JoinBroadcastSystem { + type SystemData = ( + Read<'a, EventChannel>, + ReadStorage<'a, EntityComponent>, + ReadStorage<'a, PlayerComponent>, + ReadStorage<'a, NetworkComponent>, + Entities<'a>, + ); + + fn run(&mut self, data: Self::SystemData) { + let (join_events, entity_comps, player_comps, net_comps, entities) = data; + + for event in join_events.read(&mut self.reader.as_mut().unwrap()) { + // Broadcast join + let entity_comp = entity_comps.get(event.player).unwrap(); + let player_comp = player_comps.get(event.player).unwrap(); + + let (player_info, spawn_player) = + get_player_initialization_packets(entity_comp, player_comp, event.player); + + for (player, net) in (&entities, &net_comps).join() { + // Send player info to the player who joined + // so they can see themselves in the tablist, + // but don't send spawn player. + send_packet_to_player(net, player_info.clone()); + if player != event.player { + send_packet_to_player(net, spawn_player.clone()); + } + } + + let net_comp = net_comps.get(event.player).unwrap(); + + // Send existing players to new player + for (entity_comp, player_comp, entity) in + (&entity_comps, &player_comps, &entities).join() + { + if entity != event.player { + let (player_info, spawn_player) = + get_player_initialization_packets(entity_comp, player_comp, entity); + send_packet_to_player(net_comp, player_info); + send_packet_to_player(net_comp, spawn_player); + } + } + } + } + + fn setup(&mut self, world: &mut World) { + use specs::SystemData; + Self::SystemData::setup(world); + + self.reader = Some( + world + .fetch_mut::>() + .register_reader(), + ); + } +} + +/// Returns the player info and spawn player packets +/// for the given player. +fn get_player_initialization_packets( + ecomp: &EntityComponent, + pcomp: &PlayerComponent, + player: Entity, +) -> (PlayerInfo, SpawnPlayer) { + let display_name = json!({ + "text": ecomp.display_name + }) + .to_string(); + + let mut props = vec![]; + for prop in pcomp.profile_properties.iter() { + props.push(( + prop.name.clone(), + prop.value.clone(), + prop.signature.clone(), + )); + } + + let action = PlayerInfoAction::AddPlayer( + ecomp.display_name.clone(), + props, + Gamemode::Creative, + 50, + display_name, + ); + let player_info = PlayerInfo::new(action, ecomp.uuid.clone()); + + let metadata = EntityMetadata::new().with(&[ + (0, MetaEntry::Byte(0)), + (1, MetaEntry::VarInt(300)), + (2, MetaEntry::OptChat(None)), + (3, MetaEntry::Boolean(false)), + (4, MetaEntry::Boolean(false)), + (5, MetaEntry::Boolean(false)), + (6, MetaEntry::Byte(0)), + (7, MetaEntry::Float(1.0)), + (8, MetaEntry::VarInt(0)), + (9, MetaEntry::Boolean(false)), + (10, MetaEntry::VarInt(0)), + (11, MetaEntry::Float(0.0)), + (12, MetaEntry::VarInt(0)), + (13, MetaEntry::Byte(0)), + (14, MetaEntry::Byte(1)), + // TODO NBT + ]); + + let spawn_player = SpawnPlayer::new( + player.id() as i32, + ecomp.uuid.clone(), + ecomp.position.x, + ecomp.position.y, + ecomp.position.z, + degrees_to_stops(ecomp.position.pitch), + degrees_to_stops(ecomp.position.yaw), + metadata, + ); + + (player_info, spawn_player) +} + +/// System for broadcasting when a player disconnects. +pub struct DisconnectBroadcastSystem { + reader: Option>, +} + +impl DisconnectBroadcastSystem { + pub fn new() -> Self { + Self { reader: None } + } +} + +impl<'a> System<'a> for DisconnectBroadcastSystem { + type SystemData = ( + Read<'a, EventChannel>, + ReadStorage<'a, NetworkComponent>, + ); + + fn run(&mut self, data: Self::SystemData) { + let (events, net_comps) = data; + + for event in events.read(&mut self.reader.as_mut().unwrap()) { + // Broadcast disconnect. + let player_info = PlayerInfo::new(PlayerInfoAction::RemovePlayer, event.uuid.clone()); + + let destroy_entities = DestroyEntities::new(vec![event.player.id() as i32]); + + for net in net_comps.join() { + send_packet_to_player(net, player_info.clone()); + send_packet_to_player(net, destroy_entities.clone()); + } + } + } + + fn setup(&mut self, world: &mut World) { + use specs::SystemData; + Self::SystemData::setup(world); + + self.reader = Some( + world + .fetch_mut::>() + .register_reader(), + ); + } +} + +fn send_chunk_data(chunk: &Chunk, net: &NetworkComponent) { + let packet = ChunkData::new(chunk.clone()); + send_packet_to_player(net, packet); +} + +/// Attempts to send the chunk at the given position to +/// the given player. If the chunk is not loaded, it will +/// be loaded and sent at a later time as soon as it is +/// loaded. +pub fn send_chunk_to_player( + chunk_pos: ChunkPosition, + net: &NetworkComponent, + player: Entity, + chunk_map: &ChunkMap, + chunk_handle: &ChunkWorkerHandle, + lazy: &LazyUpdate, +) { + if let Some(chunk) = chunk_map.chunk_at(chunk_pos) { + send_chunk_data(chunk, net); + } else { + // Queue for loading + load_chunk(chunk_handle, chunk_pos); + lazy.exec_mut(move |world| { + world + .write_component::() + .get_mut(player) + .unwrap() + .pending + .insert(chunk_pos); + }); + } +} + +fn degrees_to_stops(degs: f32) -> u8 { + ((degs / 360.0) * 256.) as u8 +} diff --git a/server/src/testframework.rs b/server/src/testframework.rs new file mode 100644 index 000000000..48873a724 --- /dev/null +++ b/server/src/testframework.rs @@ -0,0 +1,156 @@ +//! Helper framework for writing unit tests. + +use std::net::TcpListener; +use std::sync::atomic::AtomicUsize; +use std::sync::Arc; + +use mio_extras::channel::{channel, Receiver, Sender}; +use rand::Rng; +use specs::{Builder, Dispatcher, Entity, World, WorldExt}; +use uuid::Uuid; + +use feather_core::network::packet::{Packet, PacketType}; +use feather_core::world::Position; +use feather_core::Gamemode; + +use crate::config::Config; +use crate::entity::{EntityComponent, PlayerComponent}; +use crate::io::ServerToWorkerMessage; +use crate::network::{NetworkComponent, PacketQueue}; +use crate::PlayerCount; + +/// Initializes a Specs world and dispatcher +/// using default configuration options and an +/// available server port. +pub fn init_world<'a, 'b>() -> (World, Dispatcher<'a, 'b>) { + let mut config = Config::default(); + config.server.port = find_open_port().unwrap(); + + let config = Arc::new(config); + + let player_count = Arc::new(PlayerCount(AtomicUsize::new(0))); + let ioman = super::init_io_manager(Arc::clone(&config), Arc::clone(&player_count)); + + super::init_world(config, player_count, ioman) +} + +pub struct Player { + pub entity: Entity, + pub network_sender: Sender, + pub network_receiver: Receiver, +} + +/// Adds a player to the world, inserting +/// all the necessary components. Returns +/// a number of useful channels. +pub fn add_player(world: &mut World) -> Player { + let (ns1, nr1) = channel(); + let (ns2, nr2) = channel(); + let e = world + .create_entity() + .with(NetworkComponent::new(ns1, nr2)) + .with(PlayerComponent { + gamemode: Gamemode::Creative, + profile_properties: vec![], + }) + .with(EntityComponent { + uuid: Uuid::new_v4(), + on_ground: true, + position: Position::new(0.0, 0.0, 0.0, 0.0, 0.0), + display_name: "Test".to_string(), + }) + .build(); + + Player { + entity: e, + network_sender: ns2, + network_receiver: nr1, + } +} + +/// Asserts that the given player has received +/// a packet of the given type, returning the packet. +pub fn assert_packet_received(player: &Player, ty: PacketType) -> Box { + match player.network_receiver.try_recv().unwrap() { + ServerToWorkerMessage::SendPacket(pack) => { + assert_eq!(pack.ty(), ty); + pack + } + _ => panic!(), + } +} + +/// Retrieves up to `cap` packets sent to a player, if any. +/// If `cap` is set to `None`, all packets will be read. +/// +/// Note that this function consumes messages in +/// the network channel until enough packets have been read. +pub fn received_packets(player: &Player, cap: Option) -> Vec> { + let mut result = vec![]; + + while let Ok(msg) = player.network_receiver.try_recv() { + if let Some(cap) = cap.as_ref() { + if result.len() >= *cap { + break; + } + } + match msg { + ServerToWorkerMessage::SendPacket(pack) => result.push(pack), + _ => (), + } + } + + result +} + +/// Adds a received packet to the packet queue +/// for a given player. +pub fn receive_packet(player: &Player, world: &World, packet: P) { + let queue = world.fetch_mut::(); + queue.add_for_packet(player.entity, Box::new(packet)); +} + +/// Attempts to find an available port. +fn find_open_port() -> Option { + let start = rand::thread_rng().gen_range(10000, 30000); + (start..60000).find(|port| TcpListener::bind(("127.0.0.1", *port)).is_ok()) +} + +/// Heh... tests for the testing framework. +/// Not sure what the point of this is, since +/// all other tests would fail if the testing +/// framework didn't work. +mod tests { + use crate::entity::{EntityComponent, PlayerComponent}; + use crate::network::NetworkComponent; + + use super::*; + + #[test] + fn test_find_open_port() { + let port = find_open_port().unwrap(); + println!("Found open port: {}", port); + assert!(TcpListener::bind(("127.0.0.1", port)).is_ok()); + } + + #[test] + fn test_init_world() { + // Check that initializing the world doesn't cause + // a panic. + let (w, mut d) = init_world(); + + // Check that running the dispatcher works fine + d.dispatch(&w); + } + + #[test] + fn test_add_player() { + let (mut w, _d) = init_world(); + + let entity = add_player(&mut w).entity; + + assert!(w.read_component::().get(entity).is_some()); + assert!(w.read_component::().get(entity).is_some()); + assert!(w.read_component::().get(entity).is_some()); + } +} diff --git a/server/src/worldupdate.rs b/server/src/worldupdate.rs new file mode 100644 index 000000000..533d65af3 --- /dev/null +++ b/server/src/worldupdate.rs @@ -0,0 +1,257 @@ +//! This module handles packets relating to world updates, +//! including block update packets. + +use specs::{Entity, Join, LazyUpdate, Read, ReadStorage, System, Write}; + +use feather_core::network::cast_packet; +use feather_core::network::packet::implementation::BlockChange; +use feather_core::network::packet::implementation::{PlayerDigging, PlayerDiggingStatus}; +use feather_core::network::packet::PacketType; +use feather_core::world::block::{Block, BlockToId}; +use feather_core::world::{BlockPosition, ChunkMap}; +use feather_core::Gamemode; + +use crate::disconnect_player; +use crate::entity::PlayerComponent; +use crate::network::{send_packet_to_player, NetworkComponent, PacketQueue}; + +/// System responsible for polling for PlayerDigging +/// packets and handling them accordingly. +pub struct PlayerDiggingSystem; + +impl<'a> System<'a> for PlayerDiggingSystem { + type SystemData = ( + Write<'a, ChunkMap>, + ReadStorage<'a, PlayerComponent>, + ReadStorage<'a, NetworkComponent>, + Read<'a, PacketQueue>, + Read<'a, LazyUpdate>, + ); + + fn run(&mut self, data: Self::SystemData) { + let (mut chunk_map, pcomps, netcomps, packet_queue, lazy) = data; + + let mut packets = vec![]; + packets.append(&mut packet_queue.for_packet(PacketType::PlayerDigging)); + + // Handle packets + for (player, _packet) in packets { + let packet = cast_packet::(&_packet); + handle_player_digging(&mut chunk_map, packet, player, &netcomps, &pcomps, &lazy); + } + } +} + +/// Handles a Player Digging packet. +fn handle_player_digging( + chunk_map: &mut ChunkMap, + packet: &PlayerDigging, + player: Entity, + netcomps: &ReadStorage, + pcomps: &ReadStorage, + lazy: &LazyUpdate, +) { + let pcomp = pcomps.get(player).unwrap(); + match packet.status { + PlayerDiggingStatus::StartedDigging => { + if pcomp.gamemode == Gamemode::Creative { + if handle_block_break(chunk_map, packet.location, &netcomps, &pcomps).is_err() { + disconnect_player( + player, + format!( + "Attempted to break block in unloaded chunk ({:?})", + packet.location + ), + &lazy, + ); + } + } + } + PlayerDiggingStatus::FinishedDigging => { + if handle_block_break(chunk_map, packet.location, &netcomps, &pcomps).is_err() { + disconnect_player( + player, + format!( + "Attempted to break block in unloaded chunk ({:?})", + packet.location + ), + &lazy, + ); + } + } + status => warn!("Unhandled Player Digging status: {:?}", status), + } +} + +/// Handles a broken block by a player. +fn handle_block_break( + chunk_map: &mut ChunkMap, + pos: BlockPosition, + netcomps: &ReadStorage, + pcomps: &ReadStorage, +) -> Result<(), ()> { + chunk_map.set_block_at(pos, Block::Air)?; + + broadcast_block_update(pos, Block::Air, netcomps, pcomps); + + Ok(()) +} + +/// Broadcasts a block update to all joined players. +fn broadcast_block_update( + pos: BlockPosition, + new_block: Block, + netcomps: &ReadStorage, + pcomps: &ReadStorage, +) { + for (net, _) in (netcomps, pcomps).join() { + let block_update = BlockChange::new(pos, new_block.block_state_id() as i32); + send_packet_to_player(net, block_update); + } +} + +#[cfg(test)] +mod tests { + use specs::{RunNow, WorldExt}; + + use feather_core::world::chunk::Chunk; + use feather_core::world::ChunkPosition; + + use crate::testframework as t; + + use super::*; + + #[test] + fn test_system() { + let (mut w, _) = t::init_world(); + let player = t::add_player(&mut w); + + t::receive_packet( + &player, + &w, + PlayerDigging::new( + PlayerDiggingStatus::StartedDigging, + BlockPosition::new(0, 0, 0), + 0, + ), + ); + + let mut system = PlayerDiggingSystem; + system.run_now(&w); + } + + #[test] + fn test_handle_player_digging() { + let (mut w, _) = t::init_world(); + + let mut chunk = Chunk::new(ChunkPosition::new(0, 0)); + chunk.set_block_at(0, 0, 0, Block::Stone); + w.write_resource::() + .set_chunk_at(ChunkPosition::new(0, 0), chunk); + + let pos = BlockPosition::new(0, 0, 0); + + // Test with creative mode player + let player = t::add_player(&mut w); + + let packet = PlayerDigging::new(PlayerDiggingStatus::StartedDigging, pos, 0); + handle_player_digging( + &mut w.fetch_mut(), + &packet, + player.entity, + &w.read_component(), + &w.read_component(), + &w.read_resource(), + ); + + // Call lazily updated disconnect + w.maintain(); + + // Make sure player wasn't disconnected + let packets = t::received_packets(&player, None); + packets + .iter() + .for_each(|packet| assert_ne!(packet.ty(), PacketType::DisconnectPlay)); + + // Make sure player was notified of block update + let _block_change = packets + .iter() + .find(|packet| packet.ty() == PacketType::BlockChange) + .unwrap(); + let block_change = cast_packet::(&_block_change); + assert_eq!(block_change.location, pos); + assert_eq!(block_change.block_id, Block::Air.block_state_id() as i32); + + // Make sure block was actually updated + assert_eq!( + w.read_resource::().block_at(pos).unwrap(), + Block::Air + ); + } + + #[test] + fn test_handle_block_break() { + let (mut w, _) = t::init_world(); + + // Confirm that breaking block in unloaded chunk fails + let _player = t::add_player(&mut w); + assert!(handle_block_break( + &mut w.fetch_mut(), + BlockPosition::new(1024, 9, 1024), + &w.read_component(), + &w.read_component(), + ) + .is_err()); + + // Break block in known chunk + let _player = t::add_player(&mut w); + + let pos = ChunkPosition::new(0, 0); + let mut chunk = Chunk::new(pos); + chunk.set_block_at(0, 0, 0, Block::Stone); + w.write_resource::().set_chunk_at(pos, chunk); + + assert!(handle_block_break( + &mut w.write_resource(), + BlockPosition::new(0, 0, 0), + &w.read_component(), + &w.read_component(), + ) + .is_ok()); + assert_eq!( + w.read_resource::() + .block_at(BlockPosition::new(0, 0, 0)) + .unwrap(), + Block::Air + ); + } + + #[test] + fn test_broadcast_block_update() { + let (mut w, _) = t::init_world(); + + let player = t::add_player(&mut w); + let player2 = t::add_player(&mut w); + w.write_component::() + .remove(player2.entity) + .unwrap(); + + broadcast_block_update( + BlockPosition::new(0, 0, 0), + Block::Sand, + &w.read_component(), + &w.read_component(), + ); + + // Check that the joined player received block update but the unjoined + // player did not + let packet = t::assert_packet_received(&player, PacketType::BlockChange); + + let packet = cast_packet::(&packet); + assert_eq!(packet.location, BlockPosition::new(0, 0, 0)); + assert_eq!(packet.block_id, Block::Sand.block_state_id() as i32); + + let p2_packets = t::received_packets(&player2, None); + assert_eq!(p2_packets.len(), 0); + } +}