chore: don't spam so many packets

also use CancellationToken instead of Notify in some places

pumpkin/src/entity/living.rs

@@ -1161,8 +1161,8 @@ impl EntityBase for LivingEntity {
             self.tick_movement(server, caller.clone()).await;
             // TODO
             if caller.get_player().is_none() {
-                self.entity.send_pos_rot().await;
+                // self.entity.send_pos_rot().await;
-                self.entity.send_velocity().await;
+                // self.entity.send_velocity().await;
             }
             self.tick_effects().await;
             // Current active item

pumpkin/src/entity/mob/mod.rs

@@ -75,22 +75,22 @@ impl MobEntity {
     }
 
     async fn set_mob_flag(&self, flag: u8, value: bool) {
-        let index = flag;
+        let old_b = self.mob_flags.load(Ordering::Relaxed);
-        let mut b = self.mob_flags.load(Ordering::Relaxed);
+
-        if value {
+        let new_b = if value { old_b | flag } else { old_b & !flag };
-            b |= index;
+
-        } else {
+        if new_b != old_b {
-            b &= !index;
+            self.mob_flags.store(new_b, Ordering::Relaxed);
+
+            self.living_entity
+                .entity
+                .send_meta_data(&[Metadata::new(
+                    TrackedData::DATA_MOB_FLAGS,
+                    MetaDataType::Byte,
+                    new_b,
+                )])
+                .await;
         }
-        self.mob_flags.store(b, Ordering::Relaxed);
-        self.living_entity
-            .entity
-            .send_meta_data(&[Metadata::new(
-                TrackedData::DATA_MOB_FLAGS,
-                MetaDataType::Byte,
-                b,
-            )])
-            .await;
     }
 }
 

pumpkin/src/lib.rs

@@ -28,8 +28,9 @@ use std::time::Duration;
 use std::{net::SocketAddr, sync::LazyLock};
 use tokio::net::{TcpListener, UdpSocket};
 use tokio::select;
-use tokio::sync::{Mutex, Notify, RwLock};
+use tokio::sync::{Mutex, RwLock};
 use tokio::time::sleep;
+use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 
 pub mod block;
@@ -159,11 +160,11 @@ pub fn init_logger(advanced_config: &AdvancedConfiguration) {
 }
 
 pub static SHOULD_STOP: AtomicBool = AtomicBool::new(false);
-pub static STOP_INTERRUPT: LazyLock<Notify> = LazyLock::new(Notify::new);
+pub static STOP_INTERRUPT: LazyLock<CancellationToken> = LazyLock::new(CancellationToken::new);
 
 pub fn stop_server() {
     SHOULD_STOP.store(true, Ordering::Relaxed);
-    STOP_INTERRUPT.notify_waiters();
+    STOP_INTERRUPT.cancel();
 }
 
 fn resolve_some<T: Future, D, F: FnOnce(D) -> T>(
@@ -492,7 +493,7 @@ impl PumpkinServer {
             },
 
             // Branch for the global stop signal
-            () = STOP_INTERRUPT.notified() => {
+            () = STOP_INTERRUPT.cancelled() => {
                 return false;
             }
         }
@@ -583,7 +584,7 @@ fn setup_console(mut rl: Editor<PumpkinCommandCompleter, FileHistory>, server: A
     server.clone().spawn_task(async move {
         while !SHOULD_STOP.load(Ordering::Relaxed) {
             let t1 = rx.recv();
-            let t2 = STOP_INTERRUPT.notified();
+            let t2 = STOP_INTERRUPT.cancelled();
 
             let result = select! {
                 line = t1 => line,

pumpkin/src/net/java/mod.rs

@@ -37,7 +37,6 @@ use pumpkin_protocol::{
     ser::{NetworkWriteExt, ReadingError, WritingError},
 };
 use pumpkin_util::text::TextComponent;
-use tokio::sync::Notify;
 use tokio::{
     io::{BufReader, BufWriter},
     net::{
@@ -50,6 +49,7 @@ use tokio::{
     sync::mpsc::{Receiver, Sender},
     task::JoinHandle,
 };
+use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 
 pub mod config;
@@ -82,7 +82,7 @@ pub struct JavaClient {
     /// A collection of tasks associated with this client. The tasks await completion when removing the client.
     tasks: TaskTracker,
     /// An notifier that is triggered when this client is closed.
-    close_interrupt: Arc<Notify>,
+    close_token: CancellationToken,
     /// A queue of serialized packets to send to the network
     outgoing_packet_queue_send: Sender<Bytes>,
     /// A queue of serialized packets to send to the network
@@ -106,7 +106,7 @@ impl JavaClient {
             address: Mutex::new(address),
             connection_state: AtomicCell::new(ConnectionState::HandShake),
             closed: Arc::new(AtomicBool::new(false)),
-            close_interrupt: Arc::new(Notify::new()),
+            close_token: CancellationToken::new(),
             tasks: TaskTracker::new(),
             outgoing_packet_queue_send: send,
             outgoing_packet_queue_recv: Some(recv),
@@ -221,7 +221,7 @@ impl JavaClient {
     }
 
     pub async fn await_close_interrupt(&self) {
-        self.close_interrupt.notified().await;
+        self.close_token.cancelled().await;
     }
 
     pub async fn get_packet(&self) -> Option<RawPacket> {
@@ -406,19 +406,15 @@ impl JavaClient {
             .outgoing_packet_queue_recv
             .take()
             .expect("This was set in the new fn");
-        let close_interrupt = self.close_interrupt.clone();
+        let close_interrupt = self.close_token.clone();
         let closed = self.closed.clone();
         let writer = self.network_writer.clone();
         let id = self.id;
         self.spawn_task(async move {
             while !closed.load(Ordering::Relaxed) {
                 let recv_result = tokio::select! {
-                    () = close_interrupt.notified() => {
+                    () =  close_interrupt.cancelled() => None,
-                        None
+                    res = packet_receiver.recv() => res,
-                    },
-                    recv_result = packet_receiver.recv() => {
-                        recv_result
-                    }
                 };
 
                 let Some(packet_data) = recv_result else {
@@ -431,7 +427,7 @@ impl JavaClient {
                         log::warn!("Failed to send packet to client {id}: {err}",);
                         // We now need to close the connection to the client since the stream is in an
                         // unknown state
-                        close_interrupt.notify_waiters();
+                        close_interrupt.cancel();
                         closed.store(true, Ordering::Relaxed);
                         break;
                     }
@@ -450,7 +446,7 @@ impl JavaClient {
     ///
     /// This function does not attempt to send any disconnect packets to the client.
     pub fn close(&self) {
-        self.close_interrupt.notify_waiters();
+        self.close_token.cancel();
         self.closed.store(true, Ordering::Relaxed);
     }
 

pumpkin/src/net/lan_broadcast.rs

@@ -56,7 +56,7 @@ impl LANBroadcast {
 
         while !SHOULD_STOP.load(Ordering::Relaxed) {
             let t1 = interval.tick();
-            let t2 = STOP_INTERRUPT.notified();
+            let t2 = STOP_INTERRUPT.cancelled();
 
             let should_continue = select! {
                 _ = t1 => true,

pumpkin/src/net/query.rs

@@ -51,7 +51,7 @@ pub async fn start_query_handler(server: Arc<Server>, query_addr: SocketAddr) {
 
         let recv_result = tokio::select! {
             result = socket.recv_from(&mut buf) => Some(result),
-            () = STOP_INTERRUPT.notified() => None,
+            () = STOP_INTERRUPT.cancelled() => None,
         };
 
         let Some(Ok((_, addr))) = recv_result else {

pumpkin/src/net/rcon/mod.rs

@@ -24,7 +24,7 @@ impl RCONServer {
         while !SHOULD_STOP.load(Ordering::Relaxed) {
             let await_new_client = || async {
                 let t1 = listener.accept();
-                let t2 = STOP_INTERRUPT.notified();
+                let t2 = STOP_INTERRUPT.cancelled();
 
                 select! {
                     client = t1 => Some(client),