Adapt to Closeable contract change

Author: vietj

vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleConnectionFactory.java

@@ -10,6 +10,7 @@
  */
 package io.vertx.oracleclient.impl;
 
+import io.vertx.core.Completable;
 import io.vertx.core.Context;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
@@ -40,8 +41,8 @@ public OracleConnectionFactory(VertxInternal vertx) {
   }
 
   @Override
-  public void close(Promise<Void> promise) {
-    promise.complete();
+  public void close(Completable<Void> promise) {
+    promise.succeed();
   }
 
   private OracleDataSource getDatasource(SqlConnectOptions options) {

vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleJdbcConnection.java

@@ -121,7 +121,7 @@ public DatabaseMetadata getDatabaseMetaData() {
   }
 
   @Override
-  public void close(Holder holder, Promise<Void> promise) {
+  public void close(Holder holder, Completable<Void> promise) {
     if (Vertx.currentContext() == context) {
       Future<Void> future;
       if (closePromise == null) {

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/ConnectionFactoryBase.java

@@ -10,6 +10,7 @@
  */
 package io.vertx.sqlclient.impl;
 
+import io.vertx.core.Completable;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.net.NetClient;
@@ -46,8 +47,8 @@ protected ConnectionFactoryBase(VertxInternal vertx, NetClientOptions tcpOptions
     this.tcpOptions = tcpOptions;
   }
 
-  private void doClose(Promise<Void> p) {
-    client.close().onComplete(ar -> p.complete());
+  private void doClose(Completable<Void> p) {
+    client.close().onComplete(ar -> p.succeed());
   }
 
   public static ContextInternal asEventLoopContext(ContextInternal ctx) {
@@ -64,7 +65,7 @@ public Future<Connection> connect(ContextInternal context, C options) {
   }
 
   @Override
-  public void close(Promise<Void> promise) {
+  public void close(Completable<Void> promise) {
     clientCloseFuture.close(promise);
   }
 

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SocketConnectionBase.java

@@ -178,7 +178,7 @@ public int getSecretKey() {
   }
 
   @Override
-  public void close(Holder holder, Promise<Void> promise) {
+  public void close(Holder holder, Completable<Void> promise) {
     if (Vertx.currentContext() == context) {
       Channel ch = socket.channelHandlerContext().channel();
       if (status == Status.CONNECTED) {
@@ -188,7 +188,7 @@ public void close(Holder holder, Promise<Void> promise) {
         checkPending();
       }
       ch.closeFuture()
-        .addListener((ChannelFutureListener) channelFuture -> promise.complete());
+        .addListener((ChannelFutureListener) channelFuture -> promise.succeed());
     } else {
       context.runOnContext(v -> close(holder, promise));
     }

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java

@@ -415,11 +415,11 @@ public void init(Holder holder) {
     }
 
     @Override
-    public void close(Holder holder, Promise<Void> promise) {
+    public void close(Holder holder, Completable<Void> promise) {
       doClose(holder, promise);
     }
 
-    private void doClose(Holder holder, Promise<Void> promise) {
+    private void doClose(Holder holder, Completable<Void> promise) {
       if (holder != this.holder) {
         String msg;
         if (this.holder == null) {
@@ -434,7 +434,7 @@ private void doClose(Holder holder, Promise<Void> promise) {
         Promise<ConnectResult<PooledConnection>> resultHandler = poolCallback;
         if (resultHandler != null) {
           poolCallback = null;
-          promise.complete();
+          promise.succeed();
           resultHandler.complete(new ConnectResult<>(this, pipelined ? conn.pipeliningLimit() : 1, 0));
           return;
         }
@@ -446,12 +446,12 @@ private void doClose(Holder holder, Promise<Void> promise) {
       }
     }
 
-    private void cleanup(Promise<Void> promise) {
+    private void cleanup(Completable<Void> promise) {
       Lease<PooledConnection> l = this.lease;
       this.lease = null;
       refresh();
       l.recycle();
-      promise.complete();
+      promise.succeed();
     }
 
     @Override

vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/Connection.java

@@ -17,6 +17,7 @@
 
 package io.vertx.sqlclient.internal;
 
+import io.vertx.core.Completable;
 import io.vertx.core.Promise;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.spi.metrics.ClientMetrics;
@@ -60,7 +61,7 @@ default String system() {
 
   DatabaseMetadata getDatabaseMetaData();
 
-  void close(Holder holder, Promise<Void> promise);
+  void close(Holder holder, Completable<Void> promise);
 
   int getProcessId();
 

vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/SqlConnectionBase.java

@@ -199,15 +199,24 @@ public Future<Void> close() {
   }
 
   @Override
-  public void close(Promise<Void> completion) {
+  public void close(Completable<Void> completion) {
+    if (closeFactoryAfterUsage) {
+      Completable<Void> next = completion;
+      completion = (res, err) -> {
+        try {
+          next.complete(res, err);
+        } finally {
+          factory.close((res2, err2) -> {});
+        }
+      };
+    }
     doClose(completion);
     if (closeFactoryAfterUsage) {
-      completion.future().onComplete(v -> factory.close(Promise.promise()));
       context.removeCloseHook(this);
     }
   }
 
-  private void doClose(Promise<Void> promise) {
+  private void doClose(Completable<Void> promise) {
     context.execute(promise, p -> {
       if (tx != null) {
         tx.rollback(ar -> conn.close(this, p));

vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/pool/PoolImpl.java

@@ -179,7 +179,7 @@ private void acquire(ContextInternal context, long timeout, Completable<SqlConne
   }
 
   @Override
-  public void close(Promise<Void> completion) {
+  public void close(Completable<Void> completion) {
     doClose().onComplete(completion);
   }