Let the socket connection take care of the lifecycle of command message and the associated flow.

Motivation:

Currently each codec takes care of failing the inflight command messages when a failure occurs in the pipeline. The socket connection already takes care of translating command responses to completion of the command handler.

We should let the socket connection take care of handling the case of failure, this avoids propagating the command response handler to each codec and the codec only cares about writing an inbound command response message.

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/CloseCursorDB2CommandMessage.java

@@ -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/CloseStatementDB2CommandMessage.java

@@ -27,9 +27,10 @@ class CloseStatementDB2CommandMessage extends DB2CommandMessage<Void, CloseState
 
   @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;
 
@@ -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/DB2CommandMessage.java

@@ -30,7 +30,6 @@
 
 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;
@@ -39,6 +38,14 @@ public abstract class DB2CommandMessage<R, C extends CommandBase<R>> extends Com
     super(cmd);
   }
 
+  void fireCommandFailure(Throwable err) {
+    encoder.fireCommandFailure(this, err);
+  }
+
+  void fireCommandSuccess(R result) {
+    encoder.fireCommandSuccess(this, result);
+  }
+
   @SuppressWarnings({ "rawtypes", "unchecked" })
   public static DB2CommandMessage<?, ?> wrap(CommandBase<?> cmd) {
     DB2CommandMessage<?, ?> codec = null;

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

@@ -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

@@ -55,19 +55,33 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise 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(DB2CommandMessage<?, ?> msg) {
-    msg.completionHandler = resp -> {
-      DB2CommandMessage<?, ?> c = inflight.poll();
-      resp.handler = (Completable) c.handler;
-      chctx.fireChannelRead(resp);
-    };
-    inflight.add(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));
     }
   }
 }

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

@@ -21,6 +21,7 @@
 import java.util.stream.Collectors;
 
 import io.netty.buffer.ByteBuf;
+import io.vertx.core.VertxException;
 import io.vertx.core.internal.logging.Logger;
 import io.vertx.core.internal.logging.LoggerFactory;
 import io.vertx.db2client.impl.codec.DB2PreparedStatement.QueryInstance;
@@ -49,7 +50,7 @@ public ExtendedBatchQueryDB2CommandMessage(ExtendedQueryCommand<R> cmd, DB2Prepa
   @Override
   void encode(DB2Encoder encoder) {
     if (params.isEmpty()) {
-      completionHandler.handle(CommandResponse.failure("Can not execute batch query with 0 sets of batch parameters."));
+      fireCommandFailure(VertxException.noStackTrace("Can not execute batch query with 0 sets of batch parameters."));
       return;
     }
 
@@ -88,7 +89,7 @@ void decodeQuery(ByteBuf payload) {
       hasMoreResults &= !queryComplete;
       handleQueryResult(decoder);
     }
-    completionHandler.handle(CommandResponse.success(hasMoreResults));
+    fireCommandSuccess(hasMoreResults);
   }
 
   void decodeUpdate(ByteBuf payload) {
@@ -99,7 +100,7 @@ void decodeUpdate(ByteBuf payload) {
     if (cmd.autoCommit()) {
       updateResponse.readLocalCommit();
     }
-    completionHandler.handle(CommandResponse.success(true));
+    fireCommandSuccess(true);
   }
 
   @Override

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

@@ -19,6 +19,7 @@
 import java.util.stream.Collector;
 
 import io.netty.buffer.ByteBuf;
+import io.vertx.core.VertxException;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.internal.buffer.BufferInternal;
 import io.vertx.db2client.impl.codec.DB2PreparedStatement.QueryInstance;
@@ -44,8 +45,8 @@ abstract class ExtendedQueryDB2CommandBaseMessage<R, C extends ExtendedQueryComm
   void encodePreparedQuery(DRDAQueryRequest queryRequest, QueryInstance queryInstance, Tuple params) {
     int requiredParams = statement.paramDesc.paramDefinitions().columns_;
     if (params.size() != requiredParams) {
-      completionHandler.handle(CommandResponse.failure("Only " + params.size()
-          + " prepared statement parameters were provided " + "but " + requiredParams + " parameters are required."));
+      fireCommandFailure(VertxException.noStackTrace("Only " + params.size()
+        + " prepared statement parameters were provided " + "but " + requiredParams + " parameters are required."));
       return;
     }