Merge branch 'main' of https://github.com/oracle/oracle-r2dbc into descriptor-option

Author: Michael-A-McMahon

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

@@ -24,11 +24,16 @@
 import java.sql.Connection;
 import java.util.LinkedList;
 import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
 
 import io.r2dbc.spi.Batch;
 import io.r2dbc.spi.R2dbcException;
 import io.r2dbc.spi.Result;
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
 import io.r2dbc.spi.Statement;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -104,6 +109,18 @@ public Batch add(String sql) {
    * are executed in the order they were added. Calling this method clears all
    * statements that have been added to the current batch.
    * </p><p>
+   * A {@code Result} emitted by the returned {@code Publisher} must be
+   * <a href="OracleStatementImpl.html#fully-consumed-result">
+   *   fully-consumed
+   * </a>
+   * before the next {@code Result} is emitted. This ensures that a command in
+   * the batch can not be executed while the {@code Result} of a previous
+   * command is consumed concurrently. It is a known limitation of the Oracle
+   * R2DBC Driver that concurrent operations on a single {@code Connection}
+   * will result in blocked threads. Deferring {@code Statement} execution
+   * until full consumption of the previous {@code Statement}'s {@code Result}
+   * is necessary in order to avoid blocked threads.
+   * </p><p>
    * If the execution of any statement in the sequence results in a failure,
    * then the returned publisher emits {@code onError} with an
    * {@link R2dbcException} that describes the failure, and all subsequent
@@ -126,17 +143,121 @@ public Publisher<? extends Result> execute() {
     statements = new LinkedList<>();
 
     AtomicBoolean isSubscribed = new AtomicBoolean(false);
-    return Flux.defer(() -> {
-      if (isSubscribed.compareAndSet(false, true)) {
-        return Flux.fromIterable(currentStatements)
-          .concatMap(Statement::execute);
-      }
-      else {
-        return Mono.error(new IllegalStateException(
+    return Flux.defer(() -> isSubscribed.compareAndSet(false, true)
+      ? executeBatch(currentStatements)
+      : Mono.error(new IllegalStateException(
           "Multiple subscribers are not supported by the Oracle R2DBC" +
-            " Batch.execute() publisher"));
-      }
-    });
+            " Batch.execute() publisher")));
+  }
+
+  /**
+   * Executes each {@code Statement} in a {@code Queue} of {@code statements}.
+   * A {@code Statement} is not executed until the {@code Result} of any
+   * previous {@code Statement} is fully-consumed.
+   * @param statements {@code Statement}s to execute. Not null.
+   * @return A {@code Publisher} of each {@code Statement}'s {@code Result}.
+   * Not null.
+   */
+  private static Publisher<? extends Result> executeBatch(
+    Queue<Statement> statements) {
+
+    // Reference a Publisher that terminates when the previous Statement's
+    // Result has been consumed.
+    AtomicReference<Publisher<Void>> previous =
+      new AtomicReference<>(Mono.empty());
+
+    return Flux.fromIterable(statements)
+      .concatMap(statement -> {
+
+        // Complete when this statement's result is consumed
+        CompletableFuture<Void> next = new CompletableFuture<>();
+
+        return Flux.from(statement.execute())
+          // Delay execution by delaying Publisher.subscribe(Subscriber) until the
+          // previous statement's result is consumed.
+          .delaySubscription(
+            // Update the reference; This statement is now the "previous"
+            // statement.
+            previous.getAndSet(Mono.fromCompletionStage(next)))
+          // Batch result completes the "next" future when fully consumed.
+          .map(result -> new BatchResult(next, result));
+      });
+  }
+
+  /**
+   * <p>
+   * A {@code Result} that completes a {@link CompletableFuture} when it has
+   * been fully consumed. Instances of {@code BatchResult} are used by Oracle
+   * R2DBC to ensure that statement execution and row data processing do
+   * not occur concurrently; The completion of the future signals that the row
+   * data of a result has been fully consumed, and that no more database
+   * calls will be initiated to fetch additional rows.
+   * </p><p>
+   * Instances of {@code BatchResult} delegate invocations of
+   * {@link #getRowsUpdated()} and {@link #map(BiFunction)} to a
+   * {@code Result} provided on construction; The behavior of {@code Publisher}s
+   * returned by these methods is identical to those returned by the delegate
+   * {@code Result}.
+   * </p>
+   */
+  private static final class BatchResult implements Result {
+
+    /** Completed when this {@code BatchResult} is fully consumed */
+     final CompletableFuture<Void> consumeFuture;
+
+    /** Delegate {@code Result} that provides row data or an update count */
+    final Result delegateResult;
+
+    /**
+     * Constructs a new result that completes a {@code consumeFuture} when the
+     * row data or update count of a {@code delegateResult} has been fully
+     * consumed.
+     * @param consumeFuture Future completed upon consumption
+     * @param delegateResult Result of row data or an update count
+     */
+    BatchResult(CompletableFuture<Void> consumeFuture, Result delegateResult) {
+      this.consumeFuture = consumeFuture;
+      this.delegateResult = delegateResult;
+    }
+
+    /**
+     * {@inheritDoc}
+     * <p>
+     * Immediately completes the {@link #consumeFuture} and then returns the
+     * update count {@code Publisher} of the {@link #delegateResult}. After
+     * returning an update count {@code Publisher}, the {@link #delegateResult}
+     * can not initiate any more database calls (based on the assumption
+     * noted below).
+     * </p>
+     * @implNote It is assumed that the {@link #delegateResult} will throw
+     * {@link IllegalStateException} upon multiple attempts to consume it, and
+     * this method does not check for multiple consumptions.
+     */
+    @Override
+    public Publisher<Integer> getRowsUpdated() {
+      consumeFuture.complete(null);
+      return Flux.from(delegateResult.getRowsUpdated());
+    }
+
+    /**
+     * {@inheritDoc}
+     * <p>
+     * Completes the {@link #consumeFuture} after the row data {@code
+     * Publisher} of the {@link #delegateResult} emits a terminal signal or
+     * has it's {@code Subscription} cancelled. After emitting a terminal
+     * signal or having it's {@code Subscription} cancelled, the
+     * {@link #delegateResult} can not initiate any more database calls.
+     * </p>
+     * @implNote It is assumed that the {@link #delegateResult} will throw
+     * {@link IllegalStateException} upon multiple attempts to consume it, and
+     * this method does not check for multiple consumptions.
+     */
+    @Override
+    public <T> Publisher<T> map(
+      BiFunction<Row, RowMetadata, ? extends T> mappingFunction) {
+      return Flux.<T>from(delegateResult.map(mappingFunction))
+        .doFinally(signalType -> consumeFuture.complete(null));
+    }
   }
 }
 

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

