CommandMessage are named XXXCodec due to the original design

We should name those classes XXXCommandMessage instead to implement a more idiomatic design.

This rename the hierarchies for the implemented codecs.

In addition we try to avoid command message completion handler leaking into each codec and instead we keep this confined in the database socket connection.

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.impl.codec.CommandCodec;
+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.ExtendedBatchQueryCommandCodec;
-import io.vertx.db2client.impl.codec.ExtendedQueryCommandCodec;
+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,
@@ -90,15 +108,15 @@ public void init() {
   @Override
   protected CommandMessage<?, ?> toMessage(ExtendedQueryCommand<?> command, PreparedStatement preparedStatement) {
     if (command.isBatch()) {
-      return new ExtendedBatchQueryCommandCodec<>(command, (DB2PreparedStatement) preparedStatement);
+      return new ExtendedBatchQueryDB2CommandMessage<>(command, (DB2PreparedStatement) preparedStatement);
     } else {
-      return new ExtendedQueryCommandCodec(command, (DB2PreparedStatement) preparedStatement);
+      return new ExtendedQueryDB2CommandMessage(command, (DB2PreparedStatement) preparedStatement);
     }
   }
 
   @Override
   protected CommandMessage<?, ?> toMessage(CommandBase<?> command) {
-    return CommandCodec.wrap(command);
+    return DB2CommandMessage.wrap(command);
   }
 
   @Override

vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/AuthenticationCommandBaseCodec.java renamed to vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/AuthenticationDB2CommandBaseMessage.java

@@ -17,9 +17,9 @@
 
 import io.vertx.db2client.impl.command.AuthenticationCommandBase;
 
-abstract class AuthenticationCommandBaseCodec<R, C extends AuthenticationCommandBase<R>> extends CommandCodec<R, C> {
+abstract class AuthenticationDB2CommandBaseMessage<R, C extends AuthenticationCommandBase<R>> extends DB2CommandMessage<R, C> {
 
-  AuthenticationCommandBaseCodec(C cmd) {
+  AuthenticationDB2CommandBaseMessage(C cmd) {
     super(cmd);
   }
 

vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/CloseConnectionCommandCodec.java renamed to vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/CloseConnectionDB2CommandMessage.java

@@ -20,9 +20,9 @@
 import io.vertx.db2client.impl.drda.DRDAQueryResponse;
 import io.vertx.sqlclient.spi.protocol.CloseConnectionCommand;
 
-class CloseConnectionCommandCodec extends CommandCodec<Void, CloseConnectionCommand> {
+class CloseConnectionDB2CommandMessage extends DB2CommandMessage<Void, CloseConnectionCommand> {
 
-  CloseConnectionCommandCodec(CloseConnectionCommand cmd) {
+  CloseConnectionDB2CommandMessage(CloseConnectionCommand cmd) {
     super(cmd);
   }
 

vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/CloseCursorCommandCodec.java renamed to vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/CloseCursorDB2CommandMessage.java

@@ -24,11 +24,11 @@
 import io.vertx.sqlclient.spi.protocol.CloseCursorCommand;
 import io.vertx.sqlclient.codec.CommandResponse;
 
-class CloseCursorCommandCodec extends CommandCodec<Void, CloseCursorCommand> {
+class CloseCursorDB2CommandMessage extends DB2CommandMessage<Void, CloseCursorCommand> {
 
-  private static final Logger LOG = LoggerFactory.getLogger(CloseCursorCommandCodec.class);
+  private static final Logger LOG = LoggerFactory.getLogger(CloseCursorDB2CommandMessage.class);
 
-  CloseCursorCommandCodec(CloseCursorCommand cmd) {
+  CloseCursorDB2CommandMessage(CloseCursorCommand cmd) {
     super(cmd);
   }
 
@@ -52,6 +52,6 @@ void encode(DB2Encoder encoder) {
   void decodePayload(ByteBuf payload, int payloadLength) {
     DRDAQueryResponse closeCursor = new DRDAQueryResponse(payload, encoder.socketConnection.connMetadata);
     closeCursor.readCursorClose();
-    completionHandler.handle(CommandResponse.success(null));
+    fireCommandSuccess(null);
   }
 }

vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/CloseStatementCommandCodec.java renamed to vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/CloseStatementDB2CommandMessage.java

@@ -19,17 +19,18 @@
 import io.vertx.sqlclient.spi.protocol.CloseStatementCommand;
 import io.vertx.sqlclient.codec.CommandResponse;
 
-class CloseStatementCommandCodec extends CommandCodec<Void, CloseStatementCommand> {
+class CloseStatementDB2CommandMessage extends DB2CommandMessage<Void, CloseStatementCommand> {
 
-  CloseStatementCommandCodec(CloseStatementCommand cmd) {
+  CloseStatementDB2CommandMessage(CloseStatementCommand cmd) {
     super(cmd);
   }
 
   @Override
   void encode(DB2Encoder encoder) {
+    super.encode(encoder);
     DB2PreparedStatement statement = (DB2PreparedStatement) cmd.statement();
     statement.close();
-    completionHandler.handle(CommandResponse.success(null));
+    fireCommandSuccess(null);
   }
 
   @Override

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;
 
@@ -27,23 +25,11 @@ public class DB2Codec extends CombinedChannelDuplexHandler<DB2Decoder, DB2Encode
   // TODO @AGG check what packet length limit actually is for DB2
   static final int PACKET_PAYLOAD_LENGTH_LIMIT = 0xFFFFFF;
 
-  private final ArrayDeque<CommandCodec<?, ?>> inflight = new ArrayDeque<>();
+  private final ArrayDeque<DB2CommandMessage<?, ?>> inflight = new ArrayDeque<>();
 
   public DB2Codec(DB2SocketConnection db2SocketConnection) {
     DB2Encoder encoder = new DB2Encoder(inflight, 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 (CommandCodec<?, ?> commandCodec : inflight) {
-      commandCodec.fail(failure);
-    }
-  }
 }

vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/CommandCodec.java renamed to vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/DB2CommandMessage.java

@@ -28,34 +28,41 @@
 import io.vertx.sqlclient.spi.protocol.PrepareStatementCommand;
 import io.vertx.sqlclient.spi.protocol.SimpleQueryCommand;
 
-public abstract class CommandCodec<R, C extends CommandBase<R>> extends CommandMessage<R, C> {
+public abstract class DB2CommandMessage<R, C extends CommandBase<R>> extends CommandMessage<R, C> {
 
-  Handler<? super CommandResponse<R>> completionHandler;
   public Throwable failure;
   public R result;
   DB2Encoder encoder;
 
-  CommandCodec(C cmd) {
+  DB2CommandMessage(C cmd) {
     super(cmd);
   }
 
+  void fireCommandFailure(Throwable err) {
+    encoder.fireCommandFailure(this, err);
+  }
+
+  void fireCommandSuccess(R result) {
+    encoder.fireCommandSuccess(this, result);
+  }
+
   @SuppressWarnings({ "rawtypes", "unchecked" })
-  public static CommandCodec<?, ?> wrap(CommandBase<?> cmd) {
-    CommandCodec<?, ?> codec = null;
+  public static DB2CommandMessage<?, ?> wrap(CommandBase<?> cmd) {
+    DB2CommandMessage<?, ?> codec = null;
     if (cmd instanceof InitialHandshakeCommand) {
-      codec = new InitialHandshakeCommandCodec((InitialHandshakeCommand) cmd);
+      codec = new InitialHandshakeDB2CommandMessage((InitialHandshakeCommand) cmd);
     } else if (cmd instanceof SimpleQueryCommand) {
-      codec = new SimpleQueryCommandCodec((SimpleQueryCommand) cmd);
+      codec = new SimpleQueryDB2CommandMessage((SimpleQueryCommand) cmd);
     } else if (cmd instanceof CloseConnectionCommand) {
-      codec = new CloseConnectionCommandCodec((CloseConnectionCommand) cmd);
+      codec = new CloseConnectionDB2CommandMessage((CloseConnectionCommand) cmd);
     } else if (cmd instanceof PrepareStatementCommand) {
-      codec = new PrepareStatementCodec((PrepareStatementCommand) cmd);
+      codec = new PrepareStatementDB2CommandMessage((PrepareStatementCommand) cmd);
     } else if (cmd instanceof CloseStatementCommand) {
-      codec = new CloseStatementCommandCodec((CloseStatementCommand) cmd);
+      codec = new CloseStatementDB2CommandMessage((CloseStatementCommand) cmd);
     } else if (cmd instanceof CloseCursorCommand) {
-      codec = new CloseCursorCommandCodec((CloseCursorCommand) cmd);
+      codec = new CloseCursorDB2CommandMessage((CloseCursorCommand) cmd);
     } else if (cmd instanceof PingCommand) {
-      codec = new PingCommandCodec((PingCommand) cmd);
+      codec = new PingDB2CommandMessage((PingCommand) cmd);
 //        } else if (cmd instanceof InitDbCommand) {
 //            codec = new InitDbCommandCodec((InitDbCommand) cmd);
       // } else if (cmd instanceof StatisticsCommand) {

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

@@ -31,9 +31,9 @@ class DB2Decoder extends ByteToMessageDecoder {
 
   private static final Logger LOG = LoggerFactory.getLogger(DB2Decoder.class);
 
-  private final ArrayDeque<CommandCodec<?, ?>> inflight;
+  private final ArrayDeque<DB2CommandMessage<?, ?>> inflight;
 
-  DB2Decoder(ArrayDeque<CommandCodec<?, ?>> inflight) {
+  DB2Decoder(ArrayDeque<DB2CommandMessage<?, ?>> inflight) {
     this.inflight = inflight;
   }
 
@@ -71,7 +71,7 @@ private int computeLength(ByteBuf in) {
   }
 
   private void decodePayload(ByteBuf payload, int payloadLength) {
-    CommandCodec<?, ?> ctx = inflight.peek();
+    DB2CommandMessage<?, ?> ctx = inflight.peek();
     int startIndex = payload.readerIndex();
     try {
       if (LOG.isDebugEnabled())
@@ -93,7 +93,7 @@ private void decodePayload(ByteBuf payload, int payloadLength) {
           LOG.error("Command failed with: " + t.getMessage() + "\nCommand=" + ctx, t);
         }
       }
-      ctx.completionHandler.handle(CommandResponse.failure(t));
+      ctx.fireCommandFailure(t);
     } finally {
       payload.clear();
       payload.release();

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

@@ -30,12 +30,12 @@ class DB2Encoder extends ChannelOutboundHandlerAdapter {
 
   public static final Logger LOG = LoggerFactory.getLogger(DB2Encoder.class);
 
-  private final ArrayDeque<CommandCodec<?, ?>> inflight;
+  private final ArrayDeque<DB2CommandMessage<?, ?>> inflight;
   ChannelHandlerContext chctx;
 
   final DB2SocketConnection socketConnection;
 
-  DB2Encoder(ArrayDeque<CommandCodec<?, ?>> inflight, DB2SocketConnection db2SocketConnection) {
+  DB2Encoder(ArrayDeque<DB2CommandMessage<?, ?>> inflight, DB2SocketConnection db2SocketConnection) {
     this.inflight = inflight;
     this.socketConnection = db2SocketConnection;
   }
@@ -47,27 +47,41 @@ public void handlerAdded(ChannelHandlerContext ctx) {
 
   @Override
   public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
-    if (msg instanceof CommandCodec<?, ?>) {
-      CommandCodec<?, ?> cmd = (CommandCodec<?, ?>) msg;
+    if (msg instanceof DB2CommandMessage<?, ?>) {
+      DB2CommandMessage<?, ?> cmd = (DB2CommandMessage<?, ?>) msg;
       write(cmd);
     } else {
       super.write(ctx, msg, promise);
     }
   }
 
+  void fireCommandFailure(DB2CommandMessage msg, Throwable err) {
+    DB2CommandMessage<?, ?> c = inflight.poll();
+    if (c == msg) {
+      chctx.fireChannelRead(CommandResponse.failure(err));
+    } else {
+      throw new IllegalStateException();
+    }
+  }
+
+  <R> void fireCommandSuccess(DB2CommandMessage msg, R result) {
+    DB2CommandMessage<?, ?> c = inflight.poll();
+    if (c == msg) {
+      chctx.fireChannelRead(CommandResponse.success(result));
+    } else {
+      throw new IllegalStateException();
+    }
+  }
+
   @SuppressWarnings({ "unchecked", "rawtypes" })
-  void write(CommandCodec<?, ?> msg) {
-    msg.completionHandler = resp -> {
-      CommandCodec<?, ?> c = inflight.poll();
-      resp.handler = (Completable) c.handler;
-      chctx.fireChannelRead(resp);
-    };
-    inflight.add(msg);
+  void write(DB2CommandMessage<?, ?> msg) {
     try {
+      inflight.add(msg);
       msg.encode(this);
     } catch (Throwable e) {
+      inflight.pollLast();
       LOG.error("FATAL: Unable to encode command: " + msg, e);
-      msg.completionHandler.handle(CommandResponse.failure(e));
+      chctx.fireChannelRead(CommandResponse.failure(e));
     }
   }
 }