Use a pending list of completables for Oracle JDBC connection

Author: vietj

vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleJdbcConnection.java

@@ -35,6 +35,8 @@
 
 public class OracleJdbcConnection implements Connection {
 
+  private static final Completable<?> NULL_HANDLER = (res, err) -> {};
+
   private static final Logger log = LoggerFactory.getLogger(OracleJdbcConnection.class);
 
   private final ClientMetrics metrics;
@@ -49,6 +51,7 @@ public class OracleJdbcConnection implements Connection {
   // Command pipeline state
   @SuppressWarnings("rawtypes")
   private final Deque<CommandBase> pending = new ArrayDeque<>();
+  private final Deque<Completable<?>> completables = new ArrayDeque<>();
   private Promise<Void> closePromise;
   private boolean inflight, executing;
 
@@ -128,6 +131,7 @@ public void close(Holder holder, Completable<Void> promise) {
         closePromise = context.promise();
         future = closePromise.future().andThen(ar -> holder.handleClosed());
         pending.add(CloseConnectionCommand.INSTANCE);
+        completables.add(NULL_HANDLER);
         checkPending();
       } else {
         future = closePromise.future();
@@ -172,9 +176,9 @@ public <R> void schedule(CommandBase<R> cmd, Completable<R> handler) {
   }
 
   private <R> void doSchedule(CommandBase<R> cmd, Completable<R> handler) {
-    cmd.handler = handler;
     if (closePromise == null) {
       pending.add(cmd);
+      completables.add(handler);
       checkPending();
     } else {
       cmd.fail(VertxException.noStackTrace("Connection is no longer active"));
@@ -190,11 +194,13 @@ private void checkPending() {
       executing = true;
       CommandBase cmd;
       while (!inflight && (cmd = pending.poll()) != null) {
+        Completable handler = completables.poll();
         inflight = true;
         if (metrics != null && cmd instanceof CloseConnectionCommand) {
           metrics.close();
         }
         OracleCommand action = wrap(cmd);
+        action.handler = handler;
         Future<Void> future = action.processCommand(cmd);
         CommandBase capture = cmd;
         future.onComplete(ar -> actionComplete(capture, action, ar));

vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/commands/OracleCommand.java

@@ -10,6 +10,7 @@
  */
 package io.vertx.oracleclient.impl.commands;
 
+import io.vertx.core.Completable;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.internal.ContextInternal;
@@ -25,6 +26,7 @@
 
 public abstract class OracleCommand<T> {
 
+  public Completable<T> handler;
   protected final OracleConnection oracleConnection;
   protected final ContextInternal connectionContext;
   private CommandResponse<T> response;