Adapt to the changes for asynchronous connect in AbstractChannel (#2365)

- Adapt to the changes for asynchronous connect in AbstractChannel
netty/netty#12570
- Adapt the CI to the changes made in Future interface
netty/netty#12569

Related to #1873

.github/workflows/check_transport.yml

@@ -60,16 +60,22 @@ jobs:
           fetch-depth: 0 #needed by spotless
       - uses: actions/checkout@v3
         with:
-          repository: netty-contrib/codec-haproxy
+          # TODO temporary https://github.com/netty-contrib/codec-haproxy/pull/9
+          repository: violetagg/codec-haproxy
           path: codec-haproxy
+          ref: use-asStage
       - uses: actions/checkout@v3
         with:
-          repository: netty-contrib/codec-extras
+          # TODO temporary https://github.com/netty-contrib/codec-extras/pull/10
+          repository: violetagg/codec-extras
           path: codec-extras
+          ref: use-asStage
       - uses: actions/checkout@v3
         with:
-          repository: netty-contrib/socks-proxy
+          # TODO temporary https://github.com/netty-contrib/socks-proxy/pull/12
+          repository: violetagg/socks-proxy
           path: socks-proxy
+          ref: use-asStage
       - uses: gradle/wrapper-validation-action@v1
       - name: Set up JDK 17
         uses: actions/setup-java@v3

reactor-netty-core/src/test/java/reactor/netty/ReactorNettyTest.java

@@ -21,13 +21,16 @@
 import io.netty5.channel.ChannelMetadata;
 import io.netty5.channel.ChannelOutboundBuffer;
 import io.netty5.channel.ChannelShutdownDirection;
+import io.netty5.channel.EventLoop;
 import io.netty5.channel.embedded.EmbeddedChannel;
-import io.netty5.util.concurrent.Promise;
+import io.netty5.util.concurrent.Future;
 import org.junit.jupiter.api.Test;
 import reactor.util.annotation.Nullable;
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -119,7 +122,7 @@ static class TestChannel extends AbstractChannel<Channel, SocketAddress, SocketA
 		}
 
 		TestChannel(Channel parent, @Nullable SocketAddress localAddress, @Nullable SocketAddress remoteAddress) {
-			super(parent, parent.executor());
+			super(parent, new TestEventLoop());
 			this.localAddress = localAddress;
 			this.remoteAddress = remoteAddress;
 		}
@@ -159,7 +162,13 @@ protected void doWrite(ChannelOutboundBuffer in) {
 		}
 
 		@Override
-		protected void connectTransport(SocketAddress remoteAddress, SocketAddress localAddress, Promise<Void> promise) {
+		protected boolean doConnect(SocketAddress socketAddress, SocketAddress socketAddress1) {
+			return false;
+		}
+
+		@Override
+		protected boolean doFinishConnect(SocketAddress socketAddress) {
+			return false;
 		}
 
 		@Override
@@ -188,6 +197,89 @@ public ChannelMetadata metadata() {
 		}
 	}
 
+	static final class TestEventLoop implements EventLoop {
+
+		@Override
+		public Future<Void> registerForIo(Channel channel) {
+			return null;
+		}
+
+		@Override
+		public Future<Void> deregisterForIo(Channel channel) {
+			return null;
+		}
+
+		@Override
+		public boolean isCompatible(Class<? extends Channel> channelType) {
+			return true;
+		}
+
+		@Override
+		public boolean inEventLoop(Thread thread) {
+			return false;
+		}
+
+		@Override
+		public boolean isShuttingDown() {
+			return false;
+		}
+
+		@Override
+		public boolean isShutdown() {
+			return false;
+		}
+
+		@Override
+		public Future<Void> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
+			return null;
+		}
+
+		@Override
+		public Future<Void> terminationFuture() {
+			return null;
+		}
+
+		@Override
+		public Future<Void> submit(Runnable task) {
+			return null;
+		}
+
+		@Override
+		public <T> Future<T> submit(Runnable task, T result) {
+			return null;
+		}
+
+		@Override
+		public <T> Future<T> submit(Callable<T> task) {
+			return null;
+		}
+
+		@Override
+		public Future<Void> schedule(Runnable task, long delay, TimeUnit unit) {
+			return null;
+		}
+
+		@Override
+		public <V> Future<V> schedule(Callable<V> task, long delay, TimeUnit unit) {
+			return null;
+		}
+
+		@Override
+		public Future<Void> scheduleAtFixedRate(Runnable task, long initialDelay, long period, TimeUnit unit) {
+			return null;
+		}
+
+		@Override
+		public Future<Void> scheduleWithFixedDelay(Runnable task, long initialDelay, long delay, TimeUnit unit) {
+			return null;
+		}
+
+		@Override
+		public void execute(Runnable task) {
+
+		}
+	}
+
 	static final class ChildTestChannel extends TestChannel {
 
 		ChildTestChannel(Channel parent) {