Don't block Netty event loop threads while handling Floodgate login

bungee/src/main/java/org/geysermc/floodgate/addon/data/BungeeDataAddon.java

@@ -46,6 +46,10 @@ public class BungeeDataAddon implements InjectorAddon {
     @Named("packetHandler")
     private String packetHandler;
 
+    @Inject
+    @Named("packetDecoder")
+    private String packetDecoder;
+
     @Inject
     @Named("packetEncoder")
     private String packetEncoder;
@@ -69,9 +73,13 @@ public class BungeeDataAddon implements InjectorAddon {
             }
             return;
         }
+
+        PacketBlocker blocker = new PacketBlocker();
+        channel.pipeline().addBefore(packetDecoder, "floodgate_packet_blocker", blocker);
+
         channel.pipeline().addBefore(
                 packetHandler, "floodgate_data_handler",
-                new BungeeProxyDataHandler(config, handshakeHandler, kickMessageAttribute)
+                new BungeeProxyDataHandler(config, handshakeHandler, blocker, kickMessageAttribute)
         );
     }
 

bungee/src/main/java/org/geysermc/floodgate/addon/data/BungeeProxyDataHandler.java

@@ -27,11 +27,14 @@ package org.geysermc.floodgate.addon.data;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import com.google.common.collect.Queues;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.util.AttributeKey;
 import java.lang.reflect.Field;
 import java.net.InetSocketAddress;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
 import lombok.RequiredArgsConstructor;
 import net.md_5.bungee.connection.InitialHandler;
 import net.md_5.bungee.netty.ChannelWrapper;
@@ -42,7 +45,6 @@ import net.md_5.bungee.protocol.packet.Handshake;
 import org.geysermc.floodgate.api.handshake.HandshakeData;
 import org.geysermc.floodgate.config.ProxyFloodgateConfig;
 import org.geysermc.floodgate.player.FloodgateHandshakeHandler;
-import org.geysermc.floodgate.player.FloodgateHandshakeHandler.HandshakeResult;
 import org.geysermc.floodgate.util.Constants;
 import org.geysermc.floodgate.util.ReflectionUtils;
 
@@ -63,62 +65,96 @@ public class BungeeProxyDataHandler extends ChannelInboundHandlerAdapter {
 
     private final ProxyFloodgateConfig config;
     private final FloodgateHandshakeHandler handler;
+    private final PacketBlocker blocker;
     private final AttributeKey<String> kickMessageAttribute;
 
+    private final Queue<Object> packetQueue = Queues.newConcurrentLinkedQueue();
+
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        // prevent other packets from being handled while we handle the handshake packet
+        if (!packetQueue.isEmpty()) {
+            packetQueue.add(msg);
+            return;
+        }
+
         if (msg instanceof PacketWrapper) {
             DefinedPacket packet = ((PacketWrapper) msg).packet;
 
             // we're only interested in the Handshake packet
             if (packet instanceof Handshake) {
-                handleHandshake(ctx, (Handshake) packet);
-                ctx.pipeline().remove(this);
+                blocker.enable();
+                packetQueue.add(msg);
+
+                handleHandshake(ctx, (Handshake) packet).thenRun(() -> {
+                    Object queuedPacket;
+                    while ((queuedPacket = packetQueue.poll()) != null) {
+                        ctx.fireChannelRead(queuedPacket);
+                    }
+                    ctx.pipeline().remove(this);
+                    blocker.disable();
+                });
+                return;
             }
         }
 
         ctx.fireChannelRead(msg);
     }
 
