veloren/server/src/lib.rs

#![feature(drain_filter)]

pub mod client;
pub mod cmd;
pub mod error;
pub mod input;

// Reexports
pub use crate::{error::Error, input::Input};

use crate::{
    client::{Client, Clients},
    cmd::CHAT_COMMANDS,
};
use common::{
    comp,
    comp::character::Animation,
    msg::{ClientMsg, ClientState, RequestStateError, ServerMsg},
    net::PostOffice,
    state::{State, Uid},
    terrain::TerrainChunk,
};
use specs::{
    join::Join, saveload::MarkedBuilder, world::EntityBuilder as EcsEntityBuilder, Builder,
    Entity as EcsEntity,
};
use std::{collections::HashSet, i32, net::SocketAddr, sync::mpsc, time::Duration};
use threadpool::ThreadPool;
use vek::*;
use world::World;

const CLIENT_TIMEOUT: f64 = 20.0; // Seconds

pub enum Event {
    ClientConnected { entity: EcsEntity },
    ClientDisconnected { entity: EcsEntity },
    Chat { entity: EcsEntity, msg: String },
}

pub struct Server {
    state: State,
    world: World,

    postoffice: PostOffice<ServerMsg, ClientMsg>,
    clients: Clients,

    thread_pool: ThreadPool,
    chunk_tx: mpsc::Sender<(Vec3<i32>, TerrainChunk)>,
    chunk_rx: mpsc::Receiver<(Vec3<i32>, TerrainChunk)>,
    pending_chunks: HashSet<Vec3<i32>>,
}

impl Server {
    /// Create a new `Server` bound to the default socket.
    #[allow(dead_code)]
    pub fn new() -> Result<Self, Error> {
        Self::bind(SocketAddr::from(([0; 4], 59003)))
    }

    /// Create a new server bound to the given socket
    #[allow(dead_code)]
    pub fn bind<A: Into<SocketAddr>>(addrs: A) -> Result<Self, Error> {
        let (chunk_tx, chunk_rx) = mpsc::channel();

        let mut state = State::new();
        state.ecs_mut().register::<comp::phys::ForceUpdate>();

        let mut this = Self {
            state,
            world: World::new(),

            postoffice: PostOffice::bind(addrs.into())?,
            clients: Clients::empty(),

            thread_pool: threadpool::Builder::new()
                .thread_name("veloren-worker".into())
                .build(),
            chunk_tx,
            chunk_rx,
            pending_chunks: HashSet::new(),
        };

        for i in 0..4 {
            this.create_npc(comp::Character::random())
                .with(comp::Control::default())
                .with(comp::Agent::Wanderer(Vec2::zero()))
                .build();
        }

        Ok(this)
    }

    /// Get a reference to the server's game state.
    #[allow(dead_code)]
    pub fn state(&self) -> &State {
        &self.state
    }
    /// Get a mutable reference to the server's game state.
    #[allow(dead_code)]
    pub fn state_mut(&mut self) -> &mut State {
        &mut self.state
    }

    /// Get a reference to the server's world.
    #[allow(dead_code)]
    pub fn world(&self) -> &World {
        &self.world
    }
    /// Get a mutable reference to the server's world.
    #[allow(dead_code)]
    pub fn world_mut(&mut self) -> &mut World {
        &mut self.world
    }

    /// Build a non-player character
    #[allow(dead_code)]
    pub fn create_npc(&mut self, character: comp::Character) -> EcsEntityBuilder {
        self.state
            .ecs_mut()
            .create_entity_synced()
            .with(comp::phys::Pos(Vec3::new(0.0, 0.0, 64.0)))
            .with(comp::phys::Vel(Vec3::zero()))
            .with(comp::phys::Dir(Vec3::unit_y()))
            .with(comp::AnimationHistory::new(Animation::Idle))
            .with(character)
    }

    pub fn create_player_character(
        state: &mut State,
        entity: EcsEntity,
        client: &mut Client,
        character: comp::Character,
    ) {
        state.write_component(entity, character);
        state.write_component(entity, comp::phys::Pos(Vec3::new(0.0, 0.0, 64.0)));
        state.write_component(entity, comp::phys::Vel(Vec3::zero()));
        state.write_component(entity, comp::phys::Dir(Vec3::unit_y()));
        // Make sure everything is accepted
        state.write_component(entity, comp::phys::ForceUpdate);

        // Set initial animation
        state.write_component(entity, comp::AnimationHistory::new(Animation::Idle));

        // Tell the client his request was successful
        client.notify(ServerMsg::StateAnswer(Ok(ClientState::Character)));
        client.client_state = ClientState::Character;
    }

