Drop a PgSubscriber connection opened after close

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/pubsub/PgSubscriberImpl.java

@@ -162,11 +162,25 @@ private void tryConnect(long delayMillis, Completable<Void> handler) {
   }
 
   private void doConnect(Completable<Void> completionHandler) {
+    synchronized (this) {
+      if (closed) {
+        // Closed while a reconnect was scheduled
+        connecting = false;
+        completionHandler.complete(null, new IllegalStateException("Subscriber closed"));
+        return;
+      }
+    }
     PgConnection.connect(vertx, options).onComplete(ar -> handleConnectResult(completionHandler, ar));
   }
 
   private synchronized void handleConnectResult(Completable<Void> completionHandler, AsyncResult<PgConnection> ar1) {
     connecting = false;
+    if (ar1.succeeded() && closed) {
+      // Closed while connecting: do not keep the connection
+      ar1.result().close();
+      completionHandler.complete(null, new IllegalStateException("Subscriber closed"));
+      return;
+    }
     if (ar1.succeeded()) {
       conn = ar1.result();
       conn.notificationHandler(PgSubscriberImpl.this::handleNotification);

vertx-pg-client/src/test/java/io/vertx/tests/pgclient/PubSubTest.java

@@ -132,6 +132,31 @@ private void testConnect(TestContext ctx, String channel1Name, String channel2Na
     notifiedLatch.awaitSuccess(10000);
   }
 
+  @Test
+  public void testCloseWhileConnecting(TestContext ctx) {
+    ProxyServer proxy = ProxyServer.create(vertx, options.getPort(), options.getHost());
+    AtomicReference<ProxyServer.Connection> pending = new AtomicReference<>();
+    Async accepted = ctx.async();
+    // Hold the connection until the subscriber is closed
+    proxy.proxyHandler(conn -> {
+      pending.set(conn);
+      accepted.complete();
+    });
+    Async listening = ctx.async();
+    proxy.listen(8080, "localhost", ctx.asyncAssertSuccess(v -> listening.complete()));
+    listening.awaitSuccess(10000);
+    subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions(options).setPort(8080).setHost("localhost"));
+    subscriber.channel("the_channel").handler(notification -> ctx.fail("Notification after close: " + notification));
+    Async connected = ctx.async();
+    subscriber.connect().onComplete(ar -> connected.complete());
+    accepted.awaitSuccess(10000);
+    subscriber.close();
+    pending.get().connect();
+    connected.awaitSuccess(10000);
+    ctx.assertTrue(subscriber.closed());
+    ctx.assertNull(subscriber.actualConnection());
+  }
+
   @Test
   public void testSubscribe(TestContext ctx) {
     testSubscribe(ctx, "the_channel");