-    private void handleHandshake(ChannelHandlerContext ctx, Handshake packet) {
+    private CompletableFuture<Void> handleHandshake(ChannelHandlerContext ctx, Handshake packet) {
         String data = packet.getHost();
 
-        HandshakeResult result = handler.handle(ctx.channel(), data);
-        HandshakeData handshakeData = result.getHandshakeData();
+        return handler.handle(ctx.channel(), data).thenAccept(result -> {
+            HandshakeData handshakeData = result.getHandshakeData();
 
-        // we'll change the IP address from the proxy to the real IP of the client very early on
-        // so that almost every plugin will use the real IP of the client
-        InetSocketAddress newIp = result.getNewIp(ctx.channel());
-        if (newIp != null) {
-            HandlerBoss handlerBoss = ctx.pipeline().get(HandlerBoss.class);
-            // InitialHandler extends PacketHandler and implements PendingConnection
-            InitialHandler connection = ReflectionUtils.getCastedValue(handlerBoss, HANDLER);
+            // we'll change the IP address from the proxy to the real IP of the client very early on
+            // so that almost every plugin will use the real IP of the client
+            InetSocketAddress newIp = result.getNewIp(ctx.channel());
+            if (newIp != null) {
+                HandlerBoss handlerBoss = ctx.pipeline().get(HandlerBoss.class);
+                // InitialHandler extends PacketHandler and implements PendingConnection
+                InitialHandler connection = ReflectionUtils.getCastedValue(handlerBoss, HANDLER);
 
-            ChannelWrapper channelWrapper =
-                    ReflectionUtils.getCastedValue(connection, CHANNEL_WRAPPER);
+                ChannelWrapper channelWrapper =
+                        ReflectionUtils.getCastedValue(connection, CHANNEL_WRAPPER);
 
-            channelWrapper.setRemoteAddress(newIp);
-        }
+                channelWrapper.setRemoteAddress(newIp);
+            }
 
-        if (handshakeData.getDisconnectReason() != null) {
-            ctx.channel().attr(kickMessageAttribute).set(handshakeData.getDisconnectReason());
-            return;
-        }
+            if (handshakeData.getDisconnectReason() != null) {
+                ctx.channel().attr(kickMessageAttribute).set(handshakeData.getDisconnectReason());
+                return;
+            }
 
-        switch (result.getResultType()) {
-            case EXCEPTION:
-                ctx.channel().attr(kickMessageAttribute).set(
-                        config.getDisconnect().getInvalidKey());
-                break;
-            case INVALID_DATA_LENGTH:
-                ctx.channel().attr(kickMessageAttribute)
-                        .set(config.getDisconnect().getInvalidArgumentsLength());
-                break;
-            case TIMESTAMP_DENIED:
-                ctx.channel().attr(kickMessageAttribute).set(Constants.TIMESTAMP_DENIED_MESSAGE);
-                break;
-            default:
-                break;
+            switch (result.getResultType()) {
+                case EXCEPTION:
+                    ctx.channel().attr(kickMessageAttribute)
+                            .set(config.getDisconnect().getInvalidKey());
+                    break;
+                case INVALID_DATA_LENGTH:
+                    ctx.channel().attr(kickMessageAttribute)
+                            .set(config.getDisconnect().getInvalidArgumentsLength());
+                    break;
+                case TIMESTAMP_DENIED:
+                    ctx.channel().attr(kickMessageAttribute)
+                            .set(Constants.TIMESTAMP_DENIED_MESSAGE);
+                    break;
+                default:
+                    break;
+            }
+        }).handle((v, error) -> {
+            if (error != null) {
+                error.printStackTrace();
+            }
+            return v;
+        });
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        super.exceptionCaught(ctx, cause);
+        if (config.isDebug()) {
+            cause.printStackTrace();
         }
     }
 }

common/src/main/java/org/geysermc/floodgate/addon/data/PacketBlocker.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2019-2021 GeyserMC. http://geysermc.org
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ *
+ * @author GeyserMC
+ * @link https://github.com/GeyserMC/Floodgate
+ */
+
+package org.geysermc.floodgate.addon.data;
+
+import com.google.common.collect.Queues;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import java.util.Queue;
+
+public class PacketBlocker extends ChannelInboundHandlerAdapter {
+    private final Queue<Object> packetQueue = Queues.newConcurrentLinkedQueue();
+    private volatile boolean blockPackets;
+
+    private ChannelHandlerContext ctx;
+
+    public void enable() {
+        blockPackets = true;
+    }
+
+    public void disable() {
+        blockPackets = false;
+
+        Object packet;
+        while ((packet = packetQueue.poll()) != null) {
+            ctx.fireChannelRead(packet);
+        }
+        ctx.pipeline().remove(this);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        if (blockPackets || !packetQueue.isEmpty()) {
+            packetQueue.add(msg);
+            return;
+        }
+        ctx.fireChannelRead(msg);
+    }
+
+    @Override
+    public void handlerAdded(ChannelHandlerContext ctx) {
+        this.ctx = ctx;
+    }
+}

common/src/main/java/org/geysermc/floodgate/player/FloodgateHandshakeHandler.java

@@ -25,6 +25,8 @@
 
 package org.geysermc.floodgate.player;
 
+import static org.geysermc.floodgate.player.FloodgateHandshakeHandler.ResultType.INVALID_DATA_LENGTH;
+import static org.geysermc.floodgate.player.FloodgateHandshakeHandler.ResultType.NOT_FLOODGATE_DATA;
 import static org.geysermc.floodgate.util.BedrockData.EXPECTED_LENGTH;
 
 import com.google.common.base.Charsets;
@@ -32,9 +34,11 @@ import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import io.netty.channel.Channel;
 import io.netty.util.AttributeKey;
+import it.unimi.dsi.fastutil.Pair;
+import it.unimi.dsi.fastutil.objects.ObjectObjectMutablePair;
 import java.net.InetSocketAddress;
 import java.util.UUID;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import lombok.AccessLevel;
 import lombok.AllArgsConstructor;
@@ -75,7 +79,10 @@ public final class FloodgateHandshakeHandler {
     private final AttributeKey<FloodgatePlayer> playerAttribute;
     private final FloodgateLogger logger;
 
-    public HandshakeResult handle(Channel channel, @NonNull String originalHostname) {
+    public CompletableFuture<HandshakeResult> handle(
+            @NonNull Channel channel,
+            @NonNull String originalHostname) {
+
         String[] split = originalHostname.split("\0");
         String data = null;
 
@@ -91,74 +98,117 @@ public final class FloodgateHandshakeHandler {
         String hostname = hostnameBuilder.toString();
 
         if (data == null) {
-            return callHandlerAndReturnResult(
-                    ResultType.NOT_FLOODGATE_DATA,
-                    channel, null, hostname);
+            return CompletableFuture.completedFuture(
+                    callHandlerAndReturnResult(NOT_FLOODGATE_DATA, channel, null, hostname)
+            );
         }
 
-        try {
-            byte[] floodgateData = data.getBytes(Charsets.UTF_8);
+        byte[] floodgateData = data.getBytes(Charsets.UTF_8);
 
-            // actual decryption
-            String decrypted = cipher.decryptToString(floodgateData);
-            BedrockData bedrockData = BedrockData.fromString(decrypted);
+        return CompletableFuture.supplyAsync(() -> {
+            try {
+                // actual decryption
+                String decrypted = cipher.decryptToString(floodgateData);
+                BedrockData bedrockData = BedrockData.fromString(decrypted);
 
-            if (bedrockData.getDataLength() != EXPECTED_LENGTH) {
-                return callHandlerAndReturnResult(
-                        ResultType.INVALID_DATA_LENGTH,
-                        channel, bedrockData, hostname);
-            }
-
-            // timestamp checks
-
-            TimeSyncer timeSyncer = TimeSyncerHolder.get();
-
-            if (!timeSyncer.hasUsefulOffset()) {
-                logger.warn("We couldn't make sure that your system clock is accurate. " +
-                        "This can cause issues with logging in.");
-            }
-
-            // the time syncer is accurate, but we have to account for some minor differences
-            final int errorMargin = 150; // 150ms
-
-            long timeDifference = timeSyncer.getRealMillis() - bedrockData.getTimestamp();
-            if (timeDifference > 6000 + errorMargin || timeDifference < -errorMargin) {
-                if (Constants.DEBUG_MODE || logger.isDebug()) {
-                    logger.info("Current time: " + System.currentTimeMillis());
-                    logger.info("Stored time: " + bedrockData.getTimestamp());
-                    logger.info("Time offset: " + timeSyncer.getTimeOffset());
+                if (bedrockData.getDataLength() != EXPECTED_LENGTH) {
+                    throw callHandlerAndReturnResult(
+                            INVALID_DATA_LENGTH,
+                            channel, bedrockData, hostname
+                    );
                 }
-                return callHandlerAndReturnResult(
-                        ResultType.TIMESTAMP_DENIED,
-                        channel, bedrockData, hostname);
-            }
 
-            Long cachedTimestamp = handleCache.getIfPresent(bedrockData.getXuid());
-            if (cachedTimestamp != null) {
-                // the cached timestamp should be older than the received timestamp
-                // and it should also not be possible to reuse the handshake
-                long diff = bedrockData.getTimestamp() - cachedTimestamp;
-                if (diff == 0 || diff < 0 && -diff > errorMargin) {
-                    return callHandlerAndReturnResult(
+                // timestamp checks
+
+                TimeSyncer timeSyncer = TimeSyncerHolder.get();
+
+                if (!timeSyncer.hasUsefulOffset()) {
+                    logger.warn("We couldn't make sure that your system clock is accurate. " +
+                            "This can cause issues with logging in.");
+                }
+
+                // the time syncer is accurate, but we have to account for some minor differences
+                final int errorMargin = 150; // 150ms
+
+                long timeDifference = timeSyncer.getRealMillis() - bedrockData.getTimestamp();
+                if (timeDifference > 6000 + errorMargin || timeDifference < -errorMargin) {
+                    if (Constants.DEBUG_MODE || logger.isDebug()) {
+                        logger.info("Current time: " + System.currentTimeMillis());
+                        logger.info("Stored time: " + bedrockData.getTimestamp());
+                        logger.info("Time offset: " + timeSyncer.getTimeOffset());
+                    }
+                    throw callHandlerAndReturnResult(
                             ResultType.TIMESTAMP_DENIED,
-                            channel, bedrockData, hostname);
+                            channel, bedrockData, hostname
+                    );
                 }
+
+                Long cachedTimestamp = handleCache.getIfPresent(bedrockData.getXuid());
+                if (cachedTimestamp != null) {
+                    // the cached timestamp should be older than the received timestamp
+                    // and it should also not be possible to reuse the handshake
+                    long diff = bedrockData.getTimestamp() - cachedTimestamp;
+                    if (diff == 0 || diff < 0 && -diff > errorMargin) {
+                        throw callHandlerAndReturnResult(
+                                ResultType.TIMESTAMP_DENIED,
+                                channel, bedrockData, hostname
+                        );
+                    }
+                }
+
+                handleCache.put(bedrockData.getXuid(), bedrockData.getTimestamp());
+
+                // we'll use the LinkedPlayer provided by Bungee or Velocity (if they included one)
+                if (bedrockData.hasPlayerLink()) {
+                    throw handlePart2(channel, hostname, bedrockData, bedrockData.getLinkedPlayer());
+                }
+                //todo add option to not check for links when the data comes from a proxy
+
+                // let's check if there is a link
+                return bedrockData;
+
+            } catch (InvalidFormatException formatException) {
+                // only header exceptions should return 'not floodgate data',
+                // all the other format exceptions are because of invalid/tempered Floodgate data
+                if (formatException.isHeader()) {
+                    throw callHandlerAndReturnResult(
+                            NOT_FLOODGATE_DATA,
+                            channel, null, hostname
+                    );
+                }
+
+                formatException.printStackTrace();
+
+                throw callHandlerAndReturnResult(
+                        ResultType.EXCEPTION,
+                        channel, null, hostname
+                );
+            } catch (Exception exception) {
+                exception.printStackTrace();
+
+                throw callHandlerAndReturnResult(
+                        ResultType.EXCEPTION,
+                        channel, null, hostname
+                );
+            }
+        }).thenCompose(this::fetchLinkedPlayer).handle((result, error) -> {
+            if (error == null) {
+                return handlePart2(channel, hostname, result.left(), result.right());
             }
 
-            handleCache.put(bedrockData.getXuid(), bedrockData.getTimestamp());
-
-
-            LinkedPlayer linkedPlayer;
-
-            // we'll use the LinkedPlayer provided by Bungee or Velocity (if they included one)
-            if (bedrockData.hasPlayerLink()) {
-                linkedPlayer = bedrockData.getLinkedPlayer();
-            } else {
-                // every implementation (Bukkit, Bungee and Velocity) run this constructor async,
-                // so we should be fine doing this synchronised.
-                linkedPlayer = fetchLinkedPlayer(Utils.getJavaUuid(bedrockData.getXuid()));
+            if (error instanceof HandshakeResult) {
+                return (HandshakeResult) error;
             }
 
+            return callHandlerAndReturnResult(
+                    ResultType.EXCEPTION,
+                    channel, null, hostname
+            );
+        });
+    }
+
+    private HandshakeResult handlePart2(Channel channel, String hostname, BedrockData bedrockData, LinkedPlayer linkedPlayer) {
+        try {
             HandshakeData handshakeData = new HandshakeDataImpl(
                     channel, true, bedrockData.clone(), configHolder.get(),
                     linkedPlayer != null ? linkedPlayer.clone() : null, hostname);
@@ -176,8 +226,7 @@ public final class FloodgateHandshakeHandler {
 
             correctHostname(handshakeData);
 
-            FloodgatePlayer player =
-                    FloodgatePlayerImpl.from(bedrockData, handshakeData);
+            FloodgatePlayer player = FloodgatePlayerImpl.from(bedrockData, handshakeData);
 
             api.addPlayer(player.getJavaUniqueId(), player);
 
@@ -188,28 +237,9 @@ public final class FloodgateHandshakeHandler {
             player.addProperty(PropertyKey.SOCKET_ADDRESS, socketAddress);
 
             return new HandshakeResult(ResultType.SUCCESS, handshakeData, bedrockData, player);
-
-        } catch (InvalidFormatException formatException) {
-            // only header exceptions should return 'not floodgate data',
-            // all the other format exceptions are because of invalid/tempered Floodgate data
-            if (formatException.isHeader()) {
-                return callHandlerAndReturnResult(
-                        ResultType.NOT_FLOODGATE_DATA,
-                        channel, null, hostname);
-            }
-
-            formatException.printStackTrace();
-
-            return callHandlerAndReturnResult(
-                    ResultType.EXCEPTION,
-                    channel, null, hostname);
-
         } catch (Exception exception) {
             exception.printStackTrace();
-
-            return callHandlerAndReturnResult(
-                    ResultType.EXCEPTION,
-                    channel, null, hostname);
+            return callHandlerAndReturnResult(ResultType.EXCEPTION, channel, null, hostname);
         }
     }
 
@@ -247,17 +277,12 @@ public final class FloodgateHandshakeHandler {
         handshakeData.setHostname(String.join("\0", split));
     }
 
-    private LinkedPlayer fetchLinkedPlayer(UUID javaUniqueId) {
+    private CompletableFuture<Pair<BedrockData, LinkedPlayer>> fetchLinkedPlayer(BedrockData data) {
         if (!api.getPlayerLink().isEnabled()) {
-            return null;
-        }
-
-        try {
-            return api.getPlayerLink().getLinkedPlayer(javaUniqueId).get();
-        } catch (InterruptedException | ExecutionException exception) {
-            exception.printStackTrace();
-            return null;
+            return CompletableFuture.completedFuture(null);
         }
+        return api.getPlayerLink().getLinkedPlayer(Utils.getJavaUuid(data.getXuid()))
+                .thenApply(link -> new ObjectObjectMutablePair<>(data, link));
     }
 
     public enum ResultType {
@@ -268,9 +293,9 @@ public final class FloodgateHandshakeHandler {
         SUCCESS
     }
 
-    @Getter
     @AllArgsConstructor(access = AccessLevel.PROTECTED)
-    public static class HandshakeResult {
+    @Getter
+    public static class HandshakeResult extends IllegalStateException {
         private final ResultType resultType;
         private final HandshakeData handshakeData;
         private final BedrockData bedrockData;

ruleset.xml

@@ -18,6 +18,8 @@
   <exclude-pattern>.*/FloodgateConfig.*</exclude-pattern>
   <!-- CloseResource, there is no shutdown event and it has to load classes on the fly -->
   <exclude-pattern>.*/PlayerLinkLoader.*</exclude-pattern>
+  <!-- PreserveStackTrace -->
+  <exclude-pattern>.*/FloodgateHandshakeHandler.*</exclude-pattern>
 
   <rule ref="category/java/bestpractices.xml/MissingOverride" />
   <rule ref="category/java/bestpractices.xml/UseCollectionIsEmpty" />
@@ -46,4 +48,9 @@
   </rule>
   <rule ref="category/java/security.xml" />
 
+  <rule ref="category/java/errorprone.xml/AssignmentInOperand">
+    <properties>
+      <property name="allowWhile" value="true" />
+    </properties>
+  </rule>
 </ruleset>

spigot/src/main/java/org/geysermc/floodgate/addon/data/SpigotDataAddon.java

@@ -42,6 +42,10 @@ public final class SpigotDataAddon implements InjectorAddon {
     @Inject private SimpleFloodgateApi api;
     @Inject private FloodgateLogger logger;
 
+    @Inject
+    @Named("packetDecoder")
+    private String packetDecoder;
+
     @Inject
     @Named("packetHandler")
     private String packetHandlerName;
@@ -52,9 +56,12 @@ public final class SpigotDataAddon implements InjectorAddon {
 
     @Override
     public void onInject(Channel channel, boolean toServer) {
+        PacketBlocker blocker = new PacketBlocker();
+        channel.pipeline().addBefore(packetDecoder, "floodgate_packet_blocker", blocker);
+
         channel.pipeline().addBefore(
                 packetHandlerName, "floodgate_data_handler",
-                new SpigotDataHandler(config, handshakeHandler, logger)
+                new SpigotDataHandler(config, handshakeHandler, blocker, logger)
         );
     }
 

spigot/src/main/java/org/geysermc/floodgate/addon/data/SpigotDataHandler.java

@@ -28,17 +28,18 @@ package org.geysermc.floodgate.addon.data;
 import static org.geysermc.floodgate.util.ReflectionUtils.getCastedValue;
 import static org.geysermc.floodgate.util.ReflectionUtils.setValue;
 
+import com.google.common.collect.Queues;
 import com.mojang.authlib.GameProfile;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import java.net.InetSocketAddress;
+import java.util.Queue;
 import lombok.RequiredArgsConstructor;
 import org.geysermc.floodgate.api.handshake.HandshakeData;
 import org.geysermc.floodgate.api.logger.FloodgateLogger;
 import org.geysermc.floodgate.api.player.FloodgatePlayer;
 import org.geysermc.floodgate.config.FloodgateConfig;
 import org.geysermc.floodgate.player.FloodgateHandshakeHandler;
-import org.geysermc.floodgate.player.FloodgateHandshakeHandler.HandshakeResult;
 import org.geysermc.floodgate.util.BedrockData;
 import org.geysermc.floodgate.util.ClassNames;
 import org.geysermc.floodgate.util.Constants;
@@ -48,23 +49,30 @@ import org.geysermc.floodgate.util.ProxyUtils;
 public final class SpigotDataHandler extends ChannelInboundHandlerAdapter {
     private final FloodgateConfig config;
     private final FloodgateHandshakeHandler handshakeHandler;
+    private final PacketBlocker blocker;
     private final FloodgateLogger logger;
+
+    private final Queue<Object> packetQueue = Queues.newConcurrentLinkedQueue();
+
     private Object networkManager;
     private FloodgatePlayer player;
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object packet) throws Exception {
-        boolean isHandshake = ClassNames.HANDSHAKE_PACKET.isInstance(packet);
-        boolean isLogin = ClassNames.LOGIN_START_PACKET.isInstance(packet);
+        // prevent other packets from being handled while we handle the handshake packet
+        if (!packetQueue.isEmpty()) {
+            packetQueue.add(packet);
+            return;
+        }
 
-        boolean bungeeData = false;
+        if (ClassNames.HANDSHAKE_PACKET.isInstance(packet)) {
+            blocker.enable();
+            packetQueue.add(packet);
 
-        try {
-            if (isHandshake) {
-                networkManager = ctx.channel().pipeline().get("packet_handler");
+            networkManager = ctx.channel().pipeline().get("packet_handler");
+            String handshakeValue = getCastedValue(packet, ClassNames.HANDSHAKE_HOST);
 
-                String handshakeValue = getCastedValue(packet, ClassNames.HANDSHAKE_HOST);
-                HandshakeResult result = handshakeHandler.handle(ctx.channel(), handshakeValue);
+            handshakeHandler.handle(ctx.channel(), handshakeValue).thenApply(result -> {
                 HandshakeData handshakeData = result.getHandshakeData();
 
                 setValue(packet, ClassNames.HANDSHAKE_HOST, handshakeData.getHostname());
@@ -77,7 +85,7 @@ public final class SpigotDataHandler extends ChannelInboundHandlerAdapter {
 
                 if (handshakeData.getDisconnectReason() != null) {
                     ctx.close(); //todo disconnect with message
-                    return;
+                    return true;
                 }
 
                 //todo use kickMessageAttribute and let this be common logic
@@ -88,7 +96,7 @@ public final class SpigotDataHandler extends ChannelInboundHandlerAdapter {
                     case EXCEPTION:
                         logger.info(config.getDisconnect().getInvalidKey());
                         ctx.close();
-                        return;
+                        return true;
                     case INVALID_DATA_LENGTH:
                         int dataLength = result.getBedrockData().getDataLength();
                         logger.info(
@@ -96,62 +104,82 @@ public final class SpigotDataHandler extends ChannelInboundHandlerAdapter {
                                 BedrockData.EXPECTED_LENGTH, dataLength
                         );
                         ctx.close();
-                        return;
+                        return true;
                     case TIMESTAMP_DENIED:
                         logger.info(Constants.TIMESTAMP_DENIED_MESSAGE);
                         ctx.close();
-                        return;
+                        return true;
                     default: // only continue when SUCCESS
-                        return;
+                        return true;
                 }
 
                 player = result.getFloodgatePlayer();
-                bungeeData = ProxyUtils.isProxyData();
+                boolean bungeeData = ProxyUtils.isProxyData();
 
                 if (!bungeeData) {
                     // Use a spoofedUUID for initUUID (just like Bungeecord)
                     setValue(networkManager, "spoofedUUID", player.getCorrectUniqueId());
                 }
-            } else if (isLogin) {
-                // we have to fake the offline player (login) cycle
-                Object loginListener = ClassNames.PACKET_LISTENER.get(networkManager);
-
-                // check if the server is actually in the Login state
-                if (!ClassNames.LOGIN_LISTENER.isInstance(loginListener)) {
-                    // player is not in the login state, abort
-                    return;
+                return bungeeData || player == null;
+            }).thenAccept(shouldRemove -> {
+                Object queuedPacket;
+                while ((queuedPacket = packetQueue.poll()) != null) {
+                    try {
+                        if (checkLogin(ctx, packet)) {
+                            break;
+                        }
+                    } catch (Exception ignored) {}
+                    ctx.fireChannelRead(queuedPacket);
                 }
 
-                // set the player his GameProfile, we can't change the username without this
-                GameProfile gameProfile = new GameProfile(
-                        player.getCorrectUniqueId(), player.getCorrectUsername()
-                );
-                setValue(loginListener, ClassNames.LOGIN_PROFILE, gameProfile);
-
-                // just like on Spigot:
-
-                // LoginListener#initUUID
-                // new LoginHandler().fireEvents();
-
-                // and the tick of LoginListener will do the rest
-
-                ClassNames.INIT_UUID.invoke(loginListener);
-
-                Object loginHandler =
-                        ClassNames.LOGIN_HANDLER_CONSTRUCTOR.newInstance(loginListener);
-                ClassNames.FIRE_LOGIN_EVENTS.invoke(loginHandler);
-            }
-        } finally {
-            // don't let the packet through if the packet is the login packet
-            if (!isLogin) {
-                ctx.fireChannelRead(packet);
-            }
-
-            if (isHandshake && bungeeData || isLogin || player == null) {
-                // We're done
-                ctx.pipeline().remove(this);
-            }
+                if (shouldRemove) {
+                    ctx.pipeline().remove(SpigotDataHandler.this);
+                }
+                blocker.disable();
+            });
+            return;
         }
+
+        if (!checkLogin(ctx, packet)) {
+            ctx.fireChannelRead(packet);
+        }
+    }
+
+    private boolean checkLogin(ChannelHandlerContext ctx, Object packet) throws Exception {
+        if (ClassNames.LOGIN_START_PACKET.isInstance(packet)) {
+            // we have to fake the offline player (login) cycle
+            Object loginListener = ClassNames.PACKET_LISTENER.get(networkManager);
+
+            // check if the server is actually in the Login state
+            if (!ClassNames.LOGIN_LISTENER.isInstance(loginListener)) {
+                // player is not in the login state, abort
+                ctx.pipeline().remove(this);
+                return true;
+            }
+
+            // set the player his GameProfile, we can't change the username without this
+            GameProfile gameProfile = new GameProfile(
+                    player.getCorrectUniqueId(), player.getCorrectUsername()
+            );
+            setValue(loginListener, ClassNames.LOGIN_PROFILE, gameProfile);
+
+            // just like on Spigot:
+
+            // LoginListener#initUUID
+            // new LoginHandler().fireEvents();
+
+            // and the tick of LoginListener will do the rest
+
+            ClassNames.INIT_UUID.invoke(loginListener);
+
+            Object loginHandler =
+                    ClassNames.LOGIN_HANDLER_CONSTRUCTOR.newInstance(loginListener);
+            ClassNames.FIRE_LOGIN_EVENTS.invoke(loginHandler);
+
+            ctx.pipeline().remove(this);
+            return true;
+        }
+        return false;
     }
 
     @Override

velocity/src/main/java/org/geysermc/floodgate/addon/data/VelocityDataAddon.java

@@ -48,6 +48,10 @@ public final class VelocityDataAddon implements InjectorAddon {
     @Named("packetHandler")
     private String packetHandler;
 
+    @Inject
+    @Named("packetDecoder")
+    private String packetDecoder;
+
     @Inject
     @Named("packetEncoder")
     private String packetEncoder;
@@ -71,10 +75,14 @@ public final class VelocityDataAddon implements InjectorAddon {
             }
             return;
         }
+
+        PacketBlocker blocker = new PacketBlocker();
+        channel.pipeline().addBefore(packetDecoder, "floodgate_packet_blocker", blocker);
+
         // The handler is already added so we should add our handler before it
         channel.pipeline().addBefore(
                 packetHandler, "floodgate_data_handler",
-                new VelocityProxyDataHandler(config, handshakeHandler, kickMessageAttribute, logger)
+                new VelocityProxyDataHandler(config, handshakeHandler, blocker, kickMessageAttribute, logger)
         );
     }
 

velocity/src/main/java/org/geysermc/floodgate/addon/data/VelocityProxyDataHandler.java

@@ -31,18 +31,20 @@ import static org.geysermc.floodgate.util.ReflectionUtils.getField;
 import static org.geysermc.floodgate.util.ReflectionUtils.getPrefixedClass;
 import static org.geysermc.floodgate.util.ReflectionUtils.setValue;
 
+import com.google.common.collect.Queues;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.util.AttributeKey;
 import java.lang.reflect.Field;
 import java.net.InetSocketAddress;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
 import lombok.RequiredArgsConstructor;
 import org.geysermc.floodgate.api.handshake.HandshakeData;
 import org.geysermc.floodgate.api.logger.FloodgateLogger;
 import org.geysermc.floodgate.api.player.FloodgatePlayer;
 import org.geysermc.floodgate.config.ProxyFloodgateConfig;
 import org.geysermc.floodgate.player.FloodgateHandshakeHandler;
-import org.geysermc.floodgate.player.FloodgateHandshakeHandler.HandshakeResult;
 import org.geysermc.floodgate.util.Constants;
 
 @RequiredArgsConstructor
@@ -71,61 +73,95 @@ public final class VelocityProxyDataHandler extends ChannelInboundHandlerAdapter
 
     private final ProxyFloodgateConfig config;
     private final FloodgateHandshakeHandler handshakeHandler;
+    private final PacketBlocker blocker;
     private final AttributeKey<String> kickMessageAttribute;
     private final FloodgateLogger logger;
 
+    private final Queue<Object> packetQueue = Queues.newConcurrentLinkedQueue();
+
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        // prevent other packets from being handled while we handle the handshake packet
+        if (!packetQueue.isEmpty()) {
+            packetQueue.add(msg);
+            return;
+        }
+
         // we're only interested in the Handshake packet.
         // it should be the first packet but you never know
         if (HANDSHAKE_PACKET.isInstance(msg)) {
-            handleClientToProxy(ctx, msg);
-            ctx.pipeline().remove(this);
+            blocker.enable();
+            packetQueue.add(msg);
+
+            handleClientToProxy(ctx, msg).thenRun(() -> {
+                Object packet;
+                while ((packet = packetQueue.poll()) != null) {
+                    ctx.fireChannelRead(packet);
+                }
+                ctx.pipeline().remove(this);
+                blocker.disable();
+            });
+            return;
         }
 
         ctx.fireChannelRead(msg);
     }
 
-    private void handleClientToProxy(ChannelHandlerContext ctx, Object packet) {
+    private CompletableFuture<Void> handleClientToProxy(ChannelHandlerContext ctx, Object packet) {
         String address = getCastedValue(packet, HANDSHAKE_SERVER_ADDRESS);
 
-        HandshakeResult result = handshakeHandler.handle(ctx.channel(), address);
-        HandshakeData handshakeData = result.getHandshakeData();
+        return handshakeHandler.handle(ctx.channel(), address).thenAccept(result -> {
+            HandshakeData handshakeData = result.getHandshakeData();
 
-        InetSocketAddress newIp = result.getNewIp(ctx.channel());
-        if (newIp != null) {
-            Object connection = ctx.pipeline().get("handler");
-            setValue(connection, REMOTE_ADDRESS, newIp);
+            InetSocketAddress newIp = result.getNewIp(ctx.channel());
+            if (newIp != null) {
+                Object connection = ctx.pipeline().get("handler");
+                setValue(connection, REMOTE_ADDRESS, newIp);
+            }
+
+            if (handshakeData.getDisconnectReason() != null) {
+                ctx.channel().attr(kickMessageAttribute).set(handshakeData.getDisconnectReason());
+                return;
+            }
+
+            switch (result.getResultType()) {
+                case SUCCESS:
+                    break;
+                case EXCEPTION:
+                    ctx.channel().attr(kickMessageAttribute)
+                            .set(config.getDisconnect().getInvalidKey());
+                    return;
+                case INVALID_DATA_LENGTH:
+                    ctx.channel().attr(kickMessageAttribute)
+                            .set(config.getDisconnect().getInvalidArgumentsLength());
+                    return;
+                case TIMESTAMP_DENIED:
+                    ctx.channel().attr(kickMessageAttribute)
+                            .set(Constants.TIMESTAMP_DENIED_MESSAGE);
+                    return;
+                default: // only continue when SUCCESS
+                    return;
+            }
+
+            FloodgatePlayer player = result.getFloodgatePlayer();
+
+            setValue(packet, HANDSHAKE_SERVER_ADDRESS, handshakeData.getHostname());
+
+            logger.info("Floodgate player who is logged in as {} {} joined",
+                    player.getCorrectUsername(), player.getCorrectUniqueId());
+        }).handle((v, error) -> {
+            if (error != null) {
+                error.printStackTrace();
+            }
+            return v;
+        });
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        super.exceptionCaught(ctx, cause);
+        if (config.isDebug()) {
+            cause.printStackTrace();
         }
-
-        if (handshakeData.getDisconnectReason() != null) {
-            ctx.channel().attr(kickMessageAttribute).set(handshakeData.getDisconnectReason());
-            return;
-        }
-
-        switch (result.getResultType()) {
-            case SUCCESS:
-                break;
-            case EXCEPTION:
-                ctx.channel().attr(kickMessageAttribute)
-                        .set(config.getDisconnect().getInvalidKey());
-                return;
-            case INVALID_DATA_LENGTH:
-                ctx.channel().attr(kickMessageAttribute)
-                        .set(config.getDisconnect().getInvalidArgumentsLength());
-                return;
-            case TIMESTAMP_DENIED:
-                ctx.channel().attr(kickMessageAttribute).set(Constants.TIMESTAMP_DENIED_MESSAGE);
-                return;
-            default: // only continue when SUCCESS
-                return;
-        }
-
-        FloodgatePlayer player = result.getFloodgatePlayer();
-
-        setValue(packet, HANDSHAKE_SERVER_ADDRESS, handshakeData.getHostname());
-
-        logger.info("Floodgate player who is logged in as {} {} joined",
-                player.getCorrectUsername(), player.getCorrectUniqueId());
     }
 }