The internal command codec for PostgreSQL/MSSQL/MySQL should act differently after it has been signaled a failure. When the codec is signaled a failure, it should record it, then any processed command should be failed an not written in the channel pipeline.

The codec code has also been improved with more encapsulation of the infight command queue in the codec class, simplified and made more homogeneous accross implementations.

Author: vietj

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/codec/CloseCursorCommandCodec.java

@@ -33,7 +33,7 @@ void encode() {
     if (cursorData != null && cursorData.serverCursorId > 0) {
       sendCursorClose();
     } else {
-      completionHandler.handle(CommandResponse.success(null));
+      tdsMessageCodec.decoder().fireCommandResponse(CommandResponse.success(null));
     }
   }
 

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/codec/CloseStatementCommandCodec.java

@@ -30,7 +30,7 @@ void encode() {
     if (ps.handle > 0) {
       sendUnprepareRequest();
     } else {
-      completionHandler.handle(CommandResponse.success(null));
+      tdsMessageCodec.decoder().fireCommandResponse(CommandResponse.success(null));
     }
   }
 

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/codec/ExtendedBatchQueryCommandCodec.java

@@ -33,7 +33,7 @@ class ExtendedBatchQueryCommandCodec<T> extends ExtendedQueryCommandBaseCodec<T>
   @Override
   void encode() {
     if (paramsList.isEmpty()) {
-      completionHandler.handle(CommandResponse.failure("Can not execute batch query with 0 sets of batch parameters."));
+      tdsMessageCodec.decoder().fireCommandResponse(CommandResponse.failure("Can not execute batch query with 0 sets of batch parameters."));
       return;
     }
     super.encode();

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/codec/MSSQLCommandCodec.java

@@ -12,7 +12,6 @@
 package io.vertx.mssqlclient.impl.codec;
 
 import io.netty.buffer.ByteBuf;
-import io.vertx.core.Handler;
 import io.vertx.mssqlclient.MSSQLException;
 import io.vertx.mssqlclient.MSSQLInfo;
 import io.vertx.sqlclient.impl.command.CommandBase;
@@ -30,7 +29,6 @@ abstract class MSSQLCommandCodec<R, C extends CommandBase<R>> {
   final C cmd;
   public MSSQLException failure;
   public R result;
-  Handler<? super CommandResponse<R>> completionHandler;
 
   MSSQLCommandCodec(TdsMessageCodec tdsMessageCodec, C cmd) {
     this.tdsMessageCodec = tdsMessageCodec;
@@ -221,6 +219,6 @@ void complete() {
     } else {
       resp = CommandResponse.success(result);
     }
-    completionHandler.handle(resp);
+    tdsMessageCodec.decoder().fireCommandResponse(resp);
   }
 }

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/codec/PreLoginCommandCodec.java

@@ -89,6 +89,6 @@ void decode(ByteBuf payload) {
       }
       payload.resetReaderIndex();
     }
-    completionHandler.handle(CommandResponse.success(new PreLoginResponse(metadata, encryptionLevel)));
+    tdsMessageCodec.decoder().fireCommandResponse(CommandResponse.success(new PreLoginResponse(metadata, encryptionLevel)));
   }
 }

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/codec/PrepareStatementCodec.java

@@ -24,6 +24,6 @@ class PrepareStatementCodec extends MSSQLCommandCodec<PreparedStatement, Prepare
   void encode() {
     // we use sp_prepexec instead of sp_prepare + sp_exec
     PreparedStatement preparedStatement = new MSSQLPreparedStatement(cmd.sql());
-    completionHandler.handle(CommandResponse.success(preparedStatement));
+    tdsMessageCodec.decoder().fireCommandResponse(CommandResponse.success(preparedStatement));
   }
 }

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/codec/TdsMessageCodec.java

@@ -27,14 +27,16 @@ public class TdsMessageCodec extends CombinedChannelDuplexHandler<TdsMessageDeco
 
   private final ArrayDeque<MSSQLCommandCodec<?, ?>> inflight = new ArrayDeque<>();
   private final TdsMessageEncoder encoder;
+  private final TdsMessageDecoder decoder;
 
   private ChannelHandlerContext chctx;
   private ByteBufAllocator alloc;
   private long transactionDescriptor;
   private Map<String, CursorData> cursorDataMap;
+  private Throwable failure;
 
   public TdsMessageCodec(int packetSize) {
-    TdsMessageDecoder decoder = new TdsMessageDecoder(this);
+    decoder = new TdsMessageDecoder(this);
     encoder = new TdsMessageEncoder(this, packetSize);
     init(decoder, encoder);
   }
@@ -48,30 +50,41 @@ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
 
   @Override
   public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-    fail(ctx, cause);
+    fail(cause);
     super.exceptionCaught(ctx, cause);
   }
 
