Support EXECUTOR option

Author: Michael-A-McMahon

src/main/java/oracle/r2dbc/impl/OracleConnectionFactoryImpl.java

@@ -28,12 +28,15 @@
 import io.r2dbc.spi.IsolationLevel;
 import io.r2dbc.spi.R2dbcException;
 import io.r2dbc.spi.Statement;
+import oracle.r2dbc.OracleR2dbcOptions;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 
 import javax.sql.DataSource;
 import java.time.Duration;
 import java.util.Optional;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
 
 /**
  * <p>
@@ -102,6 +105,12 @@ final class OracleConnectionFactoryImpl implements ConnectionFactory {
   /** JDBC data source that this factory uses to open connections */
   private final DataSource dataSource;
 
+  /**
+   * Executor configured by {@link oracle.r2dbc.OracleR2dbcOptions#EXECUTOR},
+   * or a default one if none was configured.
+   */
+  private final Executor executor;
+
   /**
    * <p>
    * Timeout applied to the execution of {@link Statement}s created by
@@ -189,6 +198,19 @@ final class OracleConnectionFactoryImpl implements ConnectionFactory {
         : Duration.parse(timeout.toString()))
       .orElse(Duration.ZERO);
 
+    Object executor = options.getValue(OracleR2dbcOptions.EXECUTOR);
+    if (executor == null) {
+      this.executor = ForkJoinPool.commonPool();
+    }
+    else if (executor instanceof Executor) {
+      this.executor = (Executor) executor;
+    }
+    else {
+      throw new IllegalArgumentException(
+        "Value of " + OracleR2dbcOptions.EXECUTOR
+          + " is not an instance of Executor: " + executor.getClass());
+    }
+
   }
 
   /**
@@ -221,7 +243,7 @@ public Publisher<Connection> create() {
       // to a particular connection.
       ReactiveJdbcAdapter adapter = ReactiveJdbcAdapter.getOracleAdapter();
 
-      return Mono.fromDirect(adapter.publishConnection(dataSource))
+      return Mono.fromDirect(adapter.publishConnection(dataSource, executor))
         .flatMap(conn -> {
           OracleConnectionImpl connection =
             new OracleConnectionImpl(conn, adapter);

src/main/java/oracle/r2dbc/impl/OracleReactiveJdbcAdapter.java

@@ -57,6 +57,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -108,7 +109,10 @@
  *     {@link UsingConnectionSubscriber} for more details.
  *   </li>
  * </ul><p>
- * A instance of this class is obtained by invoking {@link #getInstance()}.
+ * A instance of this class is obtained by invoking {@link #getInstance()}. A
+ * new instance should be created each time a JDBC {@code Connection} is
+ * created, and that instance should be used to execute database calls with
+ * that {@code Connection} only.
  * </p><p>
  * All JDBC type parameters supplied to the methods of this class must
  * {@linkplain Wrapper#isWrapperFor(Class) wrap} an Oracle JDBC interface
@@ -137,7 +141,7 @@ final class OracleReactiveJdbcAdapter implements ReactiveJdbcAdapter {
    * does or how it should be configured.
    */
   private static final Set<Option<CharSequence>>
