Fix packet compression related bugs

CloudburstMC · Oct 3, 2024 · 258f38c · 258f38c
1 parent c6302d7
commit 258f38c
Show file tree

Hide file tree

Showing 10 changed files with 42 additions and 65 deletions.
diff --git a/src/main/java/cn/nukkit/Server.java b/src/main/java/cn/nukkit/Server.java
@@ -18,7 +18,6 @@
 import cn.nukkit.event.HandlerList;
 import cn.nukkit.event.level.LevelInitEvent;
 import cn.nukkit.event.level.LevelLoadEvent;
-import cn.nukkit.event.server.BatchPacketsEvent;
 import cn.nukkit.event.server.PlayerDataSerializeEvent;
 import cn.nukkit.event.server.QueryRegenerateEvent;
 import cn.nukkit.event.server.ServerStopEvent;
@@ -724,17 +723,7 @@ public static void broadcastPacket(Player[] players, DataPacket packet) {
     }
 
     public void batchPackets(Player[] players, DataPacket[] packets) {
-        if (players == null || packets == null || players.length == 0 || packets.length == 0) {
-            return;
-        }
-
-        BatchPacketsEvent ev = new BatchPacketsEvent(players, packets, true);
-        pluginManager.callEvent(ev);
-        if (ev.isCancelled()) {
-            return;
-        }
-
-        this.batchingHelper.batchPackets(players, packets);
+        this.batchingHelper.batchPackets(this, players, packets);
     }
 
     /**

diff --git a/src/main/java/cn/nukkit/nbt/stream/FastByteArrayOutputStream.java b/src/main/java/cn/nukkit/nbt/stream/FastByteArrayOutputStream.java
@@ -132,7 +132,6 @@ public long length() throws IOException {
     }
 
     public byte[] toByteArray() {
-        if (position == array.length) return array;
         return Arrays.copyOfRange(array, 0, position);
     }
 }
diff --git a/src/main/java/cn/nukkit/network/BatchingHelper.java b/src/main/java/cn/nukkit/network/BatchingHelper.java
@@ -2,6 +2,7 @@
 
 import cn.nukkit.Player;
 import cn.nukkit.Server;
+import cn.nukkit.event.server.BatchPacketsEvent;
 import cn.nukkit.network.protocol.DataPacket;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
@@ -19,16 +20,24 @@ public BatchingHelper() {
         this.threadedExecutor = Executors.newSingleThreadExecutor(builder.build());
     }
 
-    public void batchPackets(Player[] players, DataPacket[] packets) {
-        if (players.length > 0 && packets.length > 0) {
-            this.threadedExecutor.execute(() -> {
-                for (Player player : players) {
-                    for (DataPacket packet : packets) {
-                        player.getNetworkSession().sendPacket(packet);
-                    }
-                }
-            });
+    public void batchPackets(Server server, Player[] players, DataPacket[] packets) {
+        if (players == null || packets == null || players.length == 0 || packets.length == 0) {
+            return;
+        }
+
+        BatchPacketsEvent ev = new BatchPacketsEvent(players, packets, true);
+        server.getPluginManager().callEvent(ev);
+        if (ev.isCancelled()) {
+            return;
         }
+
+        this.threadedExecutor.execute(() -> { // Maybe players could have separate threads assigned to them?
+            for (Player player : players) {
+                for (DataPacket packet : packets) {
+                    player.getNetworkSession().sendPacket(packet);
+                }
+            }
+        });
     }
 
     public void shutdown() {

diff --git a/src/main/java/cn/nukkit/network/CompressionProvider.java b/src/main/java/cn/nukkit/network/CompressionProvider.java
@@ -29,23 +29,6 @@ public byte getPrefix() {
         }
     };
 
-    CompressionProvider ZLIB = new CompressionProvider() {
-        @Override
-        public byte[] compress(BinaryStream packet, int level) throws Exception {
-            return Zlib.deflatePre16Packet(packet.getBuffer(), level);
-        }
-
-        @Override
-        public byte[] decompress(byte[] compressed) throws Exception {
-            return Zlib.inflate(compressed, 6291456);
-        }
-
-        @Override
-        public byte[] decompress(byte[] compressed, int maxSize) throws Exception {
-            return Zlib.inflate(compressed, maxSize);
-        }
-    };
-
     CompressionProvider ZLIB_RAW = new CompressionProvider() {
         @Override
         public byte[] compress(BinaryStream packet, int level) throws Exception {
@@ -98,11 +81,11 @@ default byte[] decompress(byte[] compressed, int maxSize) throws Exception {
         return this.decompress(compressed);
     }
 
-    static CompressionProvider from(PacketCompressionAlgorithm algorithm, int raknetVersion) {
+    static CompressionProvider from(PacketCompressionAlgorithm algorithm) {
         if (algorithm == null) {
             return NONE;
         } else if (algorithm == PacketCompressionAlgorithm.ZLIB) {
-            return raknetVersion < 10 ? ZLIB : ZLIB_RAW;
+            return ZLIB_RAW;
         } else if (algorithm == PacketCompressionAlgorithm.SNAPPY) {
             return SNAPPY;
         }

diff --git a/src/main/java/cn/nukkit/network/protocol/BatchPacket.java b/src/main/java/cn/nukkit/network/protocol/BatchPacket.java
@@ -28,4 +28,13 @@ public void encode() {
     public void trim() {
         setBuffer(null);
     }
+
+    @Override
+    public BatchPacket clone() {
+        BatchPacket packet = (BatchPacket) super.clone();
+        if (this.payload != null) {
+            packet.payload = this.payload.clone();
+        }
+        return packet;
+    }
 }
diff --git a/src/main/java/cn/nukkit/network/protocol/ChunkRadiusUpdatedPacket.java b/src/main/java/cn/nukkit/network/protocol/ChunkRadiusUpdatedPacket.java
@@ -20,7 +20,7 @@ public void decode() {
 
     @Override
     public void encode() {
-        super.reset();
+        this.reset();
         this.putVarInt(this.radius);
     }
 

diff --git a/src/main/java/cn/nukkit/network/protocol/DataPacket.java b/src/main/java/cn/nukkit/network/protocol/DataPacket.java
@@ -20,15 +20,24 @@ public abstract class DataPacket extends BinaryStream implements Cloneable {
 
     public abstract void encode();
 
+    public final void tryEncode() {
+        if (!this.isEncoded) {
+            this.isEncoded = true;
+            this.encode();
+        }
+    }
+
     @Override
     public DataPacket reset() {
         super.reset();
+
         byte packetId = this.pid();
         if (packetId < 0 && packetId >= -56) { // Hack: (byte) 200+ --> (int) 300+
             this.putUnsignedVarInt(packetId + 356);
         } else {
             this.putUnsignedVarInt(packetId & 0xff);
         }
+
         return this;
     }
 
@@ -52,7 +61,7 @@ public DataPacket clean() {
     public DataPacket clone() {
         try {
             DataPacket packet = (DataPacket) super.clone();
-            packet.setBuffer(this.count < 0 ? null : this.getBuffer()); // prevent reflecting same buffer instance
+            packet.setBuffer(this.getBuffer()); // prevent reflecting same buffer instance
             packet.offset = this.offset;
             packet.count = this.count;
             return packet;
@@ -84,13 +93,6 @@ public BatchPacket compress(int level) {
         }
     }
 
-    public final void tryEncode() {
-        if (!this.isEncoded) {
-            this.isEncoded = true;
-            this.encode();
-        }
-    }
-
     void decodeUnsupported() {
         if (Nukkit.DEBUG > 1) {
             Server.getInstance().getLogger().debug("Warning: decode() not implemented for " + this.getClass().getName());

diff --git a/src/main/java/cn/nukkit/network/protocol/InventoryContentPacket.java b/src/main/java/cn/nukkit/network/protocol/InventoryContentPacket.java
@@ -51,11 +51,4 @@ public void encode() {
         this.putBoolean(false); // fullContainerName.optional.present
         this.putUnsignedVarInt(0); // dynamicContainerSize
     }
-
-    @Override
-    public InventoryContentPacket clone() {
-        InventoryContentPacket pk = (InventoryContentPacket) super.clone();
-        pk.slots = this.slots.clone();
-        return pk;
-    }
 }
diff --git a/src/main/java/cn/nukkit/network/session/RakNetPlayerSession.java b/src/main/java/cn/nukkit/network/session/RakNetPlayerSession.java
@@ -8,7 +8,6 @@
 import cn.nukkit.network.protocol.BatchPacket;
 import cn.nukkit.network.protocol.DataPacket;
 import cn.nukkit.network.protocol.DisconnectPacket;
-import cn.nukkit.network.protocol.ProtocolInfo;
 import cn.nukkit.utils.BinaryStream;
 import com.google.common.base.Preconditions;
 import com.nukkitx.natives.sha256.Sha256;
@@ -144,7 +143,7 @@ public void sendPacket(DataPacket packet) {
             return;
         }
 
-        if (packet.pid() != ProtocolInfo.BATCH_PACKET) {
+        if (!(packet instanceof BatchPacket)) {
             packet.tryEncode();
         }
 

diff --git a/src/main/java/cn/nukkit/utils/Zlib.java b/src/main/java/cn/nukkit/utils/Zlib.java
@@ -1,7 +1,5 @@
 package cn.nukkit.utils;
 
-import cn.nukkit.Server;
-
 import java.io.IOException;
 import java.util.zip.Deflater;
 
@@ -17,10 +15,6 @@ public static byte[] deflate(byte[] data, int level) throws Exception {
         return provider.deflate(data, level);
     }
 
-    public static byte[] deflatePre16Packet(byte[] data, int level) throws Exception {
-        return provider.deflate(data, data.length < Server.getInstance().networkCompressionThreshold ? 0 : level);
-    }
-
     public static byte[] deflate(byte[][] data, int level) throws Exception {
         return provider.deflate(data, level);
     }