tokio: Support RCON

pumpkin-core/src/text/mod.rs

@@ -40,6 +40,13 @@ impl<'a> TextComponent<'a> {
         }
     }
 
+    pub fn text_string(text: String) -> Self {
+        Self {
+            content: TextContent::Text { text: text.into() },
+            style: Style::default(),
+        }
+    }
+
     pub fn to_pretty_console(self) -> String {
         let style = self.style;
         let color = style.color;

pumpkin-world/src/level.rs

@@ -6,7 +6,9 @@ use rayon::prelude::*;
 use tokio::sync::mpsc;
 
 use crate::{
-    chunk::{anvil::AnvilChunkReader, ChunkData, ChunkParsingError, ChunkReader, ChunkReadingError},
+    chunk::{
+        anvil::AnvilChunkReader, ChunkData, ChunkParsingError, ChunkReader, ChunkReadingError,
+    },
     world_gen::{get_world_gen, Seed, WorldGenerator},
 };
 
@@ -96,7 +98,10 @@ impl Level {
             let data = match &self.save_file {
                 Some(save_file) => {
                     match self.chunk_reader.read_chunk(save_file, at) {
-                        Err(ChunkReadingError::ParsingError(ChunkParsingError::ChunkNotGenerated) | ChunkReadingError::ChunkNotExist) => {
+                        Err(
+                            ChunkReadingError::ParsingError(ChunkParsingError::ChunkNotGenerated)
+                            | ChunkReadingError::ChunkNotExist,
+                        ) => {
                             // This chunk was not generated yet.
                             Ok(self.world_gen.generate_chunk(at))
                         }

pumpkin/src/client/container.rs

@@ -1,6 +1,7 @@
 use crate::entity::player::Player;
 use crate::server::Server;
 use itertools::Itertools;
+use pumpkin_core::text::TextComponent;
 use pumpkin_core::GameMode;
 use pumpkin_inventory::container_click::{
     Click, ClickType, KeyClick, MouseClick, MouseDragState, MouseDragType,
@@ -10,7 +11,7 @@ use pumpkin_inventory::window_property::{WindowProperty, WindowPropertyTrait};
 use pumpkin_inventory::Container;
 use pumpkin_inventory::{container_click, InventoryError, OptionallyCombinedContainer};
 use pumpkin_protocol::client::play::{
-    CCloseContainer, CSetContainerContent, CSetContainerProperty, CSetContainerSlot,
+    CCloseContainer, COpenScreen, CSetContainerContent, CSetContainerProperty, CSetContainerSlot,
 };
 use pumpkin_protocol::server::play::SClickContainer;
 use pumpkin_protocol::slot::Slot;
@@ -26,28 +27,30 @@ impl Player {
         let total_opened_containers = inventory.total_opened_containers;
         let container = self.get_open_container(server);
         let container = container.as_ref().map(|container| container.lock());
-        // let menu_protocol_id = (*pumpkin_world::global_registry::REGISTRY
+        let menu_protocol_id = (*pumpkin_world::global_registry::REGISTRY
-        //     .get("minecraft:menu")
+            .get("minecraft:menu")
-        //     .unwrap()
+            .unwrap()
-        //     .entries
+            .entries
-        //     .get(minecraft_menu_id)
+            .get(minecraft_menu_id)
-        //     .expect("Should be a valid menu id")
+            .expect("Should be a valid menu id")
-        //     .get("protocol_id")
+            .get("protocol_id")
-        //     .unwrap())
+            .unwrap())
-        // .into();
+        .into();
-        // let window_title = container
+        let window_title = match container {
-        //     .as_ref()
+            Some(container) => container.await.window_name(),
-        //     .map(|container| container.window_name())
+            None => inventory.window_name(),
-        //     .unwrap_or_else(|| inventory.window_name());
+        };
-        // let title = TextComponent::text(window_title);
+        let title = TextComponent::text(window_title);
 
-        // self.client.send_packet(&COpenScreen::new(
+        self.client
-        //     total_opened_containers.into(),
+            .send_packet(&COpenScreen::new(
-        //     menu_protocol_id,
+                total_opened_containers.into(),
-        //     title,
+                menu_protocol_id,
-        // ));
+                title,
+            ))
+            .await;
         drop(inventory);
-        //  self.set_container_content(container.as_deref_mut());
+        // self.set_container_content(container.as_deref_mut());
     }
 
     pub async fn set_container_content(&self, container: Option<&mut Box<dyn Container>>) {

pumpkin/src/commands/cmd_gamemode.rs

@@ -98,17 +98,17 @@ pub fn init_command_tree<'a>() -> CommandTree<'a> {
                         let target = parse_arg_player(sender, server, ARG_TARGET, args)?;
 
                         if target.gamemode.load() == gamemode {
-                            sender.send_message(TextComponent::text(&format!(
+                            // sender.send_message(TextComponent::text(&format!(
-                                "{} is already in {:?} gamemode",
+                            //     "{} is already in {:?} gamemode",
-                                target.gameprofile.name, gamemode
+                            //     target.gameprofile.name, gamemode
-                            )));
+                            // )));
                         } else {
                             // TODO
                             target.set_gamemode(gamemode);
-                            sender.send_message(TextComponent::text(&format!(
+                            // sender.send_message(TextComponent::text(&format!(
-                                "{}'s Game mode was set to {:?}",
+                            //     "{}'s Game mode was set to {:?}",
-                                target.gameprofile.name, gamemode
+                            //     target.gameprofile.name, gamemode
-                            )));
+                            // )));
                         }
 
                         Ok(())

pumpkin/src/commands/cmd_help.rs

@@ -43,12 +43,12 @@ pub fn init_command_tree<'a>() -> CommandTree<'a> {
             argument(ARG_COMMAND, consume_arg_command).execute(&|sender, server, args| {
                 let tree = parse_arg_command(args, &server.command_dispatcher)?;
 
-                sender.send_message(TextComponent::text(&format!(
+                // sender.send_message(TextComponent::text(&format!(
-                    "{} - {} Usage: {}",
+                //     "{} - {} Usage: {}",
-                    tree.names.join("/"),
+                //     tree.names.join("/"),
-                    tree.description,
+                //     tree.description,
-                    tree
+                //     tree
-                )));
+                // )));
 
                 Ok(())
             }),
@@ -62,12 +62,12 @@ pub fn init_command_tree<'a>() -> CommandTree<'a> {
                     continue;
                 };
 
-                sender.send_message(TextComponent::text(&format!(
+                // sender.send_message(TextComponent::text(&format!(
-                    "{} - {} Usage: {}",
+                //     "{} - {} Usage: {}",
-                    tree.names.join("/"),
+                //     tree.names.join("/"),
-                    tree.description,
+                //     tree.description,
-                    tree
+                //     tree
-                )));
+                // )));
             }
 
             Ok(())

pumpkin/src/commands/cmd_pumpkin.rs

@@ -1,4 +1,7 @@
-use crate::commands::tree::CommandTree;
+use pumpkin_core::text::{color::NamedColor, TextComponent};
+use pumpkin_protocol::CURRENT_MC_PROTOCOL;
+
+use crate::{commands::tree::CommandTree, server::CURRENT_MC_VERSION};
 
 const NAMES: [&str; 1] = ["pumpkin"];
 
@@ -11,7 +14,7 @@ pub fn init_command_tree<'a>() -> CommandTree<'a> {
 
         // sender.send_message(TextComponent::text(
         //     &format!("Pumpkin {version}, {description} (Minecraft {CURRENT_MC_VERSION}, Protocol {CURRENT_MC_PROTOCOL})")
-        // ).color_named(NamedColor::Green));
+        // ).color_named(NamedColor::Green)).await;
 
         Ok(())
     })

pumpkin/src/commands/dispatcher.rs

@@ -1,3 +1,5 @@
+use pumpkin_core::text::TextComponent;
+
 use crate::commands::dispatcher::InvalidTreeError::{
     InvalidConsumptionError, InvalidRequirementError,
 };
@@ -24,11 +26,14 @@ pub struct CommandDispatcher<'a> {
 
 /// Stores registered [CommandTree]s and dispatches commands to them.
 impl<'a> CommandDispatcher<'a> {
-    pub fn handle_command(&self, sender: &mut CommandSender, server: &Server, cmd: &str) {
+    pub async fn handle_command(&self, sender: &mut CommandSender<'a>, server: &Server, cmd: &str) {
         if let Err(err) = self.dispatch(sender, server, cmd) {
-            // sender.send_message(
+            sender
-            //     TextComponent::text(&err).color_named(pumpkin_core::text::color::NamedColor::Red),
+                .send_message(
-            // )
+                    TextComponent::text_string(err)
+                        .color_named(pumpkin_core::text::color::NamedColor::Red),
+                )
+                .await;
         }
     }
 

pumpkin/src/commands/mod.rs

@@ -27,11 +27,10 @@ pub enum CommandSender<'a> {
 }
 
 impl<'a> CommandSender<'a> {
-    pub fn send_message(&mut self, text: TextComponent) {
+    pub async fn send_message(&mut self, text: TextComponent<'a>) {
         match self {
-            // TODO: add color and stuff to console
             CommandSender::Console => log::info!("{}", text.to_pretty_console()),
-            CommandSender::Player(c) => c.send_system_message(&text),
+            CommandSender::Player(c) => c.send_system_message(&text).await,
             CommandSender::Rcon(s) => s.push(text.to_pretty_console()),
         }
     }

pumpkin/src/entity/player.rs

@@ -276,7 +276,7 @@ impl Player {
             .await;
     }
 
-    pub async fn send_system_message<'a>(&self, text: TextComponent<'a>) {
+    pub async fn send_system_message<'a>(&self, text: &TextComponent<'a>) {
         self.client
             .send_packet(&CSystemChatMessage::new(text, false))
             .await;
@@ -284,7 +284,7 @@ impl Player {
 }
 
 impl Player {
-    pub async fn process_packets(&self, server: &Arc<Server>) {
+    pub async fn process_packets(self: &Arc<Self>, server: &Arc<Server>) {
         let mut packets = self.client.client_packets_queue.lock().await;
         while let Some(mut packet) = packets.pop_back() {
             match self.handle_play_packet(server, &mut packet).await {
@@ -308,7 +308,7 @@ impl Player {
     }
 
     pub async fn handle_play_packet(
-        &self,
+        self: &Arc<Self>,
         server: &Arc<Server>,
         packet: &mut RawPacket,
     ) -> Result<(), Box<dyn PumpkinError>> {

pumpkin/src/main.rs

@@ -11,7 +11,6 @@
 #![expect(clippy::significant_drop_tightening)]
 #![expect(clippy::future_not_send)]
 #![expect(clippy::single_call_fn)]
-#![expect(clippy::await_holding_lock)]
 
 #[cfg(target_os = "wasi")]
 compile_error!("Compiling for WASI targets is not supported!");
@@ -150,7 +149,7 @@ async fn main() -> io::Result<()> {
     if rcon.enabled {
         let server = server.clone();
         tokio::spawn(async move {
-            RCONServer::new(&rcon, &server).await.unwrap();
+            RCONServer::new(&rcon, server).await.unwrap();
         });
     }
     let mut unique_id = 0;

pumpkin/src/rcon/mod.rs

@@ -1,11 +1,11 @@
 use std::{
-    collections::HashMap,
     io::{self},
     net::SocketAddr,
 };
 
 use packet::{ClientboundPacket, Packet, PacketError, ServerboundPacket};
 use pumpkin_config::{RCONConfig, ADVANCED_CONFIG};
+use std::sync::Arc;
 use thiserror::Error;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
@@ -26,40 +26,29 @@ pub enum RCONError {
 pub struct RCONServer;
 
 impl RCONServer {
-    pub async fn new(config: &RCONConfig, server: &Server) -> Result<Self, io::Error> {
+    pub async fn new(config: &RCONConfig, server: Arc<Server>) -> Result<Self, io::Error> {
         assert!(config.enabled, "RCON is not enabled");
         let listener = tokio::net::TcpListener::bind(config.address).await.unwrap();
 
-        let mut connections: HashMap<usize, RCONClient> = HashMap::new();
+        let password = Arc::new(config.password.clone());
 
-        let password = config.password.clone();
+        let mut connections = 0;
-
-        let mut unique_id = 0;
         loop {
             // Asynchronously wait for an inbound socket.
             let (connection, address) = listener.accept().await?;
 
-            if config.max_connections != 0 && connections.len() >= config.max_connections as usize {
+            if config.max_connections != 0 && connections >= config.max_connections {
                 continue;
             }
 
-            unique_id += 1;
+            connections += 1;
-            let token = unique_id;
+            let mut client = RCONClient::new(connection, address);
-            connections.insert(token, RCONClient::new(connection, address));
 
-            let done = if let Some(client) = connections.get_mut(&token) {
+            let password = password.clone();
-                client.handle(server, &password).await
+            let server = server.clone();
-            } else {
+            tokio::spawn(async move { while !client.handle(&server, &password).await {} });
-                false
+            dbg!("closed");
-            };
+            connections -= 1;
-            if done {
-                if let Some(client) = connections.remove(&token) {
-                    let config = &ADVANCED_CONFIG.rcon;
-                    if config.logging.log_quit {
-                        log::info!("RCON ({}): Client closed connection", client.address);
-                    }
-                }
-            }
         }
     }
 }
@@ -83,20 +72,19 @@ impl RCONClient {
         }
     }
 
+    /// Returns if client is closed or not
     pub async fn handle(&mut self, server: &Server, password: &str) -> bool {
         if !self.closed {
-            loop {
+            match self.read_bytes().await {
-                match self.read_bytes().await {
+                // Stream closed, so we can't reply, so we just close everything.
-                    // Stream closed, so we can't reply, so we just close everything.
+                Ok(true) => return true,
-                    Ok(true) => return true,
+                Ok(false) => {}
-                    Ok(false) => {}
+                Err(e) => {
-                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => break,
+                    log::error!("could not read packet: {e}");
-                    Err(e) => {
+                    return true;
-                        log::error!("could not read packet: {e}");
-                        return true;
-                    }
                 }
             }
+            dbg!("a");
             // If we get a close here, we might have a reply, which we still want to write.
             let _ = self.poll(server, password).await.map_err(|e| {
                 log::error!("RCON error: {e}");
@@ -107,52 +95,51 @@ impl RCONClient {
     }
 
     async fn poll(&mut self, server: &Server, password: &str) -> Result<(), PacketError> {
-        loop {
+        let packet = match self.receive_packet().await? {
-            let packet = match self.receive_packet().await? {
+            Some(p) => p,
-                Some(p) => p,
+            None => return Ok(()),
-                None => return Ok(()),
+        };
-            };
+        let config = &ADVANCED_CONFIG.rcon;
-
+        match packet.get_type() {
-            let config = &ADVANCED_CONFIG.rcon;
+            ServerboundPacket::Auth => {
-            match packet.get_type() {
+                if packet.get_body() == password {
-                ServerboundPacket::Auth => {
+                    self.send(ClientboundPacket::AuthResponse, packet.get_id(), "".into())
-                    let body = packet.get_body();
+                        .await?;
-                    if !body.is_empty() && packet.get_body() == password {
+                    if config.logging.log_logged_successfully {
-                        self.send(ClientboundPacket::AuthResponse, packet.get_id(), "".into())
+                        log::info!("RCON ({}): Client logged in successfully", self.address);
-                            .await?;
-                        if config.logging.log_logged_successfully {
-                            log::info!("RCON ({}): Client logged in successfully", self.address);
-                        }
-                        self.logged_in = true;
-                    } else {
-                        if config.logging.log_wrong_password {
-                            log::info!("RCON ({}): Client has tried wrong password", self.address);
-                        }
-                        self.send(ClientboundPacket::AuthResponse, -1, "".into())
-                            .await?;
-                        self.closed = true;
                     }
+                    self.logged_in = true;
+                } else {
+                    if config.logging.log_wrong_password {
+                        log::info!("RCON ({}): Client has tried wrong password", self.address);
+                    }
+                    self.send(ClientboundPacket::AuthResponse, -1, "".into())
+                        .await?;
+                    self.closed = true;
                 }
-                ServerboundPacket::ExecCommand => {
+            }
-                    if self.logged_in {
+            ServerboundPacket::ExecCommand => {
-                        let mut output = Vec::new();
+                if self.logged_in {
-                        let dispatcher = server.command_dispatcher.clone();
+                    let mut output = Vec::new();
-                        dispatcher.handle_command(
+                    let dispatcher = server.command_dispatcher.clone();
+                    dispatcher
+                        .handle_command(
                             &mut crate::commands::CommandSender::Rcon(&mut output),
                             server,
                             packet.get_body(),
-                        );
+                        )
-                        for line in output {
+                        .await;
-                            if config.logging.log_commands {
+                    for line in output {
-                                log::info!("RCON ({}): {}", self.address, line);
+                        if config.logging.log_commands {
-                            }
+                            log::info!("RCON ({}): {}", self.address, line);
-                            self.send(ClientboundPacket::Output, packet.get_id(), line)
-                                .await?;
                         }
+                        self.send(ClientboundPacket::Output, packet.get_id(), line)
+                            .await?;
                     }
                 }
             }
         }
+        Ok(())
     }
 
     async fn read_bytes(&mut self) -> io::Result<bool> {

pumpkin/src/server/mod.rs

@@ -131,7 +131,7 @@ impl Server {
     /// Searches every world for a player by name
     pub async fn get_player_by_name(&self, name: &str) -> Option<Arc<Player>> {
         for world in self.worlds.iter() {
-            if let Some(player) = world.get_player_by_name(name).await {
+            if let Some(player) = world.get_player_by_name(name) {
                 return Some(player);
             }
         }

pumpkin/src/world/mod.rs

@@ -298,8 +298,9 @@ impl World {
     }
 
     /// Gets a Player by name
-    pub async fn get_player_by_name(&self, name: &str) -> Option<Arc<Player>> {
+    pub fn get_player_by_name(&self, name: &str) -> Option<Arc<Player>> {
-        for player in self.current_players.lock().await.values() {
+        // not sure of blocking lock
+        for player in self.current_players.blocking_lock().values() {
             if player.gameprofile.name == name {
                 return Some(player.clone());
             }

pumpkin/src/world/player_chunker.rs

@@ -78,11 +78,10 @@ pub async fn update_position(entity: &Entity, player: &Player) {
             Vector2::new(current_watched.x, current_watched.z),
             view_distance,
         );
-        let new_cylindrical =
+        let new_cylindrical = Cylindrical::new(chunk_pos, view_distance);
-            Cylindrical::new(chunk_pos, view_distance);
 
         player.watched_section.store(new_watched);
-        
+
         let mut loading_chunks = Vec::new();
         Cylindrical::for_each_changed_chunk(
             old_cylindrical,

pumpkin/src/world/scoreboard.rs

@@ -26,7 +26,7 @@ impl Scoreboard {
         }
     }
 
-    pub fn add_objective(&mut self, world: &World, objective: ScoreboardObjective) {
+    pub async fn add_objective<'a>(&mut self, world: &World, objective: ScoreboardObjective<'a>) {
         if self.objectives.contains_key(objective.name) {
             // Maybe make this an error ?
             log::warn!(
@@ -35,20 +35,24 @@ impl Scoreboard {
             );
             return;
         }
-        world.broadcast_packet_all(&CUpdateObjectives::new(
+        world
-            objective.name,
+            .broadcast_packet_all(&CUpdateObjectives::new(
-            pumpkin_protocol::client::play::Mode::Add,
+                objective.name,
-            objective.display_name,
+                pumpkin_protocol::client::play::Mode::Add,
-            objective.render_type,
+                objective.display_name,
-            objective.number_format,
+                objective.render_type,
-        ));
+                objective.number_format,
-        world.broadcast_packet_all(&CDisplayObjective::new(
+            ))
-            pumpkin_protocol::client::play::DisplaySlot::Sidebar,
+            .await;
-            "test",
+        world
-        ));
+            .broadcast_packet_all(&CDisplayObjective::new(
+                pumpkin_protocol::client::play::DisplaySlot::Sidebar,
+                "test",
+            ))
+            .await;
     }
 
-    pub fn update_score(&self, world: &World, score: ScoreboardScore) {
+    pub async fn update_score<'a>(&self, world: &World, score: ScoreboardScore<'a>) {
         if self.objectives.contains_key(score.objective_name) {
             log::warn!(
                 "Tried to place a score into a Objective which does not exist, {}",
@@ -56,13 +60,15 @@ impl Scoreboard {
             );
             return;
         }
-        world.broadcast_packet_all(&CUpdateScore::new(
+        world
-            score.entity_name,
+            .broadcast_packet_all(&CUpdateScore::new(
-            score.objective_name,
+                score.entity_name,
-            score.value,
+                score.objective_name,
-            score.display_name,
+                score.value,
-            score.number_format,
+                score.display_name,
-        ));
+                score.number_format,
+            ))
+            .await;
     }
 
     // pub fn add_team(&mut self, name: String) {