Made backend connect non blocking

src/main/java/net/raphimc/viaproxy/proxy/client2proxy/Client2ProxyHandler.java

@@ -43,9 +43,7 @@ import net.raphimc.viaproxy.proxy.session.BedrockProxyConnection;
 import net.raphimc.viaproxy.proxy.session.DummyProxyConnection;
 import net.raphimc.viaproxy.proxy.session.ProxyConnection;
 import net.raphimc.viaproxy.proxy.session.UserOptions;
-import net.raphimc.viaproxy.proxy.util.CloseAndReturn;
+import net.raphimc.viaproxy.proxy.util.*;
-import net.raphimc.viaproxy.proxy.util.ExceptionUtil;
-import net.raphimc.viaproxy.proxy.util.HAProxyUtil;
 import net.raphimc.viaproxy.util.ArrayHelper;
 import net.raphimc.viaproxy.util.logging.Logger;
 
@@ -117,7 +115,7 @@ public class Client2ProxyHandler extends SimpleChannelInboundHandler<IPacket> {
             this.proxyConnection.kickClient("§cYour client version is not supported by ViaProxy!");
         }
 
-        String[] handshakeParts = new String[]{packet.address};
+        final String[] handshakeParts;
         if (Options.PLAYER_INFO_FORWARDING) {
             handshakeParts = new String[3];
             final String[] receivedParts = packet.address.split("\0");
@@ -133,6 +131,8 @@ public class Client2ProxyHandler extends SimpleChannelInboundHandler<IPacket> {
             if (handshakeParts[2] == null) {
                 this.proxyConnection.kickClient("§cMissing player UUID in handshake. Ensure that your proxy has player info forwarding enabled.");
             }
+        } else {
+            handshakeParts = new String[]{packet.address};
         }
 
         String connectIP = Options.CONNECT_ADDRESS;
@@ -223,29 +223,35 @@ public class Client2ProxyHandler extends SimpleChannelInboundHandler<IPacket> {
         this.proxyConnection.getPacketHandlers().add(new ResourcePackPacketHandler(this.proxyConnection));
         this.proxyConnection.getPacketHandlers().add(new UnexpectedPacketHandler(this.proxyConnection));
 
+        ChannelUtil.disableAutoRead(this.proxyConnection.getC2P());
         Logger.u_info("connect", this.proxyConnection.getC2P().remoteAddress(), this.proxyConnection.getGameProfile(), "[" + clientVersion.getName() + " <-> " + serverVersion.getName() + "] Connecting to " + serverAddress.getAddress() + ":" + serverAddress.getPort());
-        try {
         PluginManager.EVENT_MANAGER.call(new ConnectEvent(this.proxyConnection));
-            this.proxyConnection.connectToServer(serverAddress, serverVersion);
-        } catch (Throwable e) {
-            if (e instanceof ConnectException || e instanceof UnresolvedAddressException) { // Trust me, this is not always false
-                this.proxyConnection.kickClient("§cCould not connect to the backend server!\n§cTry again in a few seconds.");
-            } else {
-                Logger.LOGGER.error("Error while connecting to the backend server", e);
-                this.proxyConnection.kickClient("§cAn error occurred while connecting to the backend server: " + e.getMessage() + "\n§cCheck the console for more information.");
-            }
-        }
 
+        this.proxyConnection.connectToServer(serverAddress, serverVersion).addListeners((ThrowingChannelFutureListener) f -> {
+            if (f.isSuccess()) {
                 if (Options.SERVER_HAPROXY_PROTOCOL) {
                     this.proxyConnection.getChannel().writeAndFlush(HAProxyUtil.createMessage(this.proxyConnection.getC2P(), this.proxyConnection.getChannel(), clientVersion)).addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
                 }
 
                 handshakeParts[0] = serverAddress.getAddress();
-        this.proxyConnection.getChannel().writeAndFlush(new C2SHandshakePacket(clientVersion.getOriginalVersion(), String.join("\0", handshakeParts), serverAddress.getPort(), packet.intendedState)).addListeners(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE, (ChannelFutureListener) f -> {
+                this.proxyConnection.getChannel().writeAndFlush(new C2SHandshakePacket(clientVersion.getOriginalVersion(), String.join("\0", handshakeParts), serverAddress.getPort(), packet.intendedState)).addListeners(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE, (ChannelFutureListener) f2 -> {
-            if (f.isSuccess()) {
+                    if (f2.isSuccess()) {
                         this.proxyConnection.setP2sConnectionState(packet.intendedState);
                     }
                 });
+
+                ChannelUtil.restoreAutoRead(this.proxyConnection.getC2P());
+            }
+        }, (ThrowingChannelFutureListener) f -> {
+            if (!f.isSuccess()) {
+                if (f.cause() instanceof ConnectException || f.cause() instanceof UnresolvedAddressException) {
+                    this.proxyConnection.kickClient("§cCould not connect to the backend server!\n§cTry again in a few seconds.");
+                } else {
+                    Logger.LOGGER.error("Error while connecting to the backend server", f.cause());
+                    this.proxyConnection.kickClient("§cAn error occurred while connecting to the backend server: " + f.cause().getMessage() + "\n§cCheck the console for more information.");
+                }
+            }
+        });
     }
 
 }

src/main/java/net/raphimc/viaproxy/proxy/client2proxy/passthrough/PassthroughClient2ProxyHandler.java

@@ -26,8 +26,10 @@ import net.raphimc.viaproxy.plugins.events.Proxy2ServerHandlerCreationEvent;
 import net.raphimc.viaproxy.proxy.proxy2server.passthrough.PassthroughProxy2ServerChannelInitializer;
 import net.raphimc.viaproxy.proxy.proxy2server.passthrough.PassthroughProxy2ServerHandler;
 import net.raphimc.viaproxy.proxy.session.LegacyProxyConnection;
+import net.raphimc.viaproxy.proxy.util.ChannelUtil;
 import net.raphimc.viaproxy.proxy.util.ExceptionUtil;
 import net.raphimc.viaproxy.proxy.util.HAProxyUtil;
+import net.raphimc.viaproxy.proxy.util.ThrowingChannelFutureListener;
 import net.raphimc.viaproxy.util.logging.Logger;
 
 import java.util.function.Supplier;
@@ -70,17 +72,26 @@ public class PassthroughClient2ProxyHandler extends SimpleChannelInboundHandler<
         this.proxyConnection = new LegacyProxyConnection(handlerSupplier, PassthroughProxy2ServerChannelInitializer::new, c2pChannel);
         this.proxyConnection.getC2P().attr(LegacyProxyConnection.LEGACY_PROXY_CONNECTION_ATTRIBUTE_KEY).set(this.proxyConnection);
 
-        try {
+        final ServerAddress serverAddress = this.getServerAddress();
-            this.proxyConnection.connect(this.getServerAddress());
+
-        } catch (Throwable e) {
+        ChannelUtil.disableAutoRead(this.proxyConnection.getC2P());
-            Logger.LOGGER.error("Failed to connect to target server", e);
+        Logger.u_info("connect", this.proxyConnection.getC2P().remoteAddress(), null, "[Legacy <-> Legacy] Connecting to " + serverAddress.getAddress() + ":" + serverAddress.getPort());
-            this.proxyConnection = null;
+
-            c2pChannel.close();
+        this.proxyConnection.connect(serverAddress).addListeners((ThrowingChannelFutureListener) f -> {
+            if (f.isSuccess()) {
+                if (Options.SERVER_HAPROXY_PROTOCOL) {
+                    this.proxyConnection.getChannel().writeAndFlush(HAProxyUtil.createMessage(this.proxyConnection.getC2P(), this.proxyConnection.getChannel(), null)).addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
                 }
 
-        if (Options.SERVER_HAPROXY_PROTOCOL) {
+                ChannelUtil.restoreAutoRead(this.proxyConnection.getC2P());
-            this.proxyConnection.getChannel().writeAndFlush(HAProxyUtil.createMessage(c2pChannel, this.proxyConnection.getChannel(), null)).addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
             }
+        }, (ThrowingChannelFutureListener) f -> {
+            if (!f.isSuccess()) {
+                Logger.LOGGER.error("Failed to connect to target server", f.cause());
+                this.proxyConnection.getC2P().close();
+                this.proxyConnection = null;
+            }
+        });
     }
 
     protected ServerAddress getServerAddress() {

src/main/java/net/raphimc/viaproxy/proxy/session/BedrockProxyConnection.java

@@ -19,7 +19,7 @@ package net.raphimc.viaproxy.proxy.session;
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.epoll.Epoll;
@@ -73,25 +73,24 @@ public class BedrockProxyConnection extends ProxyConnection {
     }
 
     @Override
-    public void connectToServer(ServerAddress serverAddress, VersionEnum targetVersion) {
+    public ChannelFuture connectToServer(ServerAddress serverAddress, VersionEnum targetVersion) {
         if (this.getC2pConnectionState() == ConnectionState.STATUS) {
             RStream.of(this).withSuper().fields().by("serverAddress").set(serverAddress);
             RStream.of(this).withSuper().fields().by("serverVersion").set(targetVersion);
-            this.ping(serverAddress);
+            return this.ping(serverAddress);
         } else {
-            super.connectToServer(serverAddress, targetVersion);
+            return super.connectToServer(serverAddress, targetVersion);
         }
     }
 
-    private void ping(final ServerAddress serverAddress) {
+    private ChannelFuture ping(final ServerAddress serverAddress) {
-        if (this.channelFuture == null) {
+        if (this.channelFuture == null) this.initialize(new Bootstrap());
-            this.initialize(new Bootstrap());
-        }
-        this.getChannel().bind(new InetSocketAddress(0)).addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE).syncUninterruptibly();
 
         this.getChannel().pipeline().replace(VLPipeline.VIABEDROCK_FRAME_ENCAPSULATION_HANDLER_NAME, "ping_encapsulation", new PingEncapsulationCodec(serverAddress.toSocketAddress()));
         this.getChannel().pipeline().remove(VLPipeline.VIABEDROCK_PACKET_ENCAPSULATION_HANDLER_NAME);
         this.getChannel().pipeline().remove(MCPipeline.SIZER_HANDLER_NAME);
+
+        return this.getChannel().bind(new InetSocketAddress(0));
     }
 
 }

src/main/java/net/raphimc/viaproxy/proxy/session/DummyProxyConnection.java

@@ -22,6 +22,7 @@ import com.viaversion.viaversion.api.connection.UserConnection;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
 import net.raphimc.netminecraft.constants.ConnectionState;
 import net.raphimc.netminecraft.packet.impl.login.C2SLoginHelloPacket1_7;
 import net.raphimc.netminecraft.util.ServerAddress;
@@ -42,7 +43,7 @@ public class DummyProxyConnection extends ProxyConnection {
     }
 
     @Override
-    public void connectToServer(ServerAddress serverAddress, VersionEnum targetVersion) {
+    public ChannelFuture connectToServer(ServerAddress serverAddress, VersionEnum targetVersion) {
         throw new UnsupportedOperationException();
     }
 

src/main/java/net/raphimc/viaproxy/proxy/session/LegacyProxyConnection.java

@@ -18,10 +18,7 @@
 package net.raphimc.viaproxy.proxy.session;
 
 import io.netty.bootstrap.Bootstrap;
-import io.netty.channel.Channel;
+import io.netty.channel.*;
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
 import io.netty.util.AttributeKey;
 import net.raphimc.netminecraft.netty.connection.NetClient;
 import net.raphimc.netminecraft.util.ServerAddress;
@@ -45,12 +42,6 @@ public class LegacyProxyConnection extends NetClient {
         return channel.attr(LEGACY_PROXY_CONNECTION_ATTRIBUTE_KEY).get();
     }
 
-    @Override
-    public void connect(final ServerAddress serverAddress) {
-        this.serverAddress = serverAddress;
-        super.connect(serverAddress);
-    }
-
     @Override
     public void initialize(final Bootstrap bootstrap) {
         bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 4_000);
@@ -58,6 +49,12 @@ public class LegacyProxyConnection extends NetClient {
         super.initialize(bootstrap);
     }
 
+    @Override
+    public ChannelFuture connect(final ServerAddress serverAddress) {
+        this.serverAddress = serverAddress;
+        return super.connect(serverAddress);
+    }
+
     public Channel getC2P() {
         return this.c2p;
     }

src/main/java/net/raphimc/viaproxy/proxy/session/ProxyConnection.java

@@ -97,7 +97,7 @@ public class ProxyConnection extends NetClient {
 
     @Override
     @Deprecated
-    public void connect(final ServerAddress serverAddress) {
+    public ChannelFuture connect(final ServerAddress serverAddress) {
         throw new UnsupportedOperationException();
     }
 
@@ -108,10 +108,10 @@ public class ProxyConnection extends NetClient {
         super.initialize(bootstrap);
     }
 
-    public void connectToServer(final ServerAddress serverAddress, final VersionEnum targetVersion) {
+    public ChannelFuture connectToServer(final ServerAddress serverAddress, final VersionEnum targetVersion) {
         this.serverAddress = serverAddress;
         this.serverVersion = targetVersion;
-        super.connect(serverAddress);
+        return super.connect(serverAddress);
     }
 
     public Channel getC2P() {

src/main/java/net/raphimc/viaproxy/proxy/util/ThrowingChannelFutureListener.java

@@ -0,0 +1,36 @@
+/*
+ * This file is part of ViaProxy - https://github.com/RaphiMC/ViaProxy
+ * Copyright (C) 2023 RK_01/RaphiMC and contributors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package net.raphimc.viaproxy.proxy.util;
+
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+
+public interface ThrowingChannelFutureListener extends ChannelFutureListener {
+
+    @Override
+    default void operationComplete(ChannelFuture future) {
+        try {
+            this.operationComplete0(future);
+        } catch (Throwable cause) {
+            future.channel().pipeline().fireExceptionCaught(cause);
+        }
+    }
+
+    void operationComplete0(ChannelFuture future) throws Throwable;
+
+}