The WebSocket client pool should not starve on connect error.

src/main/java/io/vertx/core/http/impl/WebSocketEndpoint.java

@@ -47,34 +47,26 @@ private static class Waiter {
     }
   }
 
-  private void tryConnect(ContextInternal ctx, Handler<AsyncResult<HttpClientConnection>> handler) {
+  private void connect(ContextInternal ctx, Handler<AsyncResult<HttpClientConnection>> handler) {
 
     class Listener implements Handler<AsyncResult<HttpClientConnection>> {
 
-      private void onEvict() {
-        decRefCount();
-        Waiter h;
-        synchronized (WebSocketEndpoint.this) {
-          if (--inflightConnections > maxPoolSize || waiters.isEmpty()) {
-            return;
-          }
-          h = waiters.poll();
-        }
-        tryConnect(h.context, h.handler);
-      }
-
       @Override
       public void handle(AsyncResult<HttpClientConnection> ar) {
         if (ar.succeeded()) {
           HttpClientConnection c = ar.result();
           if (incRefCount()) {
-            c.evictionHandler(v -> onEvict());
+            c.evictionHandler(v -> {
+              decRefCount();
+              release();
+            });
             handler.handle(Future.succeededFuture(c));
           } else {
             c.close();
             handler.handle(Future.failedFuture("Connection closed"));
           }
         } else {
+          release();
           handler.handle(Future.failedFuture(ar.cause()));
         }
       }
@@ -91,16 +83,33 @@ public void handle(AsyncResult<HttpClientConnection> ar) {
       .httpConnect(eventLoopContext, new Listener());
   }
 
-  @Override
-  public void requestConnection2(ContextInternal ctx, long timeout, Handler<AsyncResult<HttpClientConnection>> handler) {
+  private void release() {
+    Waiter h;
+    synchronized (WebSocketEndpoint.this) {
+      if (--inflightConnections > maxPoolSize || waiters.isEmpty()) {
+        return;
+      }
+      h = waiters.poll();
+    }
+    connect(h.context, h.handler);
+  }
+
+  private boolean tryAcquire(ContextInternal ctx, Handler<AsyncResult<HttpClientConnection>> handler) {
     synchronized (this) {
       if (inflightConnections >= maxPoolSize) {
         waiters.add(new Waiter(handler, ctx));
-        return;
+        return false;
       }
       inflightConnections++;
     }
-    tryConnect(ctx, handler);
+    return true;
+  }
+
+  @Override
+  public void requestConnection2(ContextInternal ctx, long timeout, Handler<AsyncResult<HttpClientConnection>> handler) {
+    if (tryAcquire(ctx, handler)) {
+      connect(ctx, handler);
+    }
   }
 
   @Override
@@ -110,6 +119,7 @@ void checkExpired() {
 
   @Override
   public void close() {
+    System.out.println("CLOSING POOL " + waiters.size());
     super.close();
     synchronized (this) {
       waiters.forEach(waiter -> {

src/test/java/io/vertx/core/http/WebSocketTest.java

@@ -38,6 +38,7 @@
 import io.vertx.core.net.impl.NetSocketInternal;
 import io.vertx.core.streams.ReadStream;
 import io.vertx.test.core.CheckingSender;
+import io.vertx.test.core.Repeat;
 import io.vertx.test.core.TestUtils;
 import io.vertx.test.core.VertxTestBase;
 import io.vertx.test.proxy.HAProxy;
@@ -47,7 +48,13 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
 import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
@@ -4026,4 +4033,88 @@ public void testCustomResponseHeadersBeforeUpgrade() {
       }));
       await();
   }
+
+  @Test
+  public void testPoolShouldNotStarveOnConnectError() throws Exception {
+
+    server = vertx.createHttpServer();
+
+    AtomicInteger accepted = new AtomicInteger();
+    server.webSocketHandler(ws -> {
+      assertTrue(accepted.getAndIncrement() == 0);
+    });
+
+    server.listen(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST).toCompletionStage().toCompletableFuture().get();
+
+    // This test requires a server socket to respond for the first connection
+    // Subsequent connections need to fail (connect error)
+    // Hence we need a custom proxy server in front of Vert.x HTTP server
+    InetAddress localhost = InetAddress.getByName("localhost");
+    ServerSocket proxy = new ServerSocket(0, 10, localhost);
+    int proxyPort = proxy.getLocalPort();
+
+    int maxConnections = 5;
+
+    client = vertx.createWebSocketClient(new WebSocketClientOptions()
+      .setMaxConnections(maxConnections)
+      .setConnectTimeout(4000));
+
+    Future<WebSocket> wsFut = client.connect(proxyPort, DEFAULT_HTTP_HOST, "/").andThen(onSuccess(v -> {
+    }));
+
+    Socket inbound = proxy.accept();
+    Socket outbound = new Socket();
+    outbound.connect(new InetSocketAddress(localhost, 8080));
+
+    class Pump extends Thread {
+      final InputStream src;
+      final OutputStream dst;
+      Pump(InputStream src, OutputStream dst) {
+        this.src = src;
+        this.dst = dst;
+      }
+      @Override
+      public void run() {
+        byte[] buffer = new byte[512];
+        while (true) {
+          try {
+            int l = src.read(buffer);
+            if (l == -1) {
+              break;
+            }
+            dst.write(buffer, 0, l);
+          } catch (IOException e) {
+            break;
+          }
+        }
+      }
+    }
+
+    Pump pump1 = new Pump(inbound.getInputStream(), outbound.getOutputStream());
+    Pump pump2 = new Pump(outbound.getInputStream(), inbound.getOutputStream());
+    pump1.start();
+    pump2.start();
+
+    // Finish handshake
+    wsFut.toCompletionStage().toCompletableFuture().get(10, TimeUnit.SECONDS);
+
+    proxy.close();
+
+    try {
+      int num = maxConnections + 10;
+      CountDownLatch latch = new CountDownLatch(num);
+      for (int i = 0;i < num;i++) {
+        client.connect(proxyPort, DEFAULT_HTTP_HOST, "/").onComplete(ar -> {
+          latch.countDown();
+        });
+      }
+
+      awaitLatch(latch, 10, TimeUnit.SECONDS);
+    } finally {
+      inbound.close();
+      outbound.close();
+      pump1.join(10_000);
+      pump2.join(10_000);
+    }
+  }
 }