Remove more usage of completion handler

Author: vietj

vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2SocketConnection.java

@@ -23,13 +23,17 @@
 import io.vertx.core.internal.net.NetSocketInternal;
 import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.db2client.DB2ConnectOptions;
+import io.vertx.db2client.DB2Exception;
+import io.vertx.db2client.impl.codec.ConnectionState;
 import io.vertx.db2client.impl.codec.DB2CommandMessage;
 import io.vertx.db2client.impl.codec.DB2Codec;
 import io.vertx.db2client.impl.codec.DB2PreparedStatement;
 import io.vertx.db2client.impl.codec.ExtendedBatchQueryDB2CommandMessage;
 import io.vertx.db2client.impl.codec.ExtendedQueryDB2CommandMessage;
 import io.vertx.db2client.impl.command.InitialHandshakeCommand;
 import io.vertx.db2client.impl.drda.ConnectionMetaData;
+import io.vertx.db2client.impl.drda.SQLState;
+import io.vertx.db2client.impl.drda.SqlCode;
 import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.codec.CommandMessage;
 import io.vertx.sqlclient.spi.connection.Connection;
@@ -51,6 +55,7 @@ public class DB2SocketConnection extends SocketConnectionBase {
   private DB2Codec codec;
   private Handler<Void> closeHandler;
   public final ConnectionMetaData connMetadata = new ConnectionMetaData();
+  public ConnectionState status = ConnectionState.CONNECTING;
 
   public DB2SocketConnection(NetSocketInternal socket,
       ClientMetrics clientMetrics,
@@ -64,6 +69,19 @@ public DB2SocketConnection(NetSocketInternal socket,
     this.connectOptions = connectOptions;
   }
 
+  @Override
+  protected <R> void fail(CommandBase<R> command, Completable<R> handler, Throwable err) {
+    if (status == ConnectionState.CONNECTING && command instanceof InitialHandshakeCommand) {
+      // Sometimes DB2 closes the connection when sending an invalid Database name.
+      // -4499 = A fatal error occurred that resulted in a disconnect from the data
+      // source.
+      // 08001 = "The connection was unable to be established"
+      err = new DB2Exception("The connection was closed by the database server.", SqlCode.CONNECTION_REFUSED,
+        SQLState.AUTH_DATABASE_CONNECTION_REFUSED);
+    }
+    super.fail(command, handler, err);
+  }
+
   // TODO RETURN FUTURE ???
   void sendStartupMessage(String username,
       String password,

vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/ConnectionState.java

@@ -0,0 +1,5 @@
+package io.vertx.db2client.impl.codec;
+
+public enum ConnectionState {
+  CONNECTING, AUTHENTICATING, CONNECTED, CONNECT_FAILED
+}

vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/DB2Codec.java

@@ -15,10 +15,8 @@
  */
 package io.vertx.db2client.impl.codec;
 
-import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.CombinedChannelDuplexHandler;
 import io.vertx.db2client.impl.DB2SocketConnection;
-import io.vertx.sqlclient.ClosedConnectionException;
 
 import java.util.ArrayDeque;
 
@@ -34,16 +32,4 @@ public DB2Codec(DB2SocketConnection db2SocketConnection) {
     DB2Decoder decoder = new DB2Decoder(inflight);
     init(decoder, encoder);
   }
-
-  @Override
-  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-    clearInflightCommands(ClosedConnectionException.INSTANCE);
-    super.channelInactive(ctx);
-  }
-
-  private void clearInflightCommands(Throwable failure) {
-    for (DB2CommandMessage<?, ?> commandMsg : inflight) {
-      commandMsg.fail(failure);
-    }
-  }
 }

vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/InitialHandshakeDB2CommandMessage.java

@@ -26,7 +26,6 @@
 import io.vertx.db2client.impl.drda.SQLState;
 import io.vertx.db2client.impl.drda.SqlCode;
 import io.vertx.sqlclient.spi.connection.Connection;
-import io.vertx.sqlclient.codec.CommandResponse;
 
 /**
  * InitialHandshakeCommandCodec encodes the packets to get a connection from the database.
@@ -37,14 +36,8 @@
  */
 class InitialHandshakeDB2CommandMessage extends AuthenticationDB2CommandBaseMessage<Connection, InitialHandshakeCommand> {
 
-  private static enum ConnectionState {
-    CONNECTING, AUTHENTICATING, CONNECTED, CONNECT_FAILED
-  }
-
   private static final int TARGET_SECURITY_MEASURE = DRDAConstants.SECMEC_USRIDPWD;
 
-  private ConnectionState status = ConnectionState.CONNECTING;
-
   InitialHandshakeDB2CommandMessage(InitialHandshakeCommand cmd) {
     super(cmd);
   }
@@ -53,16 +46,6 @@ private static enum ConnectionState {
   void encode(DB2Encoder encoder) {
     super.encode(encoder);
     encoder.socketConnection.connMetadata.databaseName = cmd.database();
-    encoder.socketConnection.closeHandler(h -> {
-      if (status == ConnectionState.CONNECTING) {
-        // Sometimes DB2 closes the connection when sending an invalid Database name.
-        // -4499 = A fatal error occurred that resulted in a disconnect from the data
-        // source.
-        // 08001 = "The connection was unable to be established"
-        fail(new DB2Exception("The connection was closed by the database server.", SqlCode.CONNECTION_REFUSED,
-            SQLState.AUTH_DATABASE_CONNECTION_REFUSED));
-      }
-    });
 
     ByteBuf packet = allocateBuffer();
     int packetStartIdx = packet.writerIndex();
@@ -90,7 +73,7 @@ void encode(DB2Encoder encoder) {
   @Override
   void decodePayload(ByteBuf payload, int payloadLength) {
     DRDAConnectResponse response = new DRDAConnectResponse(payload, encoder.socketConnection.connMetadata);
-    switch (status) {
+    switch (encoder.socketConnection.status) {
       case CONNECTING:
         response.readExchangeServerAttributes();
         // readAccessSecurity can throw a DB2Exception if there are problems connecting.
@@ -100,7 +83,7 @@ void decodePayload(ByteBuf payload, int payloadLength) {
         try {
           response.readAccessSecurity(TARGET_SECURITY_MEASURE);
         } catch (DB2Exception de) {
-          status = ConnectionState.CONNECT_FAILED;
+          encoder.socketConnection.status = ConnectionState.CONNECT_FAILED;
           throw de;
         }
 
@@ -116,7 +99,7 @@ void decodePayload(ByteBuf payload, int payloadLength) {
 
         int lenOfPayload = packet.writerIndex() - packetStartIdx;
         sendPacket(packet, lenOfPayload);
-        status = ConnectionState.AUTHENTICATING;
+        encoder.socketConnection.status = ConnectionState.AUTHENTICATING;
         break;
 
       case AUTHENTICATING:
@@ -125,7 +108,7 @@ void decodePayload(ByteBuf payload, int payloadLength) {
         if (accData.correlationToken != null) {
           encoder.socketConnection.connMetadata.correlationToken = accData.correlationToken;
         }
-        status = ConnectionState.CONNECTED;
+        encoder.socketConnection.status = ConnectionState.CONNECTED;
         fireCommandSuccess(cmd.connection());
         break;
 

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

@@ -436,22 +436,29 @@ protected void handleClose(Throwable t) {
       }
       CommandMessage<?, ?> msg;
       while ((msg = bilto.poll()) != null) {
-        Completable<?> handler = msg.handler;
-        context.runOnContext(v -> handler.fail(ClosedConnectionException.INSTANCE));
+        fail(msg, ClosedConnectionException.INSTANCE);
       }
       Throwable cause = t == null ? VertxException.noStackTrace(PENDING_CMD_CONNECTION_CORRUPT_MSG) : new VertxException(PENDING_CMD_CONNECTION_CORRUPT_MSG, t);
       CommandBase<?> cmd;
       while ((cmd = pending.poll()) != null) {
-        CommandBase<?> c = cmd;
-        Completable<?> handler = handlers.poll();
-        context.runOnContext(v -> handler.fail(cause));
+        CommandBase c = cmd;
+        Completable handler = handlers.poll();
+        fail(c, handler, cause);
       }
       if (holder != null) {
         holder.handleClosed();
       }
     }
   }
 
+  private <R> void fail(CommandMessage<R, ?> msg, Throwable err) {
+    fail(msg.cmd, msg.handler, err);
+  }
+
+  protected <R> void fail(CommandBase<R> command, Completable<R> handler, Throwable err) {
+    context.runOnContext(v -> handler.fail(err));
+  }
+
   public boolean pipeliningEnabled() {
     return pipeliningLimit > 1;
   }