    /// Execute a single server tick, handle input and update the game state by the given duration
    #[allow(dead_code)]
    pub fn tick(&mut self, input: Input, dt: Duration) -> Result<Vec<Event>, Error> {
        // This tick function is the centre of the Veloren universe. Most server-side things are
        // managed from here, and as such it's important that it stays organised. Please consult
        // the core developers before making significant changes to this code. Here is the
        // approximate order of things. Please update it as this code changes.
        //
        // 1) Collect input from the frontend, apply input effects to the state of the game
        // 2) Go through any events (timer-driven or otherwise) that need handling and apply them
        //    to the state of the game
        // 3) Go through all incoming client network communications, apply them to the game state
        // 4) Perform a single LocalState tick (i.e: update the world and entities in the world)
        // 5) Go through the terrain update queue and apply all changes to the terrain
        // 6) Send relevant state updates to all clients
        // 7) Finish the tick, passing control of the main thread back to the frontend

        // Build up a list of events for this frame, to be passed to the frontend
        let mut frontend_events = Vec::new();

        // If networking has problems, handle them
        if let Some(err) = self.postoffice.error() {
            return Err(err.into());
        }

        // Handle new client connections (step 2)
        frontend_events.append(&mut self.handle_new_connections()?);

        // Handle new messages from clients
        frontend_events.append(&mut self.handle_new_messages()?);

        // Tick the client's LocalState (step 3)
        self.state.tick(dt);

        // Fetch any generated `TerrainChunk`s and insert them into the terrain
        // Also, send the chunk data to anybody that is close by
        if let Ok((key, chunk)) = self.chunk_rx.try_recv() {
            // Send the chunk to all nearby players
            for (entity, player, pos) in (
                &self.state.ecs().entities(),
                &self.state.ecs().read_storage::<comp::Player>(),
                &self.state.ecs().read_storage::<comp::phys::Pos>(),
            )
                .join()
            {
                let chunk_pos = self.state.terrain().pos_key(pos.0.map(|e| e as i32));
                let dist = (Vec2::from(chunk_pos) - Vec2::from(key))
                    .map(|e: i32| e.abs())
                    .reduce_max();

                if dist < 7 {
                    self.clients.notify(
                        entity,
                        ServerMsg::TerrainChunkUpdate {
                            key,
                            chunk: Box::new(chunk.clone()),
                        },
                    );
                }
            }

            self.state.insert_chunk(key, chunk);
            self.pending_chunks.remove(&key);
        }

        // Remove chunks that are too far from players
        let mut chunks_to_remove = Vec::new();
        self.state.terrain().iter().for_each(|(key, _)| {
            let mut min_dist = i32::MAX;

            // For each player with a position, calculate the distance
            for (_, pos) in (
                &self.state.ecs().read_storage::<comp::Player>(),
                &self.state.ecs().read_storage::<comp::phys::Pos>(),
            )
                .join()
            {
                let chunk_pos = self.state.terrain().pos_key(pos.0.map(|e| e as i32));
                let dist = Vec2::from(chunk_pos - key)
                    .map(|e: i32| e.abs())
                    .reduce_max();
                min_dist = min_dist.min(dist);
            }

            if min_dist > 7 {
                chunks_to_remove.push(key);
            }
        });
        for key in chunks_to_remove {
            self.state.remove_chunk(key);
        }

        // Synchronise clients with the new state of the world
        self.sync_clients();

        // Finish the tick, pass control back to the frontend (step 6)
        Ok(frontend_events)
    }

    /// Clean up the server after a tick
    #[allow(dead_code)]
    pub fn cleanup(&mut self) {
        // Cleanup the local state
        self.state.cleanup();
    }

    /// Handle new client connections
    fn handle_new_connections(&mut self) -> Result<Vec<Event>, Error> {
        let mut frontend_events = Vec::new();

        for mut postbox in self.postoffice.new_postboxes() {
            let entity = self.state.ecs_mut().create_entity_synced().build();
            let mut client = Client {
                client_state: ClientState::Connected,
                postbox,
                last_ping: self.state.get_time(),
            };

            // Return the state of the current world
            // (All components Sphynx tracks)
            client.notify(ServerMsg::InitialSync {
                ecs_state: self.state.ecs().gen_state_package(),
                entity_uid: self.state.ecs().uid_from_entity(entity).unwrap().into(), // Can't fail
            });

            self.clients.add(entity, client);

            frontend_events.push(Event::ClientConnected { entity });
        }

        Ok(frontend_events)
    }

