Updating ojdbc11 and README

Author: Michael-A-McMahon

README.md

@@ -26,7 +26,7 @@
 
   <groupId>com.oracle.database.r2dbc</groupId>
   <artifactId>oracle-r2dbc</artifactId>
-  <version>1.2.0</version>
+  <version>1.3.0</version>
   <name>oracle-r2dbc</name>
   <description>
     Oracle R2DBC Driver implementing version 1.0.0 of the R2DBC SPI for Oracle Database.
@@ -65,9 +65,9 @@
 
   <properties>
     <java.version>11</java.version>
-    <ojdbc.version>23.5.0.24.07</ojdbc.version>
+    <ojdbc.version>23.6.0.24.10</ojdbc.version>
     <r2dbc.version>1.0.0.RELEASE</r2dbc.version>
-    <reactor.version>3.5.11</reactor.version>
+    <reactor.version>3.6.11</reactor.version>
     <reactive-streams.version>1.0.3</reactive-streams.version>
     <junit.version>5.9.1</junit.version>
     <spring-jdbc.version>5.3.19</spring-jdbc.version>

pom.xml

@@ -38,17 +38,17 @@
 import java.sql.SQLException;
 import java.sql.Savepoint;
 import java.time.Duration;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 import static io.r2dbc.spi.IsolationLevel.READ_COMMITTED;
 import static io.r2dbc.spi.IsolationLevel.SERIALIZABLE;
 import static io.r2dbc.spi.TransactionDefinition.ISOLATION_LEVEL;
 import static io.r2dbc.spi.TransactionDefinition.LOCK_WAIT_TIMEOUT;
 import static io.r2dbc.spi.TransactionDefinition.NAME;
 import static io.r2dbc.spi.TransactionDefinition.READ_ONLY;
