Implement command scheduling for Oracle client (#1294)

See #1087

Instead of executing commands in any order, or worse, concurrently on
the same JDBC connection.

Signed-off-by: Thomas Segismont <[email protected]>

Author: tsegismont

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011-2022 Contributors to the Eclipse Foundation
+ * Copyright (c) 2011-2023 Contributors to the Eclipse Foundation
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License 2.0 which is available at
@@ -12,20 +12,14 @@
 
 import io.vertx.core.*;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.impl.ContextInternal;
 import io.vertx.oracleclient.OracleException;
 import io.vertx.sqlclient.Tuple;
 import oracle.sql.TIMESTAMPTZ;
 
 import java.sql.*;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Flow;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
-import static io.vertx.oracleclient.impl.FailureUtil.sanitize;
-
 public class Helper {
 
   public static void closeQuietly(AutoCloseable autoCloseable) {
@@ -78,66 +72,6 @@ public static void runOrHandleSQLException(ThrowingRunnable runnable)
     }
   }
 
-  public static <T> Future<T> first(Flow.Publisher<T> publisher, ContextInternal context) {
-    Promise<T> promise = context.promise();
-    publisher.subscribe(new Flow.Subscriber<>() {
-      volatile Flow.Subscription subscription;
-
-      @Override
-      public void onSubscribe(Flow.Subscription subscription) {
-        this.subscription = subscription;
-        subscription.request(1);
-      }
-
-      @Override
-      public void onNext(T item) {
-        context.runOnContext(x -> promise.tryComplete(item));
-        subscription.cancel();
-      }
-
-      @Override
-      public void onError(Throwable throwable) {
-        promise.fail(sanitize(throwable));
-      }
-
-      @Override
-      public void onComplete() {
-        // Use tryComplete as the completion signal can be sent even if we cancelled.
-        // Also for Publisher<Void> we would get in this case.
-        promise.tryComplete(null);
-      }
-    });
-    return promise.future();
-  }
-
-  public static <T> Future<List<T>> collect(Flow.Publisher<T> publisher, ContextInternal context) {
-    Promise<List<T>> promise = context.promise();
-    publisher.subscribe(new Flow.Subscriber<>() {
-      final List<T> list = new ArrayList<>();
-
-      @Override
-      public void onSubscribe(Flow.Subscription subscription) {
-        subscription.request(Long.MAX_VALUE);
-      }
-
-      @Override
-      public void onNext(T item) {
-        list.add(item);
-      }
-
-      @Override
-      public void onError(Throwable throwable) {
-        promise.fail(sanitize(throwable));
-      }
-
-      @Override
-      public void onComplete() {
-        promise.complete(list);
-      }
-    });
-    return promise.future();
-  }
-
   public static Object convertSqlValue(Object value) throws SQLException {
     if (value == null) {
       return null;

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

@@ -22,8 +22,7 @@
 import oracle.jdbc.OracleConnection;
 import oracle.jdbc.datasource.OracleDataSource;
 
-import java.sql.SQLException;
-
+import static io.vertx.oracleclient.impl.Helper.executeBlocking;
 import static io.vertx.oracleclient.impl.OracleDatabaseHelper.createDataSource;
 
 public class OracleConnectionFactory implements ConnectionFactory {
@@ -45,14 +44,10 @@ public void close(Promise<Void> promise) {
   public Future<SqlConnection> connect(Context context) {
     ContextInternal ctx = (ContextInternal) context;
     QueryTracer tracer = ctx.tracer() == null ? null : new QueryTracer(ctx.tracer(), options);
-    return context.<OracleConnection>executeBlocking(prom -> {
-      try {
-        prom.complete(datasource.createConnectionBuilder().build());
-      } catch (SQLException e) {
-        prom.fail(e);
-      }
-    }).map(ora -> {
-      CommandHandler conn = new CommandHandler((ContextInternal) context, options, ora);
+    return executeBlocking(context, () -> {
+      OracleConnection orac = datasource.createConnectionBuilder().build();
+      OracleMetadata metadata = new OracleMetadata(orac.getMetaData());
+      OracleJdbcConnection conn = new OracleJdbcConnection(ctx, options, orac, metadata);
       OracleConnectionImpl msConn = new OracleConnectionImpl(ctx, this, conn, tracer, null);
       conn.init(msConn);
       return msConn;