0.9.0.RC1 SPI Update

pom.xml

@@ -66,7 +66,7 @@
   <properties>
     <java.version>11</java.version>
     <ojdbc.version>21.3.0.0</ojdbc.version>
-    <r2dbc.version>0.9.0.M2</r2dbc.version>
+    <r2dbc.version>0.9.0.RC1</r2dbc.version>
     <reactor.version>3.3.0.RELEASE</reactor.version>
     <reactive-streams.version>1.0.3</reactive-streams.version>
     <junit.version>5.7.0</junit.version>

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

@@ -30,6 +30,7 @@
 import io.r2dbc.spi.R2dbcTimeoutException;
 import io.r2dbc.spi.R2dbcTransientException;
 import io.r2dbc.spi.R2dbcTransientResourceException;
+import oracle.jdbc.OracleDatabaseException;
 
 import java.sql.SQLException;
 import java.sql.SQLIntegrityConstraintViolationException;
@@ -166,41 +167,42 @@ static R2dbcException toR2dbcException(SQLException sqlException) {
     final String message = sqlException.getMessage();
     final String sqlState = sqlException.getSQLState();
     final int errorCode = sqlException.getErrorCode();
+    final String sql = getSql(sqlException);
 
     if (sqlException instanceof SQLNonTransientException) {
       if (sqlException instanceof SQLSyntaxErrorException) {
         return new R2dbcBadGrammarException(
-          message, sqlState, errorCode, sqlException);
+          message, sqlState, errorCode, sql, sqlException);
       }
       else if (sqlException instanceof SQLIntegrityConstraintViolationException) {
         return new R2dbcDataIntegrityViolationException(
-          message, sqlState, errorCode, sqlException);
+          message, sqlState, errorCode, sql, sqlException);
       }
       else if (sqlException instanceof SQLNonTransientConnectionException) {
         return new R2dbcNonTransientResourceException(
-          message, sqlState, errorCode, sqlException);
+          message, sqlState, errorCode, sql, sqlException);
       }
       else {
         return new OracleR2dbcNonTransientException(
-          message, sqlState, errorCode, sqlException);
+          message, sqlState, errorCode, sql, sqlException);
       }
     }
     else if (sqlException instanceof SQLTransientException) {
       if (sqlException instanceof SQLTimeoutException) {
         return new R2dbcTimeoutException(
-          message, sqlState, errorCode, sqlException);
+          message, sqlState, errorCode, sql, sqlException);
       }
       else if (sqlException instanceof SQLTransactionRollbackException) {
         return new R2dbcRollbackException(
-          message, sqlState, errorCode, sqlException);
+          message, sqlState, errorCode, sql, sqlException);
       }
       else if (sqlException instanceof SQLTransientConnectionException) {
         return new R2dbcTransientResourceException(
-          message, sqlState, errorCode, sqlException);
+          message, sqlState, errorCode, sql, sqlException);
       }
       else {
         return new OracleR2dbcTransientException(
-          message, sqlState, errorCode, sqlException);
+          message, sqlState, errorCode, sql, sqlException);
       }
     }
     else if (sqlException instanceof SQLRecoverableException) {
@@ -209,11 +211,11 @@ else if (sqlException instanceof SQLRecoverableException) {
       // the connection is no longer valid. The R2dbcTransientResourceException
       // expresses the same conditions.
       return new R2dbcTransientResourceException(
-        message, sqlState, errorCode, sqlException);
+        message, sqlState, errorCode, sql, sqlException);
     }
     else {
       return new OracleR2dbcException(
-        message, sqlState, errorCode, sqlException);
+        message, sqlState, errorCode, sql, sqlException);
     }
   }
 
@@ -293,8 +295,30 @@ static <T> T fromJdbc(ThrowingSupplier<T> supplier)
    * @return A new non-transient exception.
    */
   static R2dbcNonTransientException newNonTransientException(
-    String message, Throwable cause) {
-    return new OracleR2dbcNonTransientException(message, null, 0, cause);
+    String message, String sql, Throwable cause) {
+    return new OracleR2dbcNonTransientException(message, null, 0, sql, cause);
+  }
+
+  /**
+   * Returns the SQL command that caused a {@code sqlException}, if it is
+   * available. This method is only implemented to support the case where the
+   * exception is caused by a {@link oracle.jdbc.OracleDatabaseException}.
+   * @param sqlException Exception to extract SQL from. Not null.
+   * @return The SQL that caused the {@code sqlException}, of {@code null} if
+   * the SQL is not available.
+   */
+  private static String getSql(SQLException sqlException) {
+    Throwable cause = sqlException.getCause();
+
+    while (cause != null) {
+
+      if (cause instanceof OracleDatabaseException)
+        return ((OracleDatabaseException)cause).getSql();
+
+      cause = cause.getCause();
+    }
+
+    return null;
   }
 
   /**
@@ -368,17 +392,17 @@ default T get() throws R2dbcException {
    * concrete subclass must be defined. This subclass does not implement any
    * behavior that is specific to the Oracle driver.
    * </p><p>
-   * This subclass is defined so that {@link #toR2dbcException(SQLException)}
-   * can throw an instance of {@code R2dbcException} when mapping a
-   * {@link SQLException}.
+   * This subclass is defined so that
+   * {@link #toR2dbcException(SQLException)} can throw an instance of
+   * {@code R2dbcException} when mapping a {@link SQLException}.
    * </p>
    */
   private static final class OracleR2dbcException
     extends R2dbcException {
     private OracleR2dbcException(
-      String message, String sqlState, int errorCode,
+      String message, String sqlState, int errorCode, String sql,
       SQLException sqlException) {
-      super(message, sqlState, errorCode, sqlException);
+      super(message, sqlState, errorCode, sql, sqlException);
     }
   }
 
