feat(player): packet response loop after login

Author: andybrwr

src/net/io.rs

@@ -20,7 +20,7 @@
 use std::{io::ErrorKind, net::SocketAddr, time::Duration};
 
 use bytes::BytesMut;
-use color_eyre::eyre::{Context, Result};
+use color_eyre::eyre::{bail, Context, Result};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::TcpStream,
@@ -132,4 +132,31 @@ impl NetIo {
         self.stream.flush().await?;
         Ok(())
     }
+
+    pub async fn rx_raw(&mut self) -> Result<Frame> {
+        if let Some(frame) = self
+            .decoder
+            .try_read_next()
+            .context("failed try_read_next")?
+        {
+            return Ok(frame);
+        };
+
+        self.decoder.reserve_additional(BUF_SIZE);
+        let mut buf = self.decoder.take_all();
+
+        if self
+            .stream
+            .read_buf(&mut buf)
+            .await
+            .context("failed read_buf")?
+            == 0
+        {
+            return Err(std::io::Error::from(ErrorKind::UnexpectedEof).into());
+        }
+
+        self.decoder.add_bytes(buf);
+
+        bail!("No packet available")
+    }
 }

src/net/player.rs

@@ -26,7 +26,7 @@ use thiserror::Error;
 use tokio::{
     net::TcpStream,
     sync::{Mutex, OwnedSemaphorePermit, RwLock},
-    time::{self, timeout},
+    time::{self, timeout, Instant},
 };
 use uuid::Uuid;
 
@@ -38,10 +38,11 @@ use crate::{
             play::{
                 ConfirmTeleportS, GameEvent, GameEventC, Gamemode, KeepAliveC, LoginPlayC,
                 PlayerInfoUpdateC, PlayerStatus, SetBorderCenterC, SetBorderSizeC, SetCenterChunkC,
-                SetTickingStateC, StepTicksC, SynchronisePositionC,
+                SetPlayerPositionAndRotationS, SetPlayerPositionS, SetTickingStateC, StepTicksC,
+                SynchronisePositionC,
             },
         },
-        PacketState,
+        Frame, Packet, PacketState,
     },
     CrawlState,
 };
@@ -62,6 +63,8 @@ pub struct Player {
 
     uuid: RwLock<Option<Uuid>>,
     tp_state: Mutex<TeleportState>,
+
+    last_keepalive: RwLock<Instant>,
 }
 
 #[derive(Debug)]
@@ -85,12 +88,14 @@ impl SharedPlayer {
             id,
             io: Mutex::new(NetIo::new(connection)),
             _permit: permit,
-            uuid: RwLock::new(None),
 
             crawlstate,
             packet_state: RwLock::new(PacketState::Handshaking),
 
             tp_state: Mutex::new(TeleportState::Clear),
+            uuid: RwLock::new(None),
+
+            last_keepalive: RwLock::new(Instant::now()),
         }))
     }
 
@@ -333,20 +338,27 @@ impl SharedPlayer {
         // thread but realistically who knows burhhhh
         state.player_send.send(self.clone()).await?;
 
-        loop {
-            tokio::select! {
-                _ = self.keepalive() => {
-                    // keepalive needs to be sent every ~15 sec after keepalive response
-                    time::sleep(Duration::from_secs(12)).await;
-                }
-                _ = state.shutdown_token.cancelled() => {
-                    return Ok(());
-                }
-            }
-        }
+        state.shutdown_token.cancelled().await;
+        Ok(())
+    }
+
+    pub async fn handle_all_packets(&self) -> Result<()> {
+        let packets = self.rx_all().await;
+        self.handle_frames(packets).await
     }
 
