Update mcpl impl

core/src/main/java/org/geysermc/geyser/network/netty/LocalSession.java

@@ -27,14 +27,25 @@ package org.geysermc.geyser.network.netty;
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBufAllocator;
-import io.netty.channel.*;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.DefaultEventLoopGroup;
 import io.netty.channel.unix.PreferredDirectByteBufAllocator;
-import io.netty.handler.codec.haproxy.*;
+import io.netty.handler.codec.haproxy.HAProxyCommand;
+import io.netty.handler.codec.haproxy.HAProxyMessage;
+import io.netty.handler.codec.haproxy.HAProxyMessageEncoder;
+import io.netty.handler.codec.haproxy.HAProxyProtocolVersion;
+import io.netty.handler.codec.haproxy.HAProxyProxiedProtocol;
+import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.handler.timeout.WriteTimeoutHandler;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.geysermc.mcprotocollib.network.BuiltinFlags;
 import org.geysermc.mcprotocollib.network.codec.PacketCodecHelper;
 import org.geysermc.mcprotocollib.network.packet.PacketProtocol;
+import org.geysermc.mcprotocollib.network.tcp.FlushHandler;
 import org.geysermc.mcprotocollib.network.tcp.TcpFlowControlHandler;
 import org.geysermc.mcprotocollib.network.tcp.TcpPacketCodec;
 import org.geysermc.mcprotocollib.network.tcp.TcpPacketSizer;
@@ -44,6 +55,7 @@ import org.geysermc.mcprotocollib.protocol.codec.MinecraftCodecHelper;
 import java.net.Inet4Address;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -76,7 +88,6 @@ public final class LocalSession extends TcpSession {
                 () -> DEFAULT_EVENT_LOOP_GROUP.shutdownGracefully(100, 500, TimeUnit.MILLISECONDS)));
         }
 