@@ -390,17 +414,18 @@ private OracleR2dbcException(
    * This subclass does implement any behavior that is specific to the
    * Oracle driver.
    * </p><p>
-   * This subclass is defined so that {@link #toR2dbcException(SQLException)}
-   * can throw an instance of {@code R2dbcTransientException} when mapping a
+   * This subclass is defined so that
+   * {@link #toR2dbcException(SQLException)} can throw an instance of
+   * {@code R2dbcTransientException} when mapping a
    * {@link SQLTransientException}.
    * </p>
    */
   private static final class OracleR2dbcTransientException
     extends R2dbcTransientException {
     private OracleR2dbcTransientException(
-      String message, String sqlState, int errorCode,
+      String message, String sqlState, int errorCode, String sql,
       SQLException sqlException) {
-      super(message, sqlState, errorCode, sqlException);
+      super(message, sqlState, errorCode, sql, sqlException);
     }
   }
 
@@ -420,9 +445,9 @@ private OracleR2dbcTransientException(
   private static final class OracleR2dbcNonTransientException
     extends R2dbcNonTransientException {
     private OracleR2dbcNonTransientException(
-      String message, String sqlState, int errorCode,
+      String message, String sqlState, int errorCode, String sql,
       Throwable cause) {
-      super(message, sqlState, errorCode, cause);
+      super(message, sqlState, errorCode, sql, cause);
     }
   }
 

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

@@ -194,7 +194,7 @@ final class OracleReactiveJdbcAdapter implements ReactiveJdbcAdapter {
       // may need to be disabled when connecting to an 18.x database. Starting
       // in 19.x, the database can detect when it's running on a system where
       // OOB is not supported and automatically disable OOB. This automated 
-      // detection is not impleneted in 18.x.
+      // detection is not implemented in 18.x.
       OracleR2dbcOptions.DISABLE_OUT_OF_BAND_BREAK,
 
       // Allow the client-side ResultSet cache to be disabled. It is
@@ -247,7 +247,7 @@ static OracleReactiveJdbcAdapter getInstance() {
    * When the "descriptor" option is provided, it is invalid to specify any
    * other options that might conflict with values also specified in the
    * descriptor. For instance, the descriptor element of
-   * {@code (ADDRESSS=(HOST=...)(PORT=...)(PROTOCOL=...))} specifies values
+   * {@code (ADDRESS=(HOST=...)(PORT=...)(PROTOCOL=...))} specifies values
    * that overlap with the standard {@code Option}s of {@code HOST}, {@code
    * PORT}, and {@code SSL}. An {@code IllegalArgumentException} is thrown
    * when the descriptor is provided with any overlapping {@code Option}s.
@@ -1136,7 +1136,7 @@ private <T> T castAsType(Object object, Class<T> type) {
     }
     else {
       throw OracleR2dbcExceptions.newNonTransientException(
-        object.getClass() + " is not an instance of " + type, null);
+        object.getClass() + " is not an instance of " + type, null, null);
     }
   }
 
@@ -1468,9 +1468,8 @@ private static final class AsyncLock {
       new ConcurrentLinkedDeque<>();
 
     /**
-     * Returns a {@code Publisher} that emits {@code onComplete} when
-     * this lock is acquired.
-     * @return
+     * Executes a {@code callback} with exclusive access to the guarded
+     * resource.
      */
     void lock(Runnable callback) {
       assert waitCount.get() >= 0 : "Wait count is less than 0: " + waitCount;

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

@@ -39,6 +39,7 @@
 import java.nio.ByteBuffer;
 import java.sql.Timestamp;
 import java.time.LocalDateTime;
+import java.util.NoSuchElementException;
 
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.requireNonNull;
 
@@ -121,8 +122,6 @@ static OutParameters createOutParameters(
    * into the specified {@code type}.
    * </p>
    * @throws IllegalArgumentException {@inheritDoc}
-   * @throws IllegalArgumentException If the {@code index} is less than 0,
-   * or greater than the maximum value index.
    * @throws IllegalArgumentException If conversion to the specified
    * {@code type} is not supported.
    */
@@ -163,14 +162,14 @@ public <T> T get(String name, Class<T> type) {
    * matching values.
    * @param name The name of a value
    * @return The index of the named value within this {@code Readable}
-   * @throws IllegalArgumentException If no column has a matching name.
+   * @throws NoSuchElementException If no column has a matching name.
    */
   private int indexOf(String name) {
     int columnIndex = readablesMetadata.getColumnIndex(name);
     if (columnIndex != -1)
       return columnIndex;
     else
-      throw new IllegalArgumentException("Unrecognized name: " + name);
+      throw new NoSuchElementException("Unrecognized name: " + name);
   }
 
   /**
@@ -206,7 +205,13 @@ else if (LocalDateTime.class.equals(type)) {
       value = getLocalDateTime(index);
     }
     else if (Object.class.equals(type)) {
-      value = convert(index, readablesMetadata.get(index).getJavaType());
+      // Use the default type mapping if Object.class has been specified.
+      // This method is invoked recursively with the default mapping, so long
+      // as Object.class is not also the default mapping.
+      Class<?> defaultType = readablesMetadata.get(index).getJavaType();
+      value = Object.class.equals(defaultType)
+        ? jdbcReadable.getObject(index, Object.class)
+        : convert(index, defaultType);
     }
     else {
       value = jdbcReadable.getObject(index, type);
@@ -323,14 +328,14 @@ private LocalDateTime getLocalDateTime(int index) {
    * for this row. This method is used to verify index value parameters
    * supplied by user code.
    * @param index 0-based column index
-   * @throws IllegalStateException if the index is not valid.
+   * @throws IndexOutOfBoundsException if the index is not valid.
    */
   private void requireValidIndex(int index) {
     if (index < 0) {
-      throw new IllegalArgumentException("Index is less than zero: " + index);
+      throw new IndexOutOfBoundsException("Index is less than zero: " + index);
     }
     else if (index >= readablesMetadata.getList().size()) {
-      throw new IllegalArgumentException(
+      throw new IndexOutOfBoundsException(
         "Index " + index + " is greater than or equal to column count: "
           + readablesMetadata.getList().size());
     }

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

@@ -296,8 +296,16 @@ static ColumnMetadata createColumnMetadata(
     Nullability nullability = getNullability(fromJdbc(() ->
       resultSetMetaData.isNullable(jdbcIndex)));
 
-    if (type == R2dbcType.BLOB || type == R2dbcType.CLOB
-      || type == R2dbcType.NCLOB) {
+    if (type == R2dbcType.NUMERIC) {
+      // For NUMBER, allow the scale to be 0
+      return new OracleColumnMetadataImpl(type, name, nullability,
+        fromJdbc(() -> resultSetMetaData.getPrecision(jdbcIndex)),
+        fromJdbc(() -> resultSetMetaData.getScale(jdbcIndex)));
+    }
+    if (type == R2dbcType.BLOB
+      || type == R2dbcType.CLOB
+      || type == R2dbcType.NCLOB
+      || type == OracleR2dbcTypes.JSON) {
       // For LOB types, use null as the precision. The actual maximum length
       // is (4GB x database-block-size), which can not be stored as an Integer
       return new OracleColumnMetadataImpl(type, name, nullability, null, null);
@@ -320,7 +328,7 @@ else if (type == R2dbcType.TIMESTAMP
     else if (type == R2dbcType.TIMESTAMP_WITH_TIME_ZONE) {
       // For the TIMESTAMP WITH TIMEZONE types, use the length of
       // OffsetDateTime.toString() as the precision. Use the scale from JDBC,
-      // even if it's 0 because a  TIMESTAMP may 0 decimal digits.
+      // even if it's 0 because a TIMESTAMP may have 0 decimal digits.
       return new OracleColumnMetadataImpl(type, name, nullability,
         OFFSET_DATE_TIME_PRECISION,
         fromJdbc(() -> resultSetMetaData.getScale(jdbcIndex)));

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

@@ -586,9 +586,18 @@ <T> Publisher<T> publishSegments(Function<Segment, T> mappingFunction) {
    */
   private static final class WarningResult extends OracleResultImpl {
 
+    /** The warning of this result */
     private final SQLWarning warning;
+
+    /** The result that follows this result */
     private final OracleResultImpl result;
 
+    /**
+     * Constructs a result that publishes a {@code warning} as a
+     * {@link Message}, and then publishes the segments of a {@code result}.
+     * @param warning Warning to publish
+     * @param result Result of segments to publish after the warning
+     */
     private WarningResult(SQLWarning warning, OracleResultImpl result) {
       this.warning = warning;
       this.result = result;