-    async fn keepalive(&self) -> Result<()> {
+    pub async fn keepalive(&self) -> Result<()> {
+        let last_keepalive = self.0.last_keepalive.read().await;
+        let now = Instant::now();
+
+        if now - *last_keepalive < Duration::from_secs(10) {
+            return Ok(());
+        }
+
+        drop(last_keepalive);
+        let mut last_keepalive = self.0.last_keepalive.write().await;
+        *last_keepalive = now;
+
         let id = {
             let mut rng = rand::thread_rng();
             rng.gen()
@@ -359,6 +371,17 @@ impl SharedPlayer {
         }
     }
 
+    async fn rx_all(&self) -> Vec<Frame> {
+        let mut io = self.0.io.lock().await;
+
+        let mut frames = Vec::new();
+        while let Ok(frame) = io.rx_raw().await {
+            frames.push(frame);
+        }
+
+        frames
+    }
+
     async fn ping(&self, id: i64) -> Result<()> {
         let mut io = self.0.io.lock().await;
         io.flush().await?;
@@ -417,6 +440,37 @@ impl SharedPlayer {
         }
         Ok(())
     }
+
+    async fn handle_frames(&self, frames: Vec<Frame>) -> Result<()> {
+        for frame in frames {
+            match frame.id {
+                SetPlayerPositionS::ID => {
+                    let packet: SetPlayerPositionS = frame.decode()?;
+                    debug!(
+                        "Player {} moved to {}, {}, {}",
+                        self.0.id, packet.x, packet.feet_y, packet.z
+                    );
+                }
+
+                SetPlayerPositionAndRotationS::ID => {
+                    let packet: SetPlayerPositionAndRotationS = frame.decode()?;
+                    debug!(
+                        "Player {} moved to {}, {}, {} rotated {} {}",
+                        self.0.id, packet.x, packet.feet_y, packet.z, packet.pitch, packet.yaw
+                    );
+                }
+
+                id => {
+                    debug!(
+                        "Got packet with id {id} from player {}, ignoring",
+                        self.0.id
+                    );
+                }
+            }
+        }
+
+        Ok(())
+    }
 }
 
 #[derive(Debug, Error)]

src/protocol/mod.rs

@@ -49,6 +49,7 @@ pub mod packets {
         mod game_event;
         mod keepalive;
         mod login;
+        mod position;
         mod status;
         mod teleport;
         mod tick;
@@ -57,6 +58,7 @@ pub mod packets {
         pub use game_event::*;
         pub use keepalive::*;
         pub use login::*;
+        pub use position::*;
         pub use status::*;
         pub use teleport::*;
         pub use tick::*;

src/protocol/packets/play/position.rs

@@ -0,0 +1,78 @@
+/*
+ * Copyright (c) 2024 Andrew Brower.
+ * This file is part of Crawlspace.
+ *
+ * Crawlspace is free software: you can redistribute it and/or
+ * modify it under the terms of the GNU Affero General Public
+ * License as published by the Free Software Foundation, either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * Crawlspace is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public
+ * License along with Crawlspace. If not, see
+ * <https://www.gnu.org/licenses/>.
+ */
+
+use byteorder::{BigEndian, ReadBytesExt};
+
+use crate::protocol::{Decode, Packet};
+
+#[derive(Debug)]
+pub struct SetPlayerPositionS {
+    pub x: f64,
+    pub feet_y: f64,
+    pub z: f64,
+    pub on_ground: bool,
+}
+
+impl Packet for SetPlayerPositionS {
+    const ID: i32 = 0x1A;
+}
+
+impl Decode<'_> for SetPlayerPositionS {
+    fn decode(r: &mut &'_ [u8]) -> color_eyre::eyre::Result<Self>
+    where
+        Self: Sized,
+    {
+        Ok(Self {
+            x: r.read_f64::<BigEndian>()?,
+            feet_y: r.read_f64::<BigEndian>()?,
+            z: r.read_f64::<BigEndian>()?,
+            on_ground: r.read_u8()? == 1,
+        })
+    }
+}
+
+#[derive(Debug)]
+pub struct SetPlayerPositionAndRotationS {
+    pub x: f64,
+    pub feet_y: f64,
+    pub z: f64,
+    pub yaw: f32,
+    pub pitch: f32,
+    pub on_ground: bool,
+}
+
+impl Packet for SetPlayerPositionAndRotationS {
+    const ID: i32 = 0x1B;
+}
+
+impl Decode<'_> for SetPlayerPositionAndRotationS {
+    fn decode(r: &mut &'_ [u8]) -> color_eyre::eyre::Result<Self>
+    where
+        Self: Sized,
+    {
+        Ok(Self {
+            x: r.read_f64::<BigEndian>()?,
+            feet_y: r.read_f64::<BigEndian>()?,
+            z: r.read_f64::<BigEndian>()?,
+            yaw: r.read_f32::<BigEndian>()?,
+            pitch: r.read_f32::<BigEndian>()?,
+            on_ground: r.read_u8()? == 1,
+        })
+    }
+}

src/protocol/packets/play/world.rs

@@ -92,11 +92,58 @@ impl From<world::BlockEntity> for BlockEntity {
         let data = fastnbt::to_bytes_with_opts(&value.raw_data, fastnbt::SerOpts::network_nbt())
             .expect("Failed to parse network nbt for block entity");
 
+        let kind = VarInt(match value.id.as_str() {
+            "minecraft:furnace" => 0,
+            "minecraft:chest" => 1,
+            "minecraft:trapped_chest" => 2,
+            "minecraft:ender_chest" => 3,
+            "minecraft:jukebox" => 4,
+            "minecraft:dispenser" => 5,
+            "minecraft:dropper" => 6,
+            "minecraft:sign" => 7,
+            "minecraft:hanging_sign" => 8,
+            "minecraft:mob_spawner" => 9,
+            "minecraft:piston" => 10,
+            "minecraft:brewing_stand" => 11,
+            "minecraft:enchanting_table" => 12,
+            "minecraft:end_portal" => 13,
+            "minecraft:beacon" => 14,
+            "minecraft:skull" => 15,
+            "minecraft:daylight_detector" => 16,
+            "minecraft:hopper" => 17,
+            "minecraft:comparator" => 18,
+            "minecraft:banner" => 19,
+            "minecraft:structure_block" => 20,
+            "minecraft:end_gateway" => 21,
+            "minecraft:command_block" => 22,
+            "minecraft:shulker_box" => 23,
+            "minecraft:bed" => 24,
+            "minecraft:conduit" => 25,
+            "minecraft:barrel" => 26,
+            "minecraft:smoker" => 27,
+            "minecraft:blast_furnace" => 28,
+            "minecraft:lectern" => 29,
+            "minecraft:bell" => 30,
+            "minecraft:jigsaw" => 31,
+            "minecraft:campfire" => 32,
+            "minecraft:beehive" => 33,
+            "minecraft:sculk_sensor" => 34,
+            "minecraft:calibrated_sculk_sensor" => 35,
+            "minecraft:sculk_catalyst" => 36,
+            "minecraft:sculk_shrieker" => 37,
+            "minecraft:chiseled_bookshelf" => 38,
+            "minecraft:brushable_block" => 39,
+            "minecraft:decorated_pot" => 40,
+            "minecraft:crafter" => 41,
+            "minecraft:trial_spawner" => 42,
+            "minecraft:vault" => 43,
+            i => unimplemented!("Block Entity type {i} is unimplemented"),
+        });
+
         Self {
             packed_xz: (((value.x & 15) << 4) | (value.z & 15)) as u8,
             y: value.y as i16,
-            // FIXME: use correct block entity type from registry
-            kind: VarInt(7),
+            kind,
             data,
         }
     }

src/server/mod.rs

@@ -59,6 +59,11 @@ impl Server {
             self.players.push(p.clone());
             tokio::spawn(Self::send_world_to(p.clone(), self.world_cache.clone()));
         }
+
+        for player in &self.players {
+            let _ = player.keepalive().await;
+            let _ = player.handle_all_packets().await;
+        }
     }
 
     async fn send_world_to(player: SharedPlayer, world_cache: Arc<WorldCache>) -> Result<()> {