Correctly handle pending reads during a start tls upgrade.

Motivation:

NetSocket implementation does pause the socket on a TLS upgrade but relies on the VertxConnection pause that still cumulates buffers in its pending queue.

Since obtaining an SSL context is asynchronous, there can be a client hello sent by the client that will be read and sit in the VertxConnection pending list and will be missed by the SSL handler.

Changes:

During a TLS upgrade we want buffers to actually remains in the channel so we should pause channel reading instead of pausing the VertxConnection to ensure the pipeline won't see any read during the SSL context resolution.

Author: vietj

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

@@ -127,7 +127,7 @@ private Future<Void> sslUpgrade(String serverName, SSLOptions sslOptions, ByteBu
       applicationProtocols = null;
     }
     if (chctx.pipeline().get("ssl") == null) {
-      doPause();
+      chctx.channel().config().setAutoRead(false);
       Future<SslChannelProvider> f;
       if (sslOptions instanceof ClientSSLOptions) {
         ClientSSLOptions clientSSLOptions =  (ClientSSLOptions) sslOptions;
@@ -145,6 +145,7 @@ private Future<Void> sslUpgrade(String serverName, SSLOptions sslOptions, ByteBu
         ChannelPromise promise = chctx.newPromise();
         writeToChannel(msg, true, promise);
         promise.addListener(res -> {
+          chctx.channel().config().setAutoRead(true);
           if (res.isSuccess()) {
             ChannelPromise channelPromise = chctx.newPromise();
             chctx.pipeline().addFirst("handshaker", new SslHandshakeCompletionHandler(channelPromise));
@@ -159,13 +160,16 @@ private Future<Void> sslUpgrade(String serverName, SSLOptions sslOptions, ByteBu
             }
             chctx.pipeline().addFirst("ssl", sslHandler);
             channelPromise.addListener(p);
+            doPause();
           } else {
             p.fail(res.cause());
           }
         });
         return p.future();
       }).transform(ar -> {
-        doResume();
+        if (ar.succeeded()) {
+          doResume();
+        }
         return (Future<Void>) ar;
       });
     } else {

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

@@ -2089,6 +2089,31 @@ public void start(Promise<Void> startPromise) {
     }
   }
 
+  @Test
+  public void testDirectTlsUpgrade(Checkpoint checkpoint) throws Exception {
+    server.connectHandler(socket -> {
+      socket.upgradeToSsl(new ServerSSLOptions().setKeyCertOptions(Cert.SERVER_JKS.get()))
+        .onComplete(TestUtils.onSuccess(v1 -> {
+        socket.handler(socket::write);
+        socket.endHandler(v2 -> socket.end());
+      }));
+    });
+    server.listen(1234, "localhost").await();
+    NetSocket socket = client.connect(new ConnectOptions()
+      .setPort(1234)
+      .setHost("localhost")
+      .setSsl(true)
+      .setSslOptions(new ClientSSLOptions()
+      .setHostnameVerificationAlgorithm("")
+      .setTrustAll(true))).await();
+    socket.handler(chunk -> {
+      assertEquals("test", chunk.toString());
+      socket.end();
+    });
+    socket.endHandler(v -> checkpoint.succeed());
+    socket.write("test").await();
+  }
+
   public static class NativeVertxProvider implements VertxProvider {
     @Override
     public Vertx call() throws Exception {