Merge pull request #88 from ferrumc-rs/feature/systems

feature/systems

src/bin/Cargo.toml

@@ -25,6 +25,7 @@ ferrumc-net-codec = { workspace = true }
 ferrumc-plugins = { workspace = true }
 ferrumc-storage = { workspace = true }
 ferrumc-utils = { workspace = true }
+ferrumc-config = { workspace = true }
 ferrumc-profiling = { workspace = true }
 ferrumc-logging = { workspace = true }
 ferrumc-world = { workspace = true }
@@ -35,8 +36,9 @@ ctor = { workspace = true }
 parking_lot = { workspace = true }
 tracing = { workspace = true }
 tokio = { workspace = true }
+futures = { workspace = true }
 serde_json = { workspace = true }
-
+async-trait = "0.1.83"
 [[bin]]
 name = "ferrumc"
 path = "src/main.rs"

src/bin/src/errors.rs

@@ -37,8 +37,9 @@ pub enum BinaryError {
     #[allow(dead_code)]
     #[error("{0}")]
     Custom(String),
-    // #[error("IO error: {0}")]
-    // Io(#[from] std::io::Error),
-
-    // idk add ur own errors here
+
+    #[error("Tokio Join Error")]
+    TokioJoinError(#[from] tokio::task::JoinError),
+    #[error("IO error: {0}")]
+    Io(#[from] std::io::Error),
 }

src/bin/src/main.rs

@@ -1,13 +1,15 @@
 // Security or something like that
 #![forbid(unsafe_code)]
 
-use std::sync::{atomic::AtomicBool, Arc};
-use tracing::{error, info};
 use ferrumc_ecs::Universe;
 use ferrumc_net::ServerState;
+use std::sync::{Arc};
+use tracing::{error, info};
+use systems::definition;
 
-pub(crate)mod errors;
+pub(crate) mod errors;
 mod packet_handlers;
+mod systems;
 
 pub type Result<T> = std::result::Result<T, errors::BinaryError>;
 
@@ -26,11 +28,29 @@ async fn main() {
 }
 
 async fn entry() -> Result<()> {
-    let listener = ferrumc_net::server::create_server_listener().await?;
+    let state = create_state().await?;
+    let global_state = Arc::new(state);
+
+    let all_systems = tokio::spawn(definition::start_all_systems(Arc::clone(&global_state)));
 
-    let state = ServerState::new(Universe::new(),AtomicBool::new(false));
-
-    ferrumc_net::server::listen(Arc::new(state), listener).await?;
+    // Start the systems and wait until all of them are done
+    all_systems.await??;
+
+    // Stop all systems
+    definition::stop_all_systems(global_state).await?;
 
     Ok(())
+}
+
+
+async fn create_state() -> Result<ServerState> {
+    let config = ferrumc_config::statics::get_global_config();
+    let addy = format!("{}:{}", config.host, config.port);
+
+    let listener = tokio::net::TcpListener::bind(addy).await?;
+
+    Ok(ServerState {
+        universe: Universe::new(),
+        tcp_listener: listener,
+    })
 }

src/bin/src/packet_handlers/login_process.rs

@@ -1,43 +1,49 @@
-use ferrumc_macros::event_handler;
-use ferrumc_net::errors::NetError;
-use ferrumc_net::packets::incoming::login_start::LoginStartEvent;
-use ferrumc_net::GlobalState;
-use tracing::{info, trace};
+use ferrumc_core::identity::player_identity::PlayerIdentity;
 use ferrumc_ecs::components::storage::ComponentRefMut;
+use ferrumc_macros::event_handler;
 use ferrumc_net::connection::{ConnectionState, StreamWriter};
+use ferrumc_net::errors::NetError;
 use ferrumc_net::packets::incoming::ack_finish_configuration::AckFinishConfigurationEvent;
-use ferrumc_net::packets::incoming::login_acknowledged::{LoginAcknowledgedEvent};
+use ferrumc_net::packets::incoming::login_acknowledged::LoginAcknowledgedEvent;
+use ferrumc_net::packets::incoming::login_start::LoginStartEvent;
 use ferrumc_net::packets::incoming::server_bound_known_packs::ServerBoundKnownPacksEvent;
 use ferrumc_net::packets::outgoing::client_bound_known_packs::ClientBoundKnownPacksPacket;
-use ferrumc_net::packets::outgoing::finish_configuration::FinishConfigurationPacket;
 use ferrumc_net::packets::outgoing::game_event::GameEventPacket;
 use ferrumc_net::packets::outgoing::keep_alive::{KeepAlive, KeepAlivePacket};
 use ferrumc_net::packets::outgoing::login_play::LoginPlayPacket;
 use ferrumc_net::packets::outgoing::login_success::LoginSuccessPacket;
 use ferrumc_net::packets::outgoing::registry_data::{get_registry_packets};
 use ferrumc_net::packets::outgoing::set_default_spawn_position::SetDefaultSpawnPositionPacket;
 use ferrumc_net::packets::outgoing::synchronize_player_position::SynchronizePlayerPositionPacket;
-use ferrumc_net_codec::encode::{NetEncodeOpts};
+use ferrumc_net::GlobalState;
+use ferrumc_net_codec::encode::NetEncodeOpts;
+use tracing::{debug, trace};
 
 #[event_handler]
 async fn handle_login_start(
     login_start_event: LoginStartEvent,
     state: GlobalState,
 ) -> Result<LoginStartEvent, NetError> {
-
-    info!("Handling login start event");
+    debug!("Handling login start event");
 
     let uuid = login_start_event.login_start_packet.uuid;
-    let username = login_start_event.login_start_packet.username.clone();
-    info!("Received login start from user with username {}", username);
+    let username = login_start_event.login_start_packet.username.as_str();
+    debug!("Received login start from user with username {}", username);
 
+
+    // Add the player identity component to the ECS for the entity.
+    state.universe.add_component::<PlayerIdentity>(
+        login_start_event.conn_id,
+        PlayerIdentity::new(username.to_string(), uuid),
+    )?;
+
     //Send a Login Success Response to further the login sequence
-    let response = LoginSuccessPacket::new(uuid, username);
     let mut writer = state
         .universe
         .get_mut::<StreamWriter>(login_start_event.conn_id)?;
 
-    writer.send_packet(&response, &NetEncodeOpts::WithLength).await?;
+    writer.send_packet(&LoginSuccessPacket::new(uuid, username), &NetEncodeOpts::WithLength).await?;
+
     Ok(login_start_event)
 }
 
@@ -46,7 +52,6 @@ async fn handle_login_acknowledged(
     login_acknowledged_event: LoginAcknowledgedEvent,
     state: GlobalState,
 ) -> Result<LoginAcknowledgedEvent, NetError> {
-
     trace!("Handling Login Acknowledged event");
 
     //Set the connection State to Configuration
@@ -79,11 +84,9 @@ async fn handle_server_bound_known_packs(
     let mut writer = state
         .universe
         .get_mut::<StreamWriter>(server_bound_known_packs_event.conn_id)?;
-    
+
     let registry_packets = get_registry_packets();
     writer.send_packet(&registry_packets, &NetEncodeOpts::None).await?;
-
-    writer.send_packet(&FinishConfigurationPacket::new(), &NetEncodeOpts::WithLength).await?;
 
     Ok(server_bound_known_packs_event)
 }
@@ -100,31 +103,31 @@ async fn handle_ack_finish_configuration(
     let mut conn_state = state
         .universe
         .get_mut::<ConnectionState>(conn_id)?;
-    
+
     *conn_state = ConnectionState::Play;
 
     let mut writer = state
         .universe
         .get_mut::<StreamWriter>(conn_id)?;
-    
+
     writer.send_packet(&LoginPlayPacket::new(conn_id), &NetEncodeOpts::WithLength).await?;
     writer.send_packet(&SetDefaultSpawnPositionPacket::default(), &NetEncodeOpts::WithLength).await?;
     writer.send_packet(&SynchronizePlayerPositionPacket::default(), &NetEncodeOpts::WithLength).await?;
     writer.send_packet(&GameEventPacket::start_waiting_for_level_chunks(), &NetEncodeOpts::WithLength).await?;
 
     send_keep_alive(conn_id, state, &mut writer).await?;
-    
+
 
     Ok(ack_finish_configuration_event)
 }
 async fn send_keep_alive(conn_id: usize, state: GlobalState, writer: &mut ComponentRefMut<'_, StreamWriter>) -> Result<(), NetError> {
     let keep_alive_packet = KeepAlivePacket::default();
     writer.send_packet(&keep_alive_packet, &NetEncodeOpts::WithLength).await?;
-    
+
     let id = keep_alive_packet.id;
-    
+
     state.universe.add_component::<KeepAlive>(conn_id, id)?;
 
-    
+
     Ok(())
 }

src/bin/src/systems/definition.rs

@@ -0,0 +1,49 @@
+use ferrumc_net::{GlobalState, NetResult};
+use futures::stream::FuturesUnordered;
+use tracing::{debug, debug_span, info, Instrument};
+use async_trait::async_trait;
+use crate::systems::keep_alive_system::KeepAliveSystem;
+use crate::systems::tcp_listener_system::TcpListenerSystem;
+
+#[async_trait]
+pub trait System: Send + Sync {
+    async fn start(&self, state: GlobalState);
+    async fn stop(&self, state: GlobalState);
+
+    fn name(&self) -> &'static str;
+}
+
+pub static ALL_SYSTEMS: &[&dyn System] = &[
+    &TcpListenerSystem,
+    &KeepAliveSystem
+];
+
+pub async fn start_all_systems(state: GlobalState) -> NetResult<()> {
+    let handles = FuturesUnordered::new();
+
+    for system in ALL_SYSTEMS {
+        let name = system.name();
+
+        let handle = tokio::spawn(
+            system
+                .start(state.clone())
+                .instrument(debug_span!("sys", %name)),
+        );
+        handles.push(handle);
+    }
+
+    futures::future::join_all(handles).await;
+
+    Ok(())
+}
+
+pub async fn stop_all_systems(state: GlobalState) -> NetResult<()> {
+    info!("Stopping all systems...");
+
+    for system in ALL_SYSTEMS {
+        debug!("Stopping system: {}", system.name());
+        system.stop(state.clone()).await;
+    }
+
+    Ok(())
+}

src/bin/src/systems/keep_alive_system.rs

@@ -0,0 +1,37 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use async_trait::async_trait;
+use ferrumc_core::identity::player_identity::PlayerIdentity;
+use ferrumc_net::GlobalState;
+use crate::systems::definition::System;
+
+pub struct KeepAliveSystem;
+
+static KILLED: AtomicBool = AtomicBool::new(false);
+
+#[async_trait]
+impl System for KeepAliveSystem {
+    async fn start(&self, state: GlobalState) {
+        loop {
+            if KILLED.load(Ordering::Relaxed) {
+                break;
+            }
+
+            let online_players = state.universe.query::<&PlayerIdentity>();
+            let online_players = online_players.into_iter().map(|player| player.username.clone()).collect::<Vec<String>>();
+            tracing::debug!("Online players: {:?}", online_players);
+            tracing::debug!("Total of {} online players", online_players.len());
+
+            tokio::time::sleep(Duration::from_secs(5)).await;
+        }
+    }
+
+    async fn stop(&self, _state: GlobalState) {
+        tracing::debug!("Stopping keep alive system...");
+        KILLED.store(true, Ordering::Relaxed);
+    }
+
+    fn name(&self) -> &'static str {
+        "keep_alive"
+    }
+}

src/bin/src/systems/mod.rs

@@ -0,0 +1,4 @@
+pub(crate) mod definition;
+
+mod tcp_listener_system;
+mod keep_alive_system;

src/bin/src/systems/tcp_listener_system.rs

@@ -0,0 +1,44 @@
+use std::sync::Arc;
+use async_trait::async_trait;
+use tracing::{debug, error, info, info_span, Instrument};
+use ferrumc_net::connection::handle_connection;
+use ferrumc_net::GlobalState;
+use crate::systems::definition::System;
+use crate::Result;
+
+pub struct TcpListenerSystem;
+
+#[async_trait]
+impl System for TcpListenerSystem {
+    async fn start(&self, state: GlobalState) {
+        if let Err(e) = TcpListenerSystem::initiate_loop(state).await {
+            error!("TCP listener system failed with error: {:?}", e);
+        }
+    }
+
+    async fn stop(&self, _state: GlobalState) {
+        debug!("Stopping TCP listener system...");
+    }
+
+    fn name(&self) -> &'static str {
+        "tcp"
+    }
+}
+
+impl TcpListenerSystem {
+    async fn initiate_loop(state: GlobalState) -> Result<()> {
+        let tcp_listener = &state.tcp_listener;
+        info!("Server is listening on [{}]", tcp_listener.local_addr()?);
+
+
+        loop {
+            let (stream, _) = tcp_listener.accept().await?;
+            let addy = stream.peer_addr()?;
+            debug!("Accepted connection from: {}", addy);
+            tokio::task::spawn(
+                handle_connection(Arc::clone(&state), stream)
+                    .instrument(info_span!("conn", %addy).or_current())
+            );
+        }
+    }
+}

src/lib/core/src/identity/player_identity.rs

@@ -2,4 +2,10 @@
 pub struct PlayerIdentity {
     pub username: String,
     pub uuid: u128,
+}
+
+impl PlayerIdentity {
+    pub fn new(username: String, uuid: u128) -> Self {
+        Self { username, uuid }
+    }
 }

src/lib/derive_macros/src/net/packets/mod.rs

@@ -131,7 +131,7 @@ pub fn bake_registry(input: TokenStream) -> TokenStream {
         pub async fn handle_packet<R: std::io::Read>(packet_id: u8, conn_id: usize, conn_state: &crate::connection::ConnectionState, cursor: &mut R, state: std::sync::Arc<crate::ServerState>) -> crate::NetResult<()> {
             match (packet_id, conn_state.as_str()) {
                 #(#match_arms)*
-                _ => tracing::warn!("No packet found for ID: 0x{:02X} in state: {}", packet_id, conn_state.as_str()),
+                _ => tracing::debug!("No packet found for ID: 0x{:02X} in state: {}", packet_id, conn_state.as_str()),
             }
 
             Ok(())

src/lib/ecs/src/components/mod.rs

@@ -331,7 +331,10 @@ impl ComponentManager {
 
     pub fn get_entities_with<T: Component>(&self) -> Vec<usize> {
         let type_id = TypeId::of::<T>();
-        let ptr = *self.components.get(&type_id).unwrap();
+        let Some(ptr) = self.components.get(&type_id) else {
+            return Vec::new();
+        };
+        let ptr = *ptr;
         let component_set = unsafe { &*(ptr as *const ComponentSparseSet<T>) };
         component_set.entities()
     }

src/lib/ecs/src/lib.rs

@@ -64,7 +64,6 @@ impl Universe {
     }
 
     pub fn query<Q: QueryItem>(&self) -> Query<Q> {
-        /*Query::new(&self.components)*/
-        unimplemented!()
+        Query::new(&self.components)
     }
 }