NetSocket implementation should resume reading after a TLS upgrade failure.

Motivation:

The recent channel TLS upgrade changes introduced a regression: the channel was only resumed on a success and not on a failure.

When a channel handler performs a TLS upgrade via the NetSocket API and fires a message, this message was not read by the NetSocket and ignored.

The NetSocket was still closed but the opportunity to consume the message was lost.

Changes:

Always resume reading the NetSocket after an upgrade result.

Author: vietj

vertx-core/src/main/java/io/vertx/core/net/impl/tcp/NetSocketImpl.java

@@ -167,9 +167,7 @@ private Future<Void> sslUpgrade(String serverName, SSLOptions sslOptions, ByteBu
         });
         return p.future();
       }).transform(ar -> {
-        if (ar.succeeded()) {
-          doResume();
-        }
+        doResume();
         return (Future<Void>) ar;
       });
     } else {

vertx-core/src/test/java/io/vertx/tests/net/NetTest.java

@@ -47,6 +47,9 @@
 import io.vertx.core.spi.tls.SslContextFactory;
 import io.vertx.core.transport.Transport;
 import io.vertx.test.core.*;
+import io.vertx.test.fakedns.DnsRecord;
+import io.vertx.test.fakedns.DnsServer;
+import io.vertx.test.fakedns.WithDnsServer;
 import io.vertx.test.proxy.*;
 import io.vertx.test.tls.Cert;
 import io.vertx.test.tls.Trust;
@@ -89,6 +92,9 @@
 @RunWith(VertxRunner.class)
 public class NetTest {
 
+  @Rule
+  public final DnsServer dnsServer = new DnsServer();
+
   public static class Provider implements VertxProvider {
 
     public VertxOptions options() {
@@ -2114,6 +2120,51 @@ public void testDirectTlsUpgrade(Checkpoint checkpoint) throws Exception {
     socket.write("test").await();
   }
 
+  @WithDnsServer(records = {@DnsRecord(name = "example.com")})
+  @Test
+  public void testDirectTlsUpgradeFailureResumeReading(Checkpoint checkpoint) throws Exception {
+    server = vertx.createNetServer(new NetServerOptions()
+      .setSsl(true)
+      .setKeyCertOptions(Cert.SERVER_JKS.get())
+    ).connectHandler(so -> {
+      fail();
+    });
+    server.listen(1234, "localhost").await();
+    Future<NetSocket> future = client.connect(new ConnectOptions()
+      .setPort(1234)
+      .setHost("example.com")
+      .setSslOptions(new ClientSSLOptions()
+        .setHostnameVerificationAlgorithm("HTTPS")
+        .setTrustAll(true)));
+    future.onComplete(TestUtils.onSuccess(socket -> {
+      NetSocketInternal soi = (NetSocketInternal) socket;
+      ChannelPipeline pipeline = soi
+        .channelHandlerContext()
+        .pipeline();
+      Object expectedMsg = new Object();
+      AtomicBoolean closed = new AtomicBoolean();
+      pipeline.addBefore("handler", "test", new ChannelDuplexHandler() {
+        @Override
+        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+          soi
+            .upgradeToSsl()
+            .onComplete(TestUtils.onFailure(err -> {
+              assertFalse(closed.get());
+              ctx.fireChannelRead(expectedMsg);
+            }));
+        }
+      });
+      soi.closeHandler(v -> {
+        closed.set(true);
+      });
+      soi.messageHandler(msg -> {
+        if (msg == expectedMsg) {
+          checkpoint.succeed();
+        }
+      });
+    }));
+  }
+
   public static class NativeVertxProvider implements VertxProvider {
     @Override
     public Vertx call() throws Exception {