Revert "Remove callbase usage in implementation"

This reverts commit 53b09ae.

Author: vietj

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLSocketConnection.java

@@ -19,7 +19,6 @@
 
 import io.netty.channel.ChannelPipeline;
 import io.vertx.core.AsyncResult;
-import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
 import io.vertx.core.buffer.Buffer;
@@ -100,8 +99,8 @@ protected <R> void doSchedule(CommandBase<R> cmd, Handler<AsyncResult<R>> handle
     }
   }
 
-  public Future<Void> upgradeToSsl() {
-    return socket.upgradeToSsl();
+  public void upgradeToSsl(Handler<AsyncResult<Void>> completionHandler) {
+    socket.upgradeToSsl(completionHandler);
   }
 
   @Override

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/codec/InitialHandshakeCommandCodec.java

@@ -143,7 +143,7 @@ private void handleInitialHandshake(ByteBuf payload) {
       encoder.clientCapabilitiesFlag |= CLIENT_SSL;
       sendSslRequest();
 
-      encoder.socketConnection.upgradeToSsl().onComplete(upgrade -> {
+      encoder.socketConnection.upgradeToSsl(upgrade -> {
         if (upgrade.succeeded()) {
           doSendHandshakeResponseMessage(serverAuthPluginName, cmd.authenticationPlugin(), authPluginData, serverCapabilitiesFlags);
         } else {

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

@@ -59,17 +59,18 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
     byteBuf.release();
     switch (b) {
       case PgProtocolConstants.MESSAGE_TYPE_SSL_YES: {
-        conn
-          .socket()
-          .upgradeToSsl()
-          .onComplete(ar -> {
-            if (ar.succeeded()) {
-              ctx.pipeline().remove(this);
-              upgradePromise.complete();
-            } else {
+        Handler handler = o -> {
+          if (o instanceof AsyncResult) {
+            AsyncResult res = (AsyncResult) o;
+            if (res.failed()) {
               // Connection close will fail the promise
+              return;
             }
-        });
+          }
+          ctx.pipeline().remove(this);
+          upgradePromise.complete();
+        };
+        conn.socket().upgradeToSsl(handler);
         break;
       }
       case PgProtocolConstants.MESSAGE_TYPE_SSL_NO: {

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

@@ -85,7 +85,7 @@ void sendCancelRequestMessage(int processId, int secretKey, Handler<AsyncResult<
     buffer.appendInt(processId);
     buffer.appendInt(secretKey);
 
-    socket.write(buffer).onComplete(ar -> {
+    socket.write(buffer, ar -> {
       if (ar.succeeded()) {
         // directly close this connection
         if (status == Status.CONNECTED) {

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

@@ -168,7 +168,7 @@ private void tryConnect(long delayMillis, Handler<AsyncResult<Void>> handler) {
   }
 
   private void doConnect(Handler<AsyncResult<Void>> completionHandler) {
-    PgConnection.connect(vertx, options).onComplete(ar -> handleConnectResult(completionHandler, ar));
+    PgConnection.connect(vertx, options, ar -> handleConnectResult(completionHandler, ar));
   }
 
   private synchronized void handleConnectResult(Handler<AsyncResult<Void>> completionHandler, AsyncResult<PgConnection> ar1) {
@@ -191,7 +191,7 @@ private synchronized void handleConnectResult(Handler<AsyncResult<Void>> complet
             return channel.quotedName;
           })
           .collect(Collectors.joining(";LISTEN ", "LISTEN ", ""));
-        conn.query(sql).execute().onComplete(ar2 -> {
+        conn.query(sql).execute(ar2 -> {
           if (ar2.failed()) {
             log.error("Cannot LISTEN to channels", ar2.cause());
             conn.close();
@@ -224,7 +224,7 @@ void add(ChannelImpl sub) {
         if (conn != null) {
           subscribed = true;
           String sql = "LISTEN " + quotedName;
-          conn.query(sql).execute().onComplete(ar -> {
+          conn.query(sql).execute(ar -> {
             if (ar.succeeded()) {
               Handler<Void> handler = sub.subscribeHandler;
               if (handler != null) {
@@ -243,7 +243,7 @@ void remove(ChannelImpl sub) {
       if (subs.isEmpty()) {
         channels.remove(name, this);
         if (conn != null) {
-          conn.query("UNLISTEN " + quotedName).execute().onComplete(ar -> {
+          conn.query("UNLISTEN " + quotedName).execute(ar -> {
             if (ar.failed()) {
               log.error("Cannot UNLISTEN channel " + name, ar.cause());
             }

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

@@ -70,8 +70,7 @@ public void execute(I parameters, Handler<AsyncResult<R>> handler) {
 
     queryMapper
       .apply(client.preparedQuery(sqlTemplate.getSql()))
-      .execute(tupleMapper.apply(parameters))
-      .onComplete(handler);
+      .execute(tupleMapper.apply(parameters), handler);
   }
 
   @Override
@@ -87,8 +86,7 @@ public void executeBatch(List<I> batch, Handler<AsyncResult<R>> handler) {
       .executeBatch(batch
         .stream()
         .map(tupleMapper)
-        .collect(Collectors.toList()))
-      .onComplete(handler);
+        .collect(Collectors.toList()), handler);
   }
 
   @Override

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

@@ -175,7 +175,7 @@ public Future<SqlConnection> getConnection() {
                   .compose(v -> context.failedFuture(err), failure -> context.failedFuture(err));
               }
             }))
-        .onComplete(ar -> conn.close().onComplete(v -> context.removeLocal(PROPAGATABLE_CONNECTION))));
+        .onComplete(ar -> conn.close(v -> context.removeLocal(PROPAGATABLE_CONNECTION))));
   }
 
   @Override

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

@@ -90,7 +90,7 @@ public RowStream<Row> handler(Handler<Row> handler) {
         return this;
       }
     }
-    c.read(fetch).onComplete(this);
+    c.read(fetch, this);
     return this;
   }
 
@@ -206,7 +206,7 @@ private void checkPending() {
               break;
             } else if (cursor.hasMore()) {
               readInProgress = true;
-              cursor.read(fetch).onComplete(this);
+              cursor.read(fetch, this);
               break;
             } else {
               cursor.close();