Merge pull request #24 from oracle/22-poolable-connections

22 poolable connections

Author: jeandelavarene

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

@@ -53,13 +53,21 @@
  *   <li>{@link ConnectionFactoryOptions#HOST}</li>
  * </ul>
  * <h3 id="supported_options">Supported Options</h3><p>
- * This implementation supports the following options for connection creation:
+ * This implementation supports the following well known options for connection
+ * creation:
  * </p><ul>
  *   <li>{@link ConnectionFactoryOptions#PORT}</li>
  *   <li>{@link ConnectionFactoryOptions#DATABASE}</li>
  *   <li>{@link ConnectionFactoryOptions#USER}</li>
  *   <li>{@link ConnectionFactoryOptions#PASSWORD}</li>
+ *   <li>{@link ConnectionFactoryOptions#CONNECT_TIMEOUT}</li>
+ *   <li>{@link ConnectionFactoryOptions#SSL}</li>
  * </ul>
+ * <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
@@ -116,7 +124,6 @@ final class OracleConnectionFactoryImpl implements ConnectionFactory {
    */
   OracleConnectionFactoryImpl(ConnectionFactoryOptions options) {
     OracleR2dbcExceptions.requireNonNull(options, "options is null.");
-
     adapter = ReactiveJdbcAdapter.getOracleAdapter();
     dataSource = adapter.createDataSource(options);
   }
@@ -138,14 +145,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

@@ -511,7 +511,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>
@@ -94,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
@@ -109,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));
       }
     };
   }
@@ -132,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
@@ -171,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

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

@@ -41,6 +41,7 @@
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
+import java.util.function.Function;
 
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.getOrHandleSQLException;
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.requireNonNull;
@@ -546,8 +547,8 @@ private Publisher<Result> createResultPublisher(
 
   /**
    * <p>
-   * Executes a JDBC statement with a single, non-batched, set of parameters.
-   * If the execution returns a {@link java.sql.ResultSet} then the
+   * Executes a JDBC statement with a single, non-batched, set of {@code
+   * bindValues}. If the execution results in a {@link java.sql.ResultSet} then the
    * {@code jdbcStatement} is closed after the {@code ResultSet} is fully
    * consumed by {@link Result#map(BiFunction)}. Otherwise, if the execution
    * only produces an update count, then the {@code jdbcStatement} is closed
@@ -569,7 +570,6 @@ private Publisher<Result> executeSingle(
     return Mono.from(adapter.publishSQLExecution(jdbcStatement))
       .map(isResultSet -> {
         if (isResultSet) {
-          runOrHandleSQLException(jdbcStatement::closeOnCompletion);
           return OracleResultImpl.createQueryResult(
               adapter, getOrHandleSQLException(jdbcStatement::getResultSet));
         }
@@ -614,22 +614,27 @@ private Publisher<Result> executeBatch(
 
     // Execute the batch. The execution won't return a ResultSet, so the JDBC
     // statement is closed immediately after the execution completes.
-    runOrHandleSQLException(jdbcStatement::closeOnCompletion);
     return Flux.from(adapter.publishBatchUpdate(jdbcStatement))
       .map(updateCount ->
-        OracleResultImpl.createUpdateCountResult(Math.toIntExact(updateCount)));
+        OracleResultImpl.createUpdateCountResult(Math.toIntExact(updateCount)))
+      .doFinally(signalType -> runOrHandleSQLException(jdbcStatement::close));
   }
 
   /**
    * <p>
-   * Executes a key generating {@code jdbcStatement} for each set
-   * of bind values in a {@code batch}. The {@code jdbcStatement} is closed
-   * after all executions have completed. If any execution results in an
-   * error, subsequent executions are skipped.
+   * Executes a JDBC statement with a single, non-batched, set of {@code
+   * bindValues}. If the execution results in a {@link java.sql.ResultSet} then
+   * the {@code jdbcStatement} is closed after the {@code ResultSet} is fully
+   * consumed by {@link Result#map(BiFunction)}. If the execution results in
+   * an update count and/or values generated by DML, then the {@code
+   * jdbcStatement} is closed after the returned {@code Publisher} emits a
+   * {@code Result} with all generated values cached, such that
+   * {@link Result#map(BiFunction)} may be called after the database connection
+   * has been closed.
    * </p><p>
-   * The returned {@code Publisher} emits a {@code Result} with an update
-   * count and generated values for each execution of the {@code
-   * jdbcStatement}.
+   * The returned publisher initiates SQL execution <i>the first time</i> a
+   * subscriber subscribes, before the subscriber emits a {@code request}
+   * signal.
    * </p>
    *
    * @implNote The 21c Oracle JDBC Driver does not support batch DML when
@@ -638,25 +643,23 @@ private Publisher<Result> executeBatch(
    *
    * @param jdbcStatement A JDBC statement
    * @param bindValues A set of bind values
-   * @return A publisher that emits the {@code Results} of executing the
-   * JDBC statement for each set of bind values in the {@code batch}
+   * @return A publisher that emits the {@code Result} of executing the
+   * JDBC statement.
    */
   private Publisher<Result> executeGeneratingValues(
     PreparedStatement jdbcStatement, Object[] bindValues) {
 
     setJdbcBindValues(bindValues, jdbcStatement);
     return Mono.from(adapter.publishSQLExecution(jdbcStatement))
-      .flatMap(isResultSet -> {
-        runOrHandleSQLException(jdbcStatement::closeOnCompletion);
-        return isResultSet
+      .flatMap(isResultSet ->
+        isResultSet
           ? Mono.just(OracleResultImpl.createQueryResult(
               adapter,
               getOrHandleSQLException(jdbcStatement::getResultSet)))
           : Mono.from(OracleResultImpl.createGeneratedValuesResult(
               adapter,
               getOrHandleSQLException(jdbcStatement::getUpdateCount),
-              getOrHandleSQLException(jdbcStatement::getGeneratedKeys)));
-      });
+              getOrHandleSQLException(jdbcStatement::getGeneratedKeys))));
   }
 
   /**

src/test/java/oracle/r2dbc/DatabaseConfig.java

@@ -25,6 +25,8 @@
 import io.r2dbc.spi.ConnectionFactories;
 import io.r2dbc.spi.ConnectionFactory;
 import io.r2dbc.spi.ConnectionFactoryOptions;
+import io.r2dbc.spi.Option;
+import oracle.jdbc.OracleConnection;
 import oracle.r2dbc.util.SharedConnectionFactory;
 import org.reactivestreams.Publisher;
 
@@ -216,6 +218,11 @@ public static int databaseVersion() {
           .option(ConnectionFactoryOptions.DATABASE, SERVICE_NAME)
           .option(ConnectionFactoryOptions.USER, USER)
           .option(ConnectionFactoryOptions.PASSWORD, PASSWORD)
+          // Disable statement caching in order to verify cursor closing;
+          // Cached statements don't close their cursors
+          .option(Option.valueOf(
+            OracleConnection.CONNECTION_PROPERTY_IMPLICIT_STATEMENT_CACHE_SIZE),
+            0)
           .build());
 
       SHARED_CONNECTION_FACTORY = new SharedConnectionFactory(

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

@@ -33,6 +33,8 @@
 
 import java.sql.SQLException;
 import java.time.Duration;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -115,21 +117,24 @@ public void testCreate() {
             .build())
         .create();
 
-    // Expect publisher to emit 1 connection
-    AtomicInteger counter = new AtomicInteger(0);
+    // Expect publisher to emit one connection to each subscriber
+    Set<Connection> connections = new HashSet<>();
     Flux.from(connectionPublisher)
-      .doOnNext(connection -> counter.incrementAndGet())
+      .doOnNext(connections::add)
       .doOnNext(connection -> Mono.from(connection.close()).subscribe())
       .blockLast(DatabaseConfig.connectTimeout());
-    assertEquals(1, counter.get());
+    assertEquals(1, connections.size());
+    Flux.from(connectionPublisher)
+      .doOnNext(connections::add)
+      .doOnNext(connection -> Mono.from(connection.close()).subscribe())
+      .blockLast(DatabaseConfig.connectTimeout());
+    assertEquals(2, connections.size());
+    Flux.from(connectionPublisher)
+      .doOnNext(connections::add)
+      .doOnNext(connection -> Mono.from(connection.close()).subscribe())
+      .blockLast(DatabaseConfig.connectTimeout());
+    assertEquals(3, connections.size());
 
-    // Expect publisher to reject multiple subscribers
-    try {
-      Mono.from(connectionPublisher)
-        .block(Duration.ofSeconds(1));
-      fail("Connection publisher did not reject multiple subscribers");
-    }
-    catch (IllegalStateException expected) { }
   }
 
   /**

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

@@ -559,7 +559,7 @@ public void testSetAutoCommit() {
             selectInSessionB);
         }
         finally {
-          awaitNone(sessionA.close());
+          awaitNone(sessionB.close());
         }
       }
       finally {