@@ -86,6 +86,11 @@
  *     TCPS (ie: SSL/TLS).
  *   </dd>
  * </dl>
+ * <h3 id="extended_options">Supported Options</h3><p>
+ * This implementation supports extended options having the name of a
+ * subset of Oracle JDBC connection properties. The list of supported
+ * connection properties is specified by {@link OracleReactiveJdbcAdapter}.
+ * </p>
  *
  * @author  harayuanwang, michael-a-mcmahon
  * @since   0.1.0
@@ -169,14 +174,15 @@ final class OracleConnectionFactoryImpl implements ConnectionFactory {
    * the returned publisher, so that the database can reclaim the resources
    * allocated for that connection.
    * </p><p>
-   * The returned publisher does not support multiple subscribers. After a
-   * subscriber has subscribed, the returned publisher emits {@code onError}
-   * with an {@link IllegalStateException} to all subsequent subscribers.
+   * The returned publisher supports multiple subscribers. One {@code
+   * Connection} is emitted to each subscriber that subscribes and signals
+   * demand.
    * </p>
    */
   @Override
   public Publisher<Connection> create() {
-    return Mono.fromDirect(adapter.publishConnection(dataSource))
+    return Mono.defer(() ->
+        Mono.fromDirect(adapter.publishConnection(dataSource)))
       .map(conn -> new OracleConnectionImpl(adapter, conn));
   }
 

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

@@ -559,7 +559,10 @@ private static void configureExtendedOptions(
 
     // Apply any extended options as connection properties
     for (Option<CharSequence> option : SUPPORTED_CONNECTION_PROPERTY_OPTIONS) {
-      CharSequence value = options.getValue(option);
+      // 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.
+      Object value = options.getValue(option);
       if (value != null) {
         runOrHandleSQLException(() ->
           oracleDataSource.setConnectionProperty(

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

@@ -35,6 +35,7 @@
 
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.getOrHandleSQLException;
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.requireNonNull;
+import static oracle.r2dbc.impl.OracleR2dbcExceptions.runOrHandleSQLException;
 
 /**
  * <p>
@@ -68,17 +69,21 @@ private OracleResultImpl() { }
 
   /**
    * Creates a {@code Result} that publishes either an empty stream of row
-   * data, or publishes an {@code updateCount} if it is greater than zero. An
-   * {@code updateCount} less than 1 is published as an empty stream.
+   * data, or publishes an {@code updateCount} if it is greater than or equal
+   * to zero. An {@code updateCount} less than zero is published as an empty
+   * stream.
    * @param updateCount Update count to publish
    * @return An update count {@code Result}
    */
   public static Result createUpdateCountResult(int updateCount) {
     return new OracleResultImpl() {
 
+      final Publisher<Integer> updateCountPublisher =
+        updateCount < 0 ? Mono.empty() : Mono.just(updateCount);
+
       @Override
       Publisher<Integer> publishUpdateCount() {
-        return updateCount < 1 ? Mono.empty() : Mono.just(updateCount);
+        return updateCountPublisher;
       }
 
       @Override
@@ -90,9 +95,13 @@ <T> Publisher<T> publishRows(
   }
 
   /**
+   * <p>
    * Creates a {@code Result} that either publishes a {@code ResultSet} of
    * row data from a query, or publishes an update count as an empty stream.
-   *
+   * </p><p>
+   * The {@link java.sql.Statement} that created the {@code resultSet} is closed
+   * when the returned result is fully consumed.
+   * </p>
    * @param adapter Adapts {@code ResultSet} API calls into reactive streams.
    *   Not null.
    * @param resultSet Row data to publish
@@ -105,19 +114,29 @@ public static Result createQueryResult(
 
       @Override
       Publisher<Integer> publishUpdateCount() {
+        runOrHandleSQLException(() ->
+          resultSet.getStatement().close());
         return Mono.empty();
       }
 
       @Override
       <T> Publisher<T> publishRows(
         BiFunction<Row, RowMetadata, ? extends T> mappingFunction) {
 
+        // Obtain a reference to the statement before the ResultSet is
+        // logically closed by its row publisher. The statement is closed when
+        // the publisher terminates.
+        java.sql.Statement jdbcStatement =
+          getOrHandleSQLException(resultSet::getStatement);
+
         OracleRowMetadataImpl metadata = new OracleRowMetadataImpl(
           getOrHandleSQLException(resultSet::getMetaData));
 
-        return Flux.from(adapter.publishRows(resultSet, jdbcRow ->
+        return Flux.<T>from(adapter.publishRows(resultSet, jdbcRow ->
           mappingFunction.apply(
-            new OracleRowImpl(jdbcRow, metadata, adapter), metadata)));
+            new OracleRowImpl(jdbcRow, metadata, adapter), metadata)))
+          .doFinally(signalType ->
+            runOrHandleSQLException(jdbcStatement::close));
       }
     };
   }
@@ -128,10 +147,15 @@ <T> Publisher<T> publishRows(
    * {@link PreparedStatement#getGeneratedKeys()} {@code ResultSet}, or
    * publishes an {@code updateCount}.
    * </p><p>
+   * The {@link java.sql.Statement} that created the {@code ResultSet} is closed
+   * when the {@code Publisher} returned by this method emits a
+   * {@code Result}.
+   * </p><p>
    * For compliance with R2DBC standards, a {@code Row} of generated column
    * values will remain valid after the {@code Connection} that created them
    * is closed. This behavior is verified by version 0.8.2 of
-   * {@code io.r2dbc.spi.test.TestKit#returnGeneratedValues()}.
+   * {@code io.r2dbc.spi.test.TestKit#returnGeneratedValues()}. The {@code Rows}
+   * of generated value
    * </p>
    *
    * @implNote
@@ -167,16 +191,25 @@ public static Publisher<Result> createGeneratedValuesResult(
 
     // Avoid invoking ResultSet.getMetaData() on an empty ResultSet, it may
     // throw a SQLException
-    if (! getOrHandleSQLException(values::isBeforeFirst))
+    if (! getOrHandleSQLException(values::isBeforeFirst)) {
+      runOrHandleSQLException(() -> values.getStatement().close());
       return Mono.just(createUpdateCountResult(updateCount));
+    }
 
     // Obtain metadata before the ResultSet is closed by publishRows(...)
     OracleRowMetadataImpl metadata =
       new OracleRowMetadataImpl(getOrHandleSQLException(values::getMetaData));
 
+    // Obtain a reference to the statement before the ResultSet is
+    // logically closed by its row publisher. The statement is closed when
+    // the publisher terminates.
+    java.sql.Statement jdbcStatement =
+      getOrHandleSQLException(values::getStatement);
+
     return Flux.from(adapter.publishRows(
       values, ReactiveJdbcAdapter.JdbcRow::copy))
       .collectList()
+      .doFinally(signalType -> runOrHandleSQLException(jdbcStatement::close))
       .map(cachedRows -> new OracleResultImpl() {
 
         @Override