Message codec improvements

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);

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

@@ -48,7 +48,7 @@ void decodePayload(ByteBuf payload, int payloadLength) {
         handleAuthMoreData(cmd.password().getBytes(StandardCharsets.UTF_8), payload);
         break;
       case OK_PACKET_HEADER:
-        encoder.handleCommandResponse(CommandResponse.success(null));
+        encoder.fireCommandResponse(CommandResponse.success(null));
         break;
       case ERROR_PACKET_HEADER:
         handleErrorPacketPayload(payload);
@@ -75,7 +75,7 @@ private void handleAuthSwitchRequest(String password, ByteBuf payload) {
         sendNonSplitPacket(buffer);
         break;
       default:
-        encoder.handleCommandResponse(CommandResponse.failure(new UnsupportedOperationException("Unsupported authentication method: " + pluginName)));
+        encoder.fireCommandResponse(CommandResponse.failure(new UnsupportedOperationException("Unsupported authentication method: " + pluginName)));
         return;
     }
   }

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

@@ -34,7 +34,7 @@ void encode(MySQLEncoder encoder) {
   void decodePayload(ByteBuf payload, int payloadLength) {
     // no statement response
     // it will be called by the connection in order
-    encoder.handleCommandResponse(CommandResponse.success(null));
+    encoder.fireCommandResponse(CommandResponse.success(null));
   }
 
   private void sendCloseStatementCommand(MySQLPreparedStatement statement) {

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

@@ -111,7 +111,7 @@ void handleOkPacketOrErrorPacketPayload(ByteBuf payload) {
     switch (header) {
       case EOF_PACKET_HEADER:
       case OK_PACKET_HEADER:
-        encoder.handleCommandResponse(CommandResponse.success(null));
+        encoder.fireCommandResponse(CommandResponse.success(null));
         break;
       case ERROR_PACKET_HEADER:
         handleErrorPacketPayload(payload);
@@ -121,7 +121,7 @@ void handleOkPacketOrErrorPacketPayload(ByteBuf payload) {
 
   void handleErrorPacketPayload(ByteBuf payload) {
     MySQLException mySQLException = decodeErrorPacketPayload(payload);
-    encoder.handleCommandResponse(CommandResponse.failure(mySQLException));
+    encoder.fireCommandResponse(CommandResponse.failure(mySQLException));
   }
 
   final MySQLException decodeErrorPacketPayload(ByteBuf payload) {

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

@@ -43,7 +43,7 @@ class ExtendedBatchQueryCommandCodec<R> extends ExtendedQueryCommandBaseCodec<R,
   void encode(MySQLEncoder encoder) {
     super.encode(encoder);
     if (params.isEmpty() && statement.paramDesc.paramDefinitions().length > 0) {
-      encoder.handleCommandResponse(CommandResponse.failure("Statement parameter is not set because of the empty batch param list"));
+      encoder.fireCommandResponse(CommandResponse.failure("Statement parameter is not set because of the empty batch param list"));
       return;
     }
     pipeliningEnabled = encoder.socketConnection.pipeliningEnabled();
@@ -61,7 +61,7 @@ void handleErrorPacketPayload(ByteBuf payload) {
     if (received == params.size()) {
       super.closePreparedStatement();
       encoder.socketConnection.resumePipeline();
-      encoder.handleCommandResponse(CommandResponse.failure(failure));
+      encoder.fireCommandResponse(CommandResponse.failure(failure));
     } else {
       doExecuteBatch();
     }

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

@@ -50,7 +50,7 @@ void encode(MySQLEncoder encoder) {
       // binding parameters
       String bindMsg = statement.bindParameters(params);
       if (bindMsg != null) {
-        encoder.handleCommandResponse(CommandResponse.failure(bindMsg));
+        encoder.fireCommandResponse(CommandResponse.failure(bindMsg));
         return;
       }
 

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

@@ -135,7 +135,7 @@ private void handleInitialHandshake(ByteBuf payload) {
         upgradeToSsl = true;
         break;
       default:
-        encoder.handleCommandResponse(CommandResponse.failure(new IllegalStateException("Unknown SSL mode to handle: " + sslMode)));
+        encoder.fireCommandResponse(CommandResponse.failure(new IllegalStateException("Unknown SSL mode to handle: " + sslMode)));
         return;
     }
 
@@ -147,7 +147,7 @@ private void handleInitialHandshake(ByteBuf payload) {
         if (upgrade.succeeded()) {
           doSendHandshakeResponseMessage(serverAuthPluginName, cmd.authenticationPlugin(), authPluginData, serverCapabilitiesFlags);
         } else {
-          encoder.handleCommandResponse(CommandResponse.failure(upgrade.cause()));
+          encoder.fireCommandResponse(CommandResponse.failure(upgrade.cause()));
         }
       });
     } else {
@@ -167,7 +167,7 @@ private void handleAuthentication(ByteBuf payload) {
     switch (header) {
       case OK_PACKET_HEADER:
         status = ST_CONNECTED;
-        encoder.handleCommandResponse(CommandResponse.success(cmd.connection()));
+        encoder.fireCommandResponse(CommandResponse.success(cmd.connection()));
         break;
       case ERROR_PACKET_HEADER:
         handleErrorPacketPayload(payload);
@@ -179,7 +179,7 @@ private void handleAuthentication(ByteBuf payload) {
         handleAuthMoreData(cmd.password().getBytes(StandardCharsets.UTF_8), payload);
         break;
       default:
-        encoder.handleCommandResponse(CommandResponse.failure(new IllegalStateException("Unhandled state with header: " + header)));
+        encoder.fireCommandResponse(CommandResponse.failure(new IllegalStateException("Unhandled state with header: " + header)));
     }
   }
 
@@ -202,7 +202,7 @@ private void handleAuthSwitchRequest(String password, ByteBuf payload) {
         sendNonSplitPacket(buffer);
         break;
       default:
-        encoder.handleCommandResponse(CommandResponse.failure(new UnsupportedOperationException("Unsupported authentication method: " + pluginName)));
+        encoder.fireCommandResponse(CommandResponse.failure(new UnsupportedOperationException("Unsupported authentication method: " + pluginName)));
         return;
     }
   }