    /// Handle new client messages
    fn handle_new_messages(&mut self) -> Result<Vec<Event>, Error> {
        let mut frontend_events = Vec::new();

        let state = &mut self.state;
        let mut new_chat_msgs = Vec::new();
        let mut disconnected_clients = Vec::new();
        let mut requested_chunks = Vec::new();

        self.clients.remove_if(|entity, client| {
            let mut disconnect = false;
            let new_msgs = client.postbox.new_messages();

            // Update client ping
            if new_msgs.len() > 0 {
                client.last_ping = state.get_time();

                // Process incoming messages
                for msg in new_msgs {
                    match msg {
                        ClientMsg::RequestState(requested_state) => match requested_state {
                            ClientState::Connected => disconnect = true, // Default state
                            ClientState::Registered => match client.client_state {
                                // Use ClientMsg::Register instead
                                ClientState::Connected => {
                                    client.error_state(RequestStateError::WrongMessage)
                                }
                                ClientState::Registered => {
                                    client.error_state(RequestStateError::Already)
                                }
                                ClientState::Spectator | ClientState::Character => {
                                    client.allow_state(ClientState::Registered)
                                }
                            },
                            ClientState::Spectator => match requested_state {
                                // Become Registered first
                                ClientState::Connected => {
                                    client.error_state(RequestStateError::Impossible)
                                }
                                ClientState::Spectator => {
                                    client.error_state(RequestStateError::Already)
                                }
                                ClientState::Registered | ClientState::Character => {
                                    client.allow_state(ClientState::Spectator)
                                }
                            },
                            // Use ClientMsg::Character instead
                            ClientState::Character => {
                                client.error_state(RequestStateError::WrongMessage)
                            }
                        },
                        ClientMsg::Register { player } => match client.client_state {
                            ClientState::Connected => {
                                Self::initialize_player(state, entity, client, player)
                            }
                            // Use RequestState instead (No need to send `player` again)
                            _ => client.error_state(RequestStateError::Impossible),
                        },
                        ClientMsg::Character(character) => match client.client_state {
                            // Become Registered first
                            ClientState::Connected => {
                                client.error_state(RequestStateError::Impossible)
                            }
                            ClientState::Registered | ClientState::Spectator => {
                                Self::create_player_character(state, entity, client, character)
                            }
                            ClientState::Character => {
                                client.error_state(RequestStateError::Already)
                            }
                        },
                        ClientMsg::Chat(msg) => match client.client_state {
                            ClientState::Connected => {
                                client.error_state(RequestStateError::Impossible)
                            }
                            ClientState::Registered
                            | ClientState::Spectator
                            | ClientState::Character => new_chat_msgs.push((entity, msg)),
                        },
                        ClientMsg::PlayerAnimation(animation_history) => {
                            match client.client_state {
                                ClientState::Character => {
                                    state.write_component(entity, animation_history)
                                }
                                // Only characters can send animations
                                _ => client.error_state(RequestStateError::Impossible),
                            }
                        }
                        ClientMsg::PlayerPhysics { pos, vel, dir } => match client.client_state {
                            ClientState::Character => {
                                state.write_component(entity, pos);
                                state.write_component(entity, vel);
                                state.write_component(entity, dir);
                            }
                            // Only characters send their position
                            _ => client.error_state(RequestStateError::Impossible),
                        },
                        ClientMsg::TerrainChunkRequest { key } => match client.client_state {
                            ClientState::Connected | ClientState::Registered => {
                                client.error_state(RequestStateError::Impossible);
                            }
                            ClientState::Spectator | ClientState::Character => {
                                match state.terrain().get_key(key) {
                                    Some(chunk) => {
                                        client.postbox.send_message(ServerMsg::TerrainChunkUpdate {
                                            key,
                                            chunk: Box::new(chunk.clone()),
                                        })
                                    }
                                    None => requested_chunks.push(key),
                                }
                            }
                        },
                        // Always possible
                        ClientMsg::Ping => client.postbox.send_message(ServerMsg::Pong),
                        ClientMsg::Pong => {}
                        ClientMsg::Disconnect => disconnect = true,
                    }
                }
            } else if state.get_time() - client.last_ping > CLIENT_TIMEOUT || // Timeout
                client.postbox.error().is_some()
            // Postbox error
            {
                disconnect = true;
            } else if state.get_time() - client.last_ping > CLIENT_TIMEOUT * 0.5 {
                // Try pinging the client if the timeout is nearing
                client.postbox.send_message(ServerMsg::Ping);
            }

            if disconnect {
                disconnected_clients.push(entity);
                client.postbox.send_message(ServerMsg::Disconnect);
                true
            } else {
                false
            }
        });

        // Handle new chat messages
        for (entity, msg) in new_chat_msgs {
            // Handle chat commands
            if msg.starts_with("/") && msg.len() > 1 {
                let argv = String::from(&msg[1..]);
                self.process_chat_cmd(entity, argv);
            } else {
                self.clients.notify_registered(ServerMsg::Chat(
                    match self.state.ecs().read_storage::<comp::Player>().get(entity) {
                        Some(player) => format!("[{}] {}", &player.alias, msg),
                        None => format!("[<anon>] {}", msg),
                    },
                ));

                frontend_events.push(Event::Chat { entity, msg });
            }
        }

        // Handle client disconnects
        for entity in disconnected_clients {
            self.state.ecs_mut().delete_entity_synced(entity);

            frontend_events.push(Event::ClientDisconnected { entity });
        }

        // Generate requested chunks
        for key in requested_chunks {
            self.generate_chunk(key);
        }

        Ok(frontend_events)
    }