-import static oracle.r2dbc.impl.OracleR2dbcExceptions.requireNonNull;
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.fromJdbc;
+import static oracle.r2dbc.impl.OracleR2dbcExceptions.requireNonNull;
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.requireOpenConnection;
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.runJdbc;
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.toR2dbcException;
@@ -129,10 +129,10 @@ final class OracleConnectionImpl implements Connection, Lifecycle {
   private TransactionDefinition currentTransaction = null;
 
   /**
-   * A queue of tasks that must complete before the {@link #jdbcConnection} is
-   * closed.
+   * A set of tasks that must complete before the {@link #jdbcConnection} is
+   * closed. The tasks are executed by subscribing to a Publisher.
    */
-  private final Queue<Publisher<?>> closeTasks = new ConcurrentLinkedQueue<>();
+  private final Set<Publisher<?>> closeTasks = ConcurrentHashMap.newKeySet();
 
   /**
    * Constructs a new connection that uses the specified {@code adapter} to
@@ -393,30 +393,36 @@ public Publisher<Void> close() {
   /**
    * <p>
    * Adds a publisher that must be subscribed to and must terminate before
-   * closing the JDBC connection. This can be used to ensure that a publisher
-   * has completed a task before the {@link #jdbcConnection()} has been closed
-   * and becomes unusable.
+   * closing the JDBC connection. This method can be used to ensure that certain
+   * tasks are completed before the {@link #jdbcConnection()} is closed and
+   * becomes unusable.
    * </p><p>
-   * <i> A call to this method should always be accompanied with a call to
-   * {@link #removeCloseTask(Publisher)}</i>: If the publisher is subscribed to
-   * and it terminates before {@link #close()} is called, then any reference
-   * to the Publisher must be removed so that it can be garbage collected.
+   * The publisher returned by this method emits the same result as the
+   * publisher passed into this method. However, when the returned publisher
+   * terminates, it will also remove any reference to the publisher that was
+   * passed into this method. <i>If the returned publisher is never subscribed
+   * to, then the reference will not be cleared until the connection is
+   * closed!</i> So, this method should only be used in cases where the user is
+   * responsible for subscribing to the returned publisher, in the same way they
+   * would be responsible for calling close() on an AutoCloseable. If the user
+   * is not responsible for subscribing, then there is no reasonable way for
+   * them to reduce the number object references that this method will create;
+   * They would either need to close this connection, or subscribe to a
+   * publisher when there is no obligation to do so.
    * </p>
-   * @param publisher Publisher that must terminate before closing the JDBC
-   * connection. Not null.
+   *
+   * @param publisher Publisher that must be subscribed to before closing the
+   * JDBC connection. Not null.
+   *
+   * @return A publisher that emits the same result as the publisher passed into
+   * this method, and clears any reference to it when terminated. Not null.
    */
-  void addCloseTask(Publisher<?> publisher) {
+  <T> Publisher<T> addCloseTask(Publisher<T> publisher) {
     closeTasks.add(publisher);
-  }
 
-  /**
-   * Removes a publisher that was previously added with
-   * {@link #addCloseTask(Publisher)}.
-   *
-   * @param publisher Publisher to remove. Not null.
-   */
-  void removeCloseTask(Publisher<?> publisher) {
-    closeTasks.remove(publisher);
+    return Publishers.concatTerminal(
+      publisher,
+      Mono.fromRunnable(() -> closeTasks.remove(publisher)));
   }
 
   /**
@@ -876,7 +882,7 @@ public Publisher<Void> preRelease() {
    * Returns the JDBC connection that this R2DBC connection executes database
    * calls with.
    *
-   * @return The JDBC connection that backs this R2DBC connection. Not null.
+   * @return The JDBC connection which backs this R2DBC connection. Not null.
    */
   java.sql.Connection jdbcConnection() {
     return jdbcConnection;

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

@@ -46,6 +46,7 @@
 import oracle.sql.INTERVALYM;
 import oracle.sql.TIMESTAMPLTZ;
 import oracle.sql.TIMESTAMPTZ;
+import org.reactivestreams.Publisher;
 
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
@@ -93,6 +94,9 @@
 class OracleReadableImpl implements io.r2dbc.spi.Readable {
 
 
+  /** The R2DBC connection that created this readable */
+  private final OracleConnectionImpl r2dbcConnection;
+
   /** The JDBC connection that created this readable */
   private final java.sql.Connection jdbcConnection;
 
@@ -117,21 +121,21 @@ class OracleReadableImpl implements io.r2dbc.spi.Readable {
    * {@code jdbcReadable} and obtains metadata of the values from
    * {@code resultMetadata}.
    * </p>
-   * @param jdbcConnection JDBC connection that created the
+   * @param r2dbcConnection R2DBC connection that created the
    *   {@code jdbcReadable}. Not null.
    * @param jdbcReadable Readable values from a JDBC Driver. Not null.
    * @param readablesMetadata Metadata of each value. Not null.
    * @param adapter Adapts JDBC calls into reactive streams. Not null.
    */
   private OracleReadableImpl(
-    java.sql.Connection jdbcConnection,  DependentCounter dependentCounter,
-    JdbcReadable jdbcReadable, ReadablesMetadata<?> readablesMetadata,
-    ReactiveJdbcAdapter adapter) {
-    this.jdbcConnection = jdbcConnection;
+    OracleConnectionImpl r2dbcConnection, DependentCounter dependentCounter,
+    JdbcReadable jdbcReadable, ReadablesMetadata<?> readablesMetadata) {
+    this.r2dbcConnection = r2dbcConnection;
+    this.jdbcConnection = r2dbcConnection.jdbcConnection();
     this.dependentCounter = dependentCounter;
     this.jdbcReadable = jdbcReadable;
     this.readablesMetadata = readablesMetadata;
-    this.adapter = adapter;
+    this.adapter = r2dbcConnection.adapter();
   }
 
   /**
@@ -151,11 +155,10 @@ private OracleReadableImpl(
    *   {@code metadata}. Not null.
    */
   static Row createRow(
-    java.sql.Connection jdbcConnection, DependentCounter dependentCounter,
-    JdbcReadable jdbcReadable, RowMetadataImpl metadata,
-    ReactiveJdbcAdapter adapter) {
+    OracleConnectionImpl r2dbcConnection, DependentCounter dependentCounter,
+    JdbcReadable jdbcReadable, RowMetadataImpl metadata) {
     return new RowImpl(
-      jdbcConnection, dependentCounter, jdbcReadable, metadata, adapter);
+      r2dbcConnection, dependentCounter, jdbcReadable, metadata);
   }
 
   /**
@@ -164,7 +167,7 @@ static Row createRow(
    * the provided {@code jdbcReadable} and {@code rowMetadata}. The metadata
    * object is used to determine the default type mapping of column values.
    * </p>
-   * @param jdbcConnection JDBC connection that created the
+   * @param r2dbcConnection R2DBC connection that created the
    *   {@code jdbcReadable}. Not null.
    * @param dependentCounter Counter that is increased for each dependent
    * {@code Result} created by the returned {@code OutParameters}
@@ -175,11 +178,10 @@ static Row createRow(
    *   {@code metadata}. Not null.
    */
   static OutParameters createOutParameters(
-    java.sql.Connection jdbcConnection, DependentCounter dependentCounter,
-    JdbcReadable jdbcReadable, OutParametersMetadataImpl metadata,
-    ReactiveJdbcAdapter adapter) {
+    OracleConnectionImpl r2dbcConnection, DependentCounter dependentCounter,
+    JdbcReadable jdbcReadable, OutParametersMetadataImpl metadata) {
     return new OutParametersImpl(
-      jdbcConnection, dependentCounter, jdbcReadable, metadata, adapter);
+      r2dbcConnection, dependentCounter, jdbcReadable, metadata);
   }
 
   /**
@@ -335,11 +337,16 @@ private ByteBuffer getByteBuffer(int index) {
    */
   private Blob getBlob(int index) {
     java.sql.Blob jdbcBlob = jdbcReadable.getObject(index, java.sql.Blob.class);
-    return jdbcBlob == null
-      ? null
-      : OracleLargeObjects.createBlob(
-          adapter.publishBlobRead(jdbcBlob),
-          adapter.publishBlobFree(jdbcBlob));
+
+    if (jdbcBlob == null)
+      return null;
+
+    Publisher<Void> freePublisher =
+      r2dbcConnection.addCloseTask(adapter.publishBlobFree(jdbcBlob));
+
+    return OracleLargeObjects.createBlob(
+      adapter.publishBlobRead(jdbcBlob),
+      freePublisher);
   }
 
   /**
@@ -367,11 +374,15 @@ private Clob getClob(int index) {
       jdbcClob = jdbcReadable.getObject(index, java.sql.Clob.class);
     }
 
-    return jdbcClob == null
-      ? null
-      : OracleLargeObjects.createClob(
-          adapter.publishClobRead(jdbcClob),
-          adapter.publishClobFree(jdbcClob));
+    if (jdbcClob == null)
+      return null;
+
+    Publisher<Void> freePublisher =
+      r2dbcConnection.addCloseTask(adapter.publishClobFree(jdbcClob));
+
+    return OracleLargeObjects.createClob(
+      adapter.publishClobRead(jdbcClob),
+      freePublisher);
   }
 
   /**
@@ -685,11 +696,10 @@ private OracleR2dbcObjectImpl getOracleR2dbcObject(int index) {
       return null;
 
     return new OracleR2dbcObjectImpl(
-      jdbcConnection,
+      r2dbcConnection,
       dependentCounter,
       new StructJdbcReadable(oracleStruct),
-      ReadablesMetadata.createAttributeMetadata(oracleStruct),
-      adapter);
+      ReadablesMetadata.createAttributeMetadata(oracleStruct));
   }
 
   /**
@@ -956,7 +966,7 @@ private Result getResult(int index) {
 
     dependentCounter.increment();
     return OracleResultImpl.createQueryResult(
-      dependentCounter, resultSet, adapter);
+      r2dbcConnection, dependentCounter, resultSet);
   }
 
   /**
@@ -994,17 +1004,16 @@ private static final class RowImpl
      * {@code jdbcReadable}, and uses the specified {@code rowMetadata} to
      * determine the default type mapping of column values.
      * </p>
-     * @param jdbcConnection JDBC connection that created the
+     * @param r2dbcConnection R2DBC connection that created the
      *   {@code jdbcReadable}. Not null.
      * @param jdbcReadable Row data from the Oracle JDBC Driver. Not null.
      * @param metadata Meta-data for the specified row. Not null.
      * @param adapter Adapts JDBC calls into reactive streams. Not null.
      */
     private RowImpl(
-      java.sql.Connection jdbcConnection, DependentCounter dependentCounter,
-      JdbcReadable jdbcReadable, RowMetadataImpl metadata,
-      ReactiveJdbcAdapter adapter) {
-      super(jdbcConnection, dependentCounter, jdbcReadable, metadata, adapter);
+      OracleConnectionImpl r2dbcConnection, DependentCounter dependentCounter,
+      JdbcReadable jdbcReadable, RowMetadataImpl metadata) {
+      super(r2dbcConnection, dependentCounter, jdbcReadable, metadata);
       this.metadata = metadata;
     }
 
@@ -1044,10 +1053,9 @@ private static final class OutParametersImpl
      * @param adapter Adapts JDBC calls into reactive streams. Not null.
      */
     private OutParametersImpl(
-      java.sql.Connection jdbcConnection, DependentCounter dependentCounter,
-      JdbcReadable jdbcReadable, OutParametersMetadataImpl metadata,
-      ReactiveJdbcAdapter adapter) {
-      super(jdbcConnection, dependentCounter, jdbcReadable, metadata, adapter);
+      OracleConnectionImpl r2dbcConnection, DependentCounter dependentCounter,
+      JdbcReadable jdbcReadable, OutParametersMetadataImpl metadata) {
+      super(r2dbcConnection, dependentCounter, jdbcReadable, metadata);
       this.metadata = metadata;
     }
 
@@ -1068,20 +1076,18 @@ private final class OracleR2dbcObjectImpl
      * {@code jdbcReadable} and obtains metadata of the values from
      * {@code outParametersMetaData}.
      * </p>
-     * @param jdbcConnection JDBC connection that created the
+     * @param r2dbcConnection R2DBC connection that created the
      *   {@code jdbcReadable}. Not null.
      * @param structJdbcReadable Readable values from a JDBC Driver. Not null.
      * @param metadata Metadata of each value. Not null.
      * @param adapter Adapts JDBC calls into reactive streams. Not null.
      */
     private OracleR2dbcObjectImpl(
-      java.sql.Connection jdbcConnection,
+      OracleConnectionImpl r2dbcConnection,
       DependentCounter dependentCounter,
       StructJdbcReadable structJdbcReadable,
-      OracleR2dbcObjectMetadataImpl metadata,
-      ReactiveJdbcAdapter adapter) {
-      super(
-        jdbcConnection, dependentCounter, structJdbcReadable, metadata, adapter);
+      OracleR2dbcObjectMetadataImpl metadata) {
+      super(r2dbcConnection, dependentCounter, structJdbcReadable, metadata);
       this.metadata = metadata;
     }