-  private void fail(ChannelHandlerContext ctx, Throwable cause) {
-    for (Iterator<MSSQLCommandCodec<?, ?>> it = inflight.iterator(); it.hasNext(); ) {
-      MSSQLCommandCodec<?, ?> codec = it.next();
-      it.remove();
-      CommandResponse<Object> failure = CommandResponse.failure(cause);
-      failure.cmd = (CommandBase) codec.cmd;
-      ctx.fireChannelRead(failure);
+  private void fail(Throwable cause) {
+    if (failure == null) {
+      failure = cause;
+      for (Iterator<MSSQLCommandCodec<?, ?>> it = inflight.iterator(); it.hasNext(); ) {
+        MSSQLCommandCodec<?, ?> codec = it.next();
+        it.remove();
+        fail(codec, cause);
+      }
     }
   }
 
+  private void fail(MSSQLCommandCodec<?, ?> codec, Throwable cause) {
+    CommandResponse<Object> failure = CommandResponse.failure(cause);
+    failure.cmd = (CommandBase) codec.cmd;
+    chctx.fireChannelRead(failure);
+  }
+
   @Override
   public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-    fail(ctx, ClosedConnectionException.INSTANCE);
+    fail(ClosedConnectionException.INSTANCE);
     super.channelInactive(ctx);
   }
 
   TdsMessageEncoder encoder() {
     return encoder;
   }
 
+  TdsMessageDecoder decoder() {
+    return decoder;
+  }
+
   ChannelHandlerContext chctx() {
     return chctx;
   }
@@ -96,8 +109,14 @@ void setTransactionDescriptor(long transactionDescriptor) {
     return inflight.poll();
   }
 
-  void add(MSSQLCommandCodec<?, ?> codec) {
-    inflight.add(codec);
+  boolean add(MSSQLCommandCodec<?, ?> codec) {
+    if (failure == null) {
+      inflight.add(codec);
+      return true;
+    } else {
+      fail(codec, failure);
+      return false;
+    }
   }
 
   CursorData getOrCreateCursorData(String cursorId) {

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/codec/TdsMessageDecoder.java

@@ -14,11 +14,14 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.vertx.sqlclient.impl.command.CommandBase;
+import io.vertx.sqlclient.impl.command.CommandResponse;
 
 public class TdsMessageDecoder extends ChannelInboundHandlerAdapter {
 
   private final TdsMessageCodec tdsMessageCodec;
 
+  private ChannelHandlerContext chctx;
   private ByteBufAllocator alloc;
   private TdsMessage message;
 
@@ -28,6 +31,7 @@ public TdsMessageDecoder(TdsMessageCodec tdsMessageCodec) {
 
   @Override
   public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+    chctx = ctx;
     alloc = ctx.alloc();
   }
 
@@ -49,6 +53,12 @@ public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
     releaseMessage();
   }
 
+  void fireCommandResponse(CommandResponse<?> commandResponse) {
+    MSSQLCommandCodec<?, ?> c = tdsMessageCodec.poll();
+    commandResponse.cmd = (CommandBase) c.cmd;
+    chctx.fireChannelRead(commandResponse);
+  }
+
   private void releaseMessage() {
     if (message != null) {
       message.release();

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/codec/TdsMessageEncoder.java

@@ -52,13 +52,9 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
 
   void write(CommandBase<?> cmd) {
     MSSQLCommandCodec<?, ?> codec = wrap(cmd);
-    codec.completionHandler = resp -> {
-      MSSQLCommandCodec<?, ?> c = this.tdsMessageCodec.poll();
-      resp.cmd = (CommandBase) c.cmd;
-      chctx.fireChannelRead(resp);
-    };
-    this.tdsMessageCodec.add(codec);
-    codec.encode();
+    if (tdsMessageCodec.add(codec)) {
+      codec.encode();
+    }
   }
 
   private MSSQLCommandCodec<?, ?> wrap(CommandBase<?> cmd) {

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

@@ -75,7 +75,7 @@ protected final void handleAuthMoreData(byte[] password, ByteBuf payload) {
       } else if (flag == FAST_AUTH_STATUS_FLAG) {
         // fast auth success
       } else {
-        encoder.handleCommandResponse(CommandResponse.failure(new UnsupportedOperationException("Unsupported flag for AuthMoreData : " + flag)));
+        encoder.fireCommandResponse(CommandResponse.failure(new UnsupportedOperationException("Unsupported flag for AuthMoreData : " + flag)));
       }
     }
   }
@@ -86,7 +86,7 @@ protected final void sendEncryptedPasswordWithServerRsaPublicKey(byte[] password
       byte[] passwordInput = Arrays.copyOf(password, password.length + 1); // need to append 0x00(NULL) to the password
       encryptedPassword = RsaPublicKeyEncryptor.encrypt(passwordInput, authPluginData, serverRsaPublicKeyContent);
     } catch (Exception e) {
-      encoder.handleCommandResponse(CommandResponse.failure(e));
+      encoder.fireCommandResponse(CommandResponse.failure(e));
       return;
     }
     sendBytesAsPacket(encryptedPassword);