-        try {
         final Bootstrap bootstrap = new Bootstrap();
         bootstrap.channel(LocalChannelWithRemoteAddress.class);
         bootstrap.handler(new ChannelInitializer<LocalChannelWithRemoteAddress>() {
@@ -86,18 +97,21 @@ public final class LocalSession extends TcpSession {
                 PacketProtocol protocol = getPacketProtocol();
                 protocol.newClientSession(LocalSession.this, transferring);
 
-                    refreshReadTimeoutHandler(channel);
-                    refreshWriteTimeoutHandler(channel);
-
                 ChannelPipeline pipeline = channel.pipeline();
-                    pipeline.addLast("sizer", new TcpPacketSizer(LocalSession.this, protocol.getPacketHeader().getLengthSize()));
+
+                initializeHAProxySupport(channel);
+
+                pipeline.addLast("read-timeout", new ReadTimeoutHandler(getFlag(BuiltinFlags.READ_TIMEOUT, 30)));
+                pipeline.addLast("write-timeout", new WriteTimeoutHandler(getFlag(BuiltinFlags.WRITE_TIMEOUT, 0)));
+
+                pipeline.addLast("sizer", new TcpPacketSizer(protocol.getPacketHeader(), getCodecHelper()));
+
                 pipeline.addLast("flow-control", new TcpFlowControlHandler());
                 pipeline.addLast("codec", new TcpPacketCodec(LocalSession.this, true));
+                pipeline.addLast("flush-handler", new FlushHandler());
                 pipeline.addLast("manager", LocalSession.this);
-
-                    addHAProxySupport(pipeline);
             }
-            }).group(DEFAULT_EVENT_LOOP_GROUP).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout() * 1000);
+        }).group(DEFAULT_EVENT_LOOP_GROUP).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getFlag(BuiltinFlags.CLIENT_CONNECT_TIMEOUT, 30) * 1000);
 
         if (PREFERRED_DIRECT_BYTE_BUF_ALLOCATOR != null) {
             bootstrap.option(ChannelOption.ALLOCATOR, PREFERRED_DIRECT_BYTE_BUF_ALLOCATOR);
@@ -105,13 +119,17 @@ public final class LocalSession extends TcpSession {
 
         bootstrap.remoteAddress(targetAddress);
 
-            bootstrap.connect().addListener((future) -> {
+        CompletableFuture<Void> handleFuture = new CompletableFuture<>();
-                if (!future.isSuccess()) {
+        bootstrap.connect().addListener((futureListener) -> {
-                    exceptionCaught(null, future.cause());
+            if (!futureListener.isSuccess()) {
+                exceptionCaught(null, futureListener.cause());
             }
+
+            handleFuture.complete(null);
         });
-        } catch (Throwable t) {
+
-            exceptionCaught(null, t);
+        if (wait) {
+            handleFuture.join();
         }
     }
 
@@ -120,32 +138,22 @@ public final class LocalSession extends TcpSession {
         return (MinecraftCodecHelper) this.codecHelper;
     }
 
-    // TODO duplicate code
+    private void initializeHAProxySupport(Channel channel) {
-    private void addHAProxySupport(ChannelPipeline pipeline) {
         InetSocketAddress clientAddress = getFlag(BuiltinFlags.CLIENT_PROXIED_ADDRESS);
-        if (getFlag(BuiltinFlags.ENABLE_CLIENT_PROXY_PROTOCOL, false) && clientAddress != null) {
+        if (clientAddress == null) {
-            pipeline.addFirst("proxy-protocol-packet-sender", new ChannelInboundHandlerAdapter() {
+            return;
-                @Override
-                public void channelActive(@NonNull ChannelHandlerContext ctx) throws Exception {
-                    HAProxyProxiedProtocol proxiedProtocol = clientAddress.getAddress() instanceof Inet4Address ? HAProxyProxiedProtocol.TCP4 : HAProxyProxiedProtocol.TCP6;
-                    InetSocketAddress remoteAddress;
-                    if (ctx.channel().remoteAddress() instanceof InetSocketAddress) {
-                        remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
-                    } else {
-                        remoteAddress = new InetSocketAddress(host, port);
         }
-                    ctx.channel().writeAndFlush(new HAProxyMessage(
+
+        channel.pipeline().addLast("proxy-protocol-encoder", HAProxyMessageEncoder.INSTANCE);
+        HAProxyProxiedProtocol proxiedProtocol = clientAddress.getAddress() instanceof Inet4Address ? HAProxyProxiedProtocol.TCP4 : HAProxyProxiedProtocol.TCP6;
+        InetSocketAddress remoteAddress = (InetSocketAddress) channel.remoteAddress();
+        channel.writeAndFlush(new HAProxyMessage(
             HAProxyProtocolVersion.V2, HAProxyCommand.PROXY, proxiedProtocol,
             clientAddress.getAddress().getHostAddress(), remoteAddress.getAddress().getHostAddress(),
             clientAddress.getPort(), remoteAddress.getPort()
-                    ));
+        )).addListener(future -> {
-                    ctx.pipeline().remove(this);
+            channel.pipeline().remove("proxy-protocol-encoder");
-                    ctx.pipeline().remove("proxy-protocol-encoder");
-                    super.channelActive(ctx);
-                }
         });
-            pipeline.addFirst("proxy-protocol-encoder", HAProxyMessageEncoder.INSTANCE);
-        }
     }
 
     /**

core/src/main/java/org/geysermc/geyser/session/GeyserSession.java

@@ -958,7 +958,6 @@ public class GeyserSession implements GeyserConnection, GeyserCommandSource {
         this.downstream.getSession().setFlag(MinecraftConstants.FOLLOW_TRANSFERS, false);
 
         if (geyser.getConfig().getRemote().isUseProxyProtocol()) {
-            downstream.setFlag(BuiltinFlags.ENABLE_CLIENT_PROXY_PROTOCOL, true);
             downstream.setFlag(BuiltinFlags.CLIENT_PROXIED_ADDRESS, upstream.getAddress());
         }
         if (geyser.getConfig().isForwardPlayerPing()) {

gradle/libs.versions.toml

@@ -15,7 +15,7 @@ protocol-common = "3.0.0.Beta4-20240828.162251-1"
 protocol-codec = "3.0.0.Beta4-20240828.162251-1"
 raknet = "1.0.0.CR3-20240416.144209-1"
 minecraftauth = "4.1.1-20240806.235051-7"
-mcprotocollib = "324f066"
+mcprotocollib = "f9b86e5"
 adventure = "4.14.0"
 adventure-platform = "4.3.0"
 junit = "5.9.2"