Fix async packet handling and utility edge cases

Co-authored-by: Spagles <106791090+Spagles@users.noreply.github.com>

Author: dmulloy2

src/main/java/com/comphenix/protocol/async/AsyncFilterManager.java

@@ -309,12 +309,10 @@ public synchronized void enqueueSyncPacket(PacketEvent syncPacket, AsyncMarker a
 
         asyncMarker.setQueuedSendingIndex(asyncMarker.getNewSendingIndex());
 
-        // The player is only be null when they're logged out,
-        // so this should be a pretty safe check
-        Player player = syncPacket.getPlayer();
-        if (player != null) {
+        PacketSendingQueue sendingQueue = getSendingQueue(syncPacket);
+        if (sendingQueue != null) {
             // Start the process
-            getSendingQueue(syncPacket).enqueue(newEvent);
+            sendingQueue.enqueue(newEvent);
 
             // We know this is occurring on the main thread, so pass TRUE
             getProcessingQueue(syncPacket).enqueue(newEvent, true);
@@ -323,12 +321,12 @@ public synchronized void enqueueSyncPacket(PacketEvent syncPacket, AsyncMarker a
 
     @Override
     public Set<PacketType> getReceivingTypes() {
-        return serverProcessingQueue.keySet();
+        return clientProcessingQueue.keySet();
     }
 
     @Override
     public Set<PacketType> getSendingTypes() {
-        return clientProcessingQueue.keySet();
+        return serverProcessingQueue.keySet();
     }
 
     /**

src/main/java/com/comphenix/protocol/async/PlayerSendingHandler.java

@@ -134,14 +134,18 @@ public PacketSendingQueue getSendingQueue(PacketEvent packet) {
      * @return The server or client sending queue the packet belongs to.
      */
     public PacketSendingQueue getSendingQueue(PacketEvent packet, boolean createNew) {
-        QueueContainer queues = playerSendingQueues.get(packet.getPlayer());
+        Player player = packet.getPlayer();
+        if (player == null)
+            return null;
+
+        QueueContainer queues = playerSendingQueues.get(player);
 
         // Safe concurrent initialization
         if (queues == null && createNew) {
             final QueueContainer newContainer = new QueueContainer();
 
             // Attempt to map the queue
-            queues = playerSendingQueues.putIfAbsent(packet.getPlayer(), newContainer);
+            queues = playerSendingQueues.putIfAbsent(player, newContainer);
 
             if (queues == null) {
                 queues = newContainer;

src/main/java/com/comphenix/protocol/collections/IntegerMap.java

@@ -58,6 +58,8 @@ public T put(int key, T value) {
      * @return The old associated value, or NULL.
      */
     public T remove(int key) {
+        if (key < 0 || key >= array.length) return null;
+
         T old = array[key];
         array[key] = null;
 

src/main/java/com/comphenix/protocol/concurrency/ConcurrentPlayerMap.java

@@ -190,7 +190,8 @@ private Object cachePlayerKey(Player player) {
         Preconditions.checkNotNull(player, "player cannot be null");
 
         Object key = keyMethod.apply(player);
-        
+        Preconditions.checkState(key != null, "Unable to derive a key for the player.");
+
         keyLookup.put(key, player);
         return key;
     }

src/main/java/com/comphenix/protocol/utility/StreamSerializer.java

@@ -326,15 +326,9 @@ public void serializeItemStack(DataOutputStream output, ItemStack stack) throws
 
     public byte[] getBytesAndRelease(ByteBuf buf) {
         try {
-            if (buf.hasArray()) {
-                // heap buffer, we can access the array directly
-                return buf.array();
-            } else {
-                // direct buffer, we need to copy the bytes into an array
-                byte[] bytes = new byte[buf.readableBytes()];
-                buf.readBytes(bytes);
-                return bytes;
-            }
+            byte[] bytes = new byte[buf.readableBytes()];
+            buf.getBytes(buf.readerIndex(), bytes);
+            return bytes;
         } finally {
             ReferenceCountUtil.safeRelease(buf);
         }

src/test/java/com/comphenix/protocol/async/AsyncFilterManagerTest.java

@@ -0,0 +1,36 @@
+package com.comphenix.protocol.async;
+
+import com.comphenix.protocol.BukkitInitialization;
+import com.comphenix.protocol.PacketType;
+import com.comphenix.protocol.ProtocolManager;
+import com.comphenix.protocol.error.ErrorReporter;
+import com.comphenix.protocol.events.PacketAdapter;
+import com.comphenix.protocol.scheduler.ProtocolScheduler;
+
+import org.bukkit.plugin.Plugin;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+class AsyncFilterManagerTest {
+
+    @BeforeAll
+    static void initializeBukkit() {
+        BukkitInitialization.initializeAll();
+    }
+
+    @Test
+    void testSendingTypesUseServerProcessingQueue() {
+        AsyncFilterManager manager = new AsyncFilterManager(
+                mock(ErrorReporter.class), mock(ProtocolScheduler.class));
+        manager.setManager(mock(ProtocolManager.class));
+        manager.registerAsyncHandler(new PacketAdapter(
+                mock(Plugin.class), PacketType.Play.Server.SYSTEM_CHAT) { }, false);
+
+        assertTrue(manager.getSendingTypes().contains(PacketType.Play.Server.SYSTEM_CHAT));
+        assertFalse(manager.getReceivingTypes().contains(PacketType.Play.Server.SYSTEM_CHAT));
+    }
+}

src/test/java/com/comphenix/protocol/collections/IntegerMapTest.java

@@ -5,6 +5,7 @@
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 class IntegerMapTest {
     @Test
@@ -21,4 +22,15 @@ public void testCapacityIncrement() {
         }
         assertEquals(map.size(), 512);
     }
-}
+
+    @Test
+    public void testRemoveOutsideBounds() {
+        IntegerMap<Boolean> map = new IntegerMap<>();
+        map.put(0, true);
+
+        assertNull(map.remove(-1));
+        assertEquals(1, map.size());
+        assertNull(map.remove(8));
+        assertEquals(1, map.size());
+    }
+}

src/test/java/com/comphenix/protocol/concurrency/ConcurrentPlayerMapTest.java

@@ -0,0 +1,20 @@
+package com.comphenix.protocol.concurrency;
+
+import org.bukkit.entity.Player;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+class ConcurrentPlayerMapTest {
+
+    @Test
+    void testPutIfAbsentWithUnavailableAddress() {
+        ConcurrentPlayerMap<Object> map = ConcurrentPlayerMap.usingAddress();
+        Player player = mock(Player.class);
+
+        assertThrows(IllegalStateException.class, () -> map.putIfAbsent(player, new Object()));
+        assertTrue(map.isEmpty());
+    }
+}

src/test/java/com/comphenix/protocol/utility/StreamSerializerTest.java

@@ -10,6 +10,8 @@
 import com.comphenix.protocol.wrappers.nbt.NbtCompound;
 import com.comphenix.protocol.wrappers.nbt.NbtFactory;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import org.bukkit.ChatColor;
 import org.bukkit.Material;
 import org.bukkit.inventory.ItemStack;
@@ -19,6 +21,7 @@
 
 import static com.comphenix.protocol.utility.TestUtils.assertItemsEqual;
 
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class StreamSerializerTest {
@@ -86,4 +89,23 @@ public void testItemMeta() throws IOException {
 
         assertItemsEqual(initial, deserialized);
     }
+
+    @Test
+    public void testGetBytesAndReleasePreservesReaderIndex() {
+        StreamSerializer serializer = new StreamSerializer();
+        ByteBuf[] buffers = {Unpooled.buffer(16), Unpooled.directBuffer(16)};
+
+        for (ByteBuf buffer : buffers) {
+            buffer.writeBytes(new byte[] {0, 1, 2, 3, 4});
+            buffer.readerIndex(2);
+            buffer.retain();
+
+            try {
+                assertArrayEquals(new byte[] {2, 3, 4}, serializer.getBytesAndRelease(buffer));
+                assertEquals(2, buffer.readerIndex());
+            } finally {
+                buffer.release();
+            }
+        }
+    }
 }