-    SUPPORTED_CONNECTION_PROPERTY_OPTIONS = Set.of(
+    JDBC_CONNECTION_PROPERTY_OPTIONS = Set.of(
 
       // Support TNS_ADMIN (tnsnames.ora, ojdbc.properties).
       OracleR2dbcOptions.TNS_ADMIN,
@@ -489,10 +493,9 @@ private static void configureStandardOptions(
 
   /**
    * Configures an {@code oracleDataSource} with the values of extended R2DBC
-   * {@code Options}. Extended options are those declared as
-   * {@link #SUPPORTED_CONNECTION_PROPERTY_OPTIONS} in this class. The values
-   * of these options are used to configure the {@code oracleDataSource} as
-   * specified in the javadoc of
+   * {@code Options}. Extended options are those declared in
+   * {@link OracleR2dbcOptions}. The values of these options are used to
+   * configure the {@code oracleDataSource} as specified in the javadoc of
    * {@link #createDataSource(ConnectionFactoryOptions)}
    * @param oracleDataSource An data source to configure
    * @param options R2DBC options. Not null.
@@ -509,8 +512,8 @@ private static void configureExtendedOptions(
           OracleConnection.CONNECTION_PROPERTY_TNS_ADMIN, tnsAdmin.toString()));
     }
 
-    // Apply any extended options as connection properties
-    for (Option<CharSequence> option : SUPPORTED_CONNECTION_PROPERTY_OPTIONS) {
+    // Apply any JDBC connection property options
+    for (Option<CharSequence> option : JDBC_CONNECTION_PROPERTY_OPTIONS) {
       // Using Object as the value type allows options to be set as types like
       // Boolean or Integer. These types make sense for numeric or boolean
       // connection property values, such as statement cache size, or enable x.
@@ -662,11 +665,12 @@ private static void setPropertyIfAbsent(
    */
   @Override
   public Publisher<? extends Connection> publishConnection(
-    DataSource dataSource) {
+    DataSource dataSource, Executor executor) {
     OracleDataSource oracleDataSource = unwrapOracleDataSource(dataSource);
     return Mono.from(adaptFlowPublisher(() ->
         oracleDataSource
           .createConnectionBuilder()
+          .executorOracle(executor)
           .buildConnectionPublisherOracle()))
       .onErrorMap(R2dbcException.class, error ->
         error.getErrorCode() == 18714 // ORA-18714 : Login timeout expired

src/main/java/oracle/r2dbc/impl/ReactiveJdbcAdapter.java

@@ -38,6 +38,7 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.concurrent.Executor;
 import java.util.function.Function;
 
 /**
@@ -124,6 +125,8 @@ static ReactiveJdbcAdapter getOracleAdapter() throws R2dbcException {
    * Publishes a single JDBC {@link Connection} to a single subscriber. The
    * connection is established as if by invoking
    * {@link DataSource#getConnection()} on the specified {@code dataSource}.
+   * The {@code Connection} is configured to execute asynchronous tasks using
+   * the provided {@code executor}.
    * </p><p>
    * The {@code dataSource} is retained by the returned publisher, meaning
    * that any mutable state of the {@code dataSource} can affect the behavior
@@ -144,10 +147,12 @@ static ReactiveJdbcAdapter getOracleAdapter() throws R2dbcException {
    * </p>
    * @param dataSource JDBC data source that is configured to establish a
    *   connection. Not null.
+   * @param executor Executor to use for executing asynchronous tasks. Not null.
    * @return A publisher that emits a JDBC connection. Not null.
    * @throws R2dbcException If a database access error occurs.
    */
-  Publisher<? extends Connection> publishConnection(DataSource dataSource)
+  Publisher<? extends Connection> publishConnection(
+    DataSource dataSource, Executor executor)
     throws R2dbcException;
 
   /**

src/test/java/oracle/r2dbc/impl/OracleReactiveJdbcAdapterTest.java

@@ -29,7 +29,9 @@
 import io.r2dbc.spi.Result;
 import oracle.jdbc.OracleConnection;
 import oracle.jdbc.datasource.OracleDataSource;
+import oracle.r2dbc.OracleR2dbcOptions;
 import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.io.IOException;
@@ -41,11 +43,17 @@
 import java.sql.SQLException;
 import java.time.Duration;
 import java.util.Properties;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 
 import static io.r2dbc.spi.ConnectionFactoryOptions.CONNECT_TIMEOUT;
 import static io.r2dbc.spi.ConnectionFactoryOptions.DATABASE;
@@ -415,6 +423,52 @@ public void testStatementTimeout() {
 
   }
 
+  /**
+   * Verifies the {@link oracle.r2dbc.OracleR2dbcOptions#EXECUTOR} option
+   */
+  @Test
+  public void testExecutorOption() {
+
+    // Create a custom executor that increments a count when Runnables are
+    // submitted, and then delegates to a single threaded executor.
+    AtomicInteger count = new AtomicInteger(0);
+    ExecutorService singleThread = Executors.newSingleThreadExecutor();
+    Executor testExecutor = runnable -> {
+      count.incrementAndGet();
+      singleThread.execute(runnable);
+    };
+
+    // Create a connection that is configured to use the custom executor
+    Connection connection = awaitOne(ConnectionFactories.get(
+      ConnectionFactoryOptions.builder()
+        .option(OracleR2dbcOptions.EXECUTOR, testExecutor)
+        .option(DRIVER, "oracle")
+        .option(HOST, host())
+        .option(PORT, port())
+        .option(DATABASE, serviceName())
+        .option(USER, user())
+        .option(PASSWORD, password())
+        .build())
+        .create());
+
+    try {
+      // Make some asynchronous database calls and expect the executor's
+      // count to be incremented
+      awaitOne(Set.of(0, 1, 2, 3), Flux.merge(
+        connection.createStatement("SELECT 0 FROM sys.dual").execute(),
+        connection.createStatement("SELECT 1 FROM sys.dual").execute(),
+        connection.createStatement("SELECT 2 FROM sys.dual").execute(),
+        connection.createStatement("SELECT 3 FROM sys.dual").execute())
+        .flatMap(result ->
+          result.map(row -> row.get(0, Integer.class)))
+        .collect(Collectors.toSet()));
+      assertTrue(count.get() != 0);
+    }
+    finally {
+      tryAwaitNone(connection.close());
+    }
+  }
+
   /**
    * Verifies that an attempt to connect with a {@code listeningChannel}
    * results in an {@link R2dbcTimeoutException}.