    /// Initialize a new client states with important information
    fn initialize_player(
        state: &mut State,
        entity: specs::Entity,
        client: &mut Client,
        player: comp::Player,
    ) {
        // Save player metadata (for example the username)
        state.write_component(entity, player);

        // Sync logical information other players have authority over, not the server
        for (other_entity, &uid, &animation_history) in (
            &state.ecs().entities(),
            &state.ecs().read_storage::<common::state::Uid>(),
            &state.ecs().read_storage::<comp::AnimationHistory>(),
        )
            .join()
        {
            // AnimationHistory
            client.postbox.send_message(ServerMsg::EntityAnimation {
                entity: uid.into(),
                animation_history: animation_history,
            });
        }

        // Tell the client his request was successful
        client.allow_state(ClientState::Registered);
    }

    /// Sync client states with the most up to date information
    fn sync_clients(&mut self) {
        // Sync 'logical' state using Sphynx
        self.clients
            .notify_registered(ServerMsg::EcsSync(self.state.ecs_mut().next_sync_package()));

        // Sync 'physical' state
        for (entity, &uid, &pos, &vel, &dir, force_update) in (
            &self.state.ecs().entities(),
            &self.state.ecs().read_storage::<Uid>(),
            &self.state.ecs().read_storage::<comp::phys::Pos>(),
            &self.state.ecs().read_storage::<comp::phys::Vel>(),
            &self.state.ecs().read_storage::<comp::phys::Dir>(),
            self.state
                .ecs()
                .read_storage::<comp::phys::ForceUpdate>()
                .maybe(),
        )
            .join()
        {
            let msg = ServerMsg::EntityPhysics {
                entity: uid.into(),
                pos,
                vel,
                dir,
            };

            match force_update {
                Some(_) => self.clients.notify_ingame(msg),
                None => self.clients.notify_ingame_except(entity, msg),
            }
        }

        // Sync animation states
        for (entity, &uid, &animation_history) in (
            &self.state.ecs().entities(),
            &self.state.ecs().read_storage::<Uid>(),
            &self.state.ecs().read_storage::<comp::AnimationHistory>(),
        )
            .join()
        {
            // Check if we need to sync
            if Some(animation_history.current) == animation_history.last {
                continue;
            }

            self.clients.notify_ingame_except(
                entity,
                ServerMsg::EntityAnimation {
                    entity: uid.into(),
                    animation_history,
                },
            );
        }

        // Update animation last/current state
        for (entity, mut animation_history) in (
            &self.state.ecs().entities(),
            &mut self.state.ecs().write_storage::<comp::AnimationHistory>(),
        )
            .join()
        {
            animation_history.last = Some(animation_history.current);
        }

        // Remove all force flags
        self.state
            .ecs_mut()
            .write_storage::<comp::phys::ForceUpdate>()
            .clear();
    }

    pub fn generate_chunk(&mut self, key: Vec3<i32>) {
        if self.pending_chunks.insert(key) {
            let chunk_tx = self.chunk_tx.clone();
            self.thread_pool
                .execute(move || chunk_tx.send((key, World::generate_chunk(key))).unwrap());
        }
    }

    fn process_chat_cmd(&mut self, entity: EcsEntity, cmd: String) {
        // separate string into keyword and arguments
        let sep = cmd.find(' ');
        let (kwd, args) = match sep {
            Some(i) => (cmd[..i].to_string(), cmd[(i + 1)..].to_string()),
            None => (cmd, "".to_string()),
        };

        // find command object and run its handler
        let action_opt = CHAT_COMMANDS.iter().find(|x| x.keyword == kwd);
        match action_opt {
            Some(action) => action.execute(self, entity, args),
            // unknown command
            None => {
                self.clients.notify(
                    entity,
                    ServerMsg::Chat(format!(
                        "Unrecognised command: '/{}'\ntype '/help' for a list of available commands",
                        kwd
                    )),
                );
            }
        }
    }
}

impl Drop for Server {
    fn drop(&mut self) {
        self.clients.notify_registered(ServerMsg::Shutdown);
    }
}