Add Segment type for warnings

Author: Michael-A-McMahon

src/main/java/oracle/r2dbc/OracleR2dbcWarning.java

@@ -3,85 +3,79 @@
 import io.r2dbc.spi.R2dbcException;
 import io.r2dbc.spi.Result;
 
-import java.sql.SQLWarning;
 import java.util.function.Function;
 import java.util.function.Predicate;
 
-import static io.r2dbc.spi.Result.*;
-
 /**
  * <p>
- * An exception that provides information on warnings raised by Oracle Database.
+ * A subtype of the {@link Result.Segment} interface that provides information
+ * on warnings raised by Oracle Database.
+ * </p><p>
+ * When a SQL command results in a warning, Oracle R2DBC emits a {@link Result}
+ * with an {@code OracleR2dbcWarning} segment in addition to any other segments
+ * that resulted from the SQL command. For example, if a SQL {@code SELECT}
+ * command results in a warning, then an {@code OracleR2dbcWarning} segment is
+ * included with the result, along with any {@link Result.RowSegment}s returned
+ * by the {@code SELECT}.
  * </p><p>
- * When a SQL command results in a warning, Oracle R2DBC generates
- * {@link Message} segments having an {@code OracleR2dbcWarning} as the
- * {@linkplain Message#exception() exception}. These segments may be consumed
- * with {@link Result#flatMap(Function)}:
+ * R2DBC drivers typically emit {@code onError} signals for {@code Message}
+ * segments that are not consumed by {@link Result#filter(Predicate)} or
+ * {@link Result#flatMap(Function)}. Oracle R2DBC does not apply this behavior
+ * for warning messages. If an {@code OracleR2dbcWarning}
+ * segment is not consumed by the {@code filter} or {@code flatMap} methods of
+ * a {@code Result}, then the warning is discarded and the result may be
+ * consumed as normal with with the {@code map} or {@code getRowsUpdated}
+ * methods.
+ * </p><p>
+ * Warning messages may be consumed with {@link Result#flatMap(Function)}:
  * </p><pre>{@code
  * result.flatMap(segment -> {
- *   if (OracleR2dbcWarning.isWarning(segment)) {
- *     logWarning(OracleR2dbcWarning.getWarning(segment));
+ *   if (segment instanceof OracleR2dbcWarning) {
+ *     logWarning(((OracleR2dbcWarning)segment).getMessage());
  *     return emptyPublisher();
  *   }
  *   else {
  *     ... handle other segment types ...
  *   }
  * })
  * }</pre><p>
- * Alternatively, these segments may be ignored using
- * {@link Result#filter(Predicate)}:
- * </p> <pre>{@code
- * result.filter(segment -> !OracleR2dbcWarning.isWarning(segment))
+ * A {@code flatMap} function may also be used to convert a warning into an
+ * {@code onError} signal:
+ * </p><pre>{@code
+ * result.flatMap(segment -> {
+ *   if (segment instanceof OracleR2dbcWarning) {
+ *     return errorPublisher(((OracleR2dbcWarning)segment).warning());
+ *   }
+ *   else {
+ *     ... handle other segment types ...
+ *   }
+ * })
  * }</pre>
- * If these segments are not flat-mapped or filtered from a {@code Result}, then
- * an {@code onError} signal with the {@code OracleR2dbcWarning} is emitted 
- * </p>
+ * @since 1.1.0
  */
-public class OracleR2dbcWarning extends R2dbcException {
+public interface OracleR2dbcWarning extends Result.Segment {
 
   /**
-   * Constructs a new warning having a {@code SQLWarning} from JDBC as its
-   * cause. The constructed warning has the same message, SQL state, and error
-   * code as the given {@code SQLWarning}.
-   * @param sql The SQL command that resulted in a warning. May be null.
-   * @param sqlWarning The SQLWarning thrown by JDBC. Not null.
+   * Returns the warning as an {@link R2dbcException}.
+   * @return The warning as an {@link R2dbcException}. Not null.
    */
-  public OracleR2dbcWarning(String sql, SQLWarning sqlWarning) {
-    super(sqlWarning.getMessage(), sqlWarning.getSQLState(),
-      sqlWarning.getErrorCode(), sql, sqlWarning);
-  }
+  R2dbcException exception();
 
   /**
-   * Checks if a segment is a {@link Message} having an
-   * {@code OracleR2dbcException} as an
-   * {@linkplain Message#exception() exception}.
-   * @param segment Segment to check. May be null, in which case {@code false}
-   * is returned.
-   * @return {@code true} if the segment is a {@code Message} with an
-   * {@code OracleR2dbcException}, or {@code false} if not.
+   * Returns the error code of the warning.
+   * @return The error code of the warning.
    */
-  public static boolean isWarning(Segment segment) {
-    return segment instanceof Message
-      && ((Message)segment).exception() instanceof OracleR2dbcWarning;
-  }
+  int errorCode();
 
   /**
-   * Returns the {@code OracleR2dbcWarning} of a {@link Message} segment. This
-   * method should only be called if {@link #isWarning(Segment)} returned
-   * {@code true} for the given segment.
-   * @param segment A {@code Message} segment with an
-   * {@code OracleR2dbcWarning}. Not null.
-   * @return The {@code OracleR2dbcWarning} of the {@code Message} segment.
-   * @throws IllegalArgumentException If the segment is not a {@code Message}
-   * with an {@code OracleR2dbcWarning}.
+   * Returns the SQLState of the warning.
+   * @return The SQLState of the warning. Not null.
    */
-  public static OracleR2dbcWarning getWarning(Segment segment) {
-    if (!isWarning(segment)) {
-      throw new IllegalArgumentException(
-        "Not a Message segment with an OracleR2dbcWarning: " + segment);
-    }
-
-    return (OracleR2dbcWarning)((Message)segment).exception();
-  }
+  String sqlState();
 
+  /**
+   * Returns the text of the warning message.
+   * @return The text of the warning message. Not null.
+   */
+  String message();
 }

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

@@ -31,7 +31,6 @@
 import io.r2dbc.spi.R2dbcTransientException;
 import io.r2dbc.spi.R2dbcTransientResourceException;
 import oracle.jdbc.OracleDatabaseException;
-import oracle.r2dbc.OracleR2dbcWarning;
 
 import java.sql.SQLException;
 import java.sql.SQLIntegrityConstraintViolationException;
@@ -43,7 +42,6 @@
 import java.sql.SQLTransactionRollbackException;
 import java.sql.SQLTransientConnectionException;
 import java.sql.SQLTransientException;
-import java.sql.SQLWarning;
 import java.util.function.Supplier;
 
 /**
@@ -165,12 +163,26 @@ static void requireOpenConnection(java.sql.Connection jdbcConnection) {
    * as the specified {@code sqlException}. Not null.
    */
   static R2dbcException toR2dbcException(SQLException sqlException) {
+    return toR2dbcException(sqlException, getSql(sqlException));
+  }
+
+  /**
+   * Converts a {@link SQLException} into an {@link R2dbcException}, as
+   * specified by {@link #toR2dbcException(SQLException)}. This method accepts
+   * a SQL string argument. It should be used in cases where the SQL can not
+   * be extracted by {@link #getSql(SQLException)}.
+   * @param sqlException A {@code SQLException} to convert. Not null.
+   * @param sql SQL that caused the exception
+   * @return an {@code R2dbcException} that indicates the same error conditions
+   * as the specified {@code sqlException}. Not null.
+   */
+  static R2dbcException toR2dbcException(
+    SQLException sqlException, String sql) {
     assert sqlException != null : "sqlException is null";
 
     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) {
@@ -216,9 +228,6 @@ else if (sqlException instanceof SQLRecoverableException) {
       return new R2dbcTransientResourceException(
         message, sqlState, errorCode, sql, sqlException);
     }
-    else if (sqlException instanceof SQLWarning) {
-      return new OracleR2dbcWarning(sql, (SQLWarning)sqlException);
-    }
     else {
       return new OracleR2dbcException(
         message, sqlState, errorCode, sql, sqlException);

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

@@ -27,6 +27,7 @@
 import io.r2dbc.spi.Result;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
+import oracle.r2dbc.OracleR2dbcWarning;
 import oracle.r2dbc.impl.ReadablesMetadata.RowMetadataImpl;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -398,14 +399,15 @@ static OracleResultImpl createErrorResult(R2dbcException r2dbcException) {
    * Creates a {@code Result} that publishes a {@code warning} as a
    * {@link Message} segment, followed by any {@code Segment}s of a
    * {@code result}.
+   * @param sql The SQL that resulted in a waring. Not null.
    * @param warning Warning to publish. Not null.
    * @param result Result to publisher. Not null.
    * @return A {@code Result} for a {@code Statement} execution that
    * completed with a warning.
    */
   static OracleResultImpl createWarningResult(
-    SQLWarning warning, OracleResultImpl result) {
-    return new WarningResult(warning, result);
+    String sql, SQLWarning warning, OracleResultImpl result) {
+    return new WarningResult(sql, warning, result);
   }
 
   /**
@@ -627,6 +629,9 @@ <T> Publisher<T> publishSegments(Function<Segment, T> mappingFunction) {
    */
   private static final class WarningResult extends OracleResultImpl {
 
+    /** The SQL that resulted in a warning */
+    private final String sql;
+
     /** The warning of this result */
     private final SQLWarning warning;
 
@@ -636,11 +641,13 @@ private static final class WarningResult extends OracleResultImpl {
     /**
      * Constructs a result that publishes a {@code warning} as a
      * {@link Message}, and then publishes the segments of a {@code result}.
+     * @param sql The SQL that resulted in a warning
      * @param warning Warning to publish. Not null.
      * @param result Result of segments to publish after the warning. Not null.
      */
     private WarningResult(
-      SQLWarning warning, OracleResultImpl result) {
+      String sql, SQLWarning warning, OracleResultImpl result) {
+      this.sql = sql;
       this.warning = warning;
       this.result = result;
     }
@@ -649,8 +656,11 @@ private WarningResult(
     <T> Publisher<T> publishSegments(Function<Segment, T> mappingFunction) {
       return Flux.fromStream(Stream.iterate(
         warning, Objects::nonNull, SQLWarning::getNextWarning)
-        .map(OracleR2dbcExceptions::toR2dbcException)
-        .map(MessageImpl::new))
+        .map(nextWarning ->
+          // It is noted that SQL can not be extracted from Oracle JDBC's
+          // SQLWarning objects, so it must be explicitly provided here.
+          OracleR2dbcExceptions.toR2dbcException(warning, sql))
+        .map(WarningImpl::new))
         .map(mappingFunction)
         // Invoke publishSegments(Class, Function) rather than
         // publishSegments(Function) to update the state of the result; Namely,
@@ -803,6 +813,38 @@ public String message() {
     }
   }
 
+  /**
+   * Implementation of {@link OracleR2dbcWarning}.
+   */
+  private static final class WarningImpl implements OracleR2dbcWarning {
+
+    private final R2dbcException warning;
+
+    private WarningImpl(R2dbcException exception) {
+      this.warning = exception;
+    }
+
+    @Override
+    public R2dbcException exception() {
+      return warning;
+    }
+
+    @Override
+    public int errorCode() {
+      return warning.getErrorCode();
+    }
+
+    @Override
+    public String sqlState() {
+      return warning.getSqlState();
+    }
+
+    @Override
+    public String message() {
+      return warning.getMessage();
+    }
+  }
+
   /**
    * Returns a {@code Publisher} that emits the signals of a {@code publisher}
    * to a single {@link org.reactivestreams.Subscriber}, and rejects additional

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

@@ -1132,7 +1132,7 @@ private OracleResultImpl getWarnings(OracleResultImpl result) {
         preparedStatement.clearWarnings();
         return warning == null
           ? result
-          : OracleResultImpl.createWarningResult(warning, result);
+          : OracleResultImpl.createWarningResult(sql, warning, result);
       });
     }
 

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

@@ -22,6 +22,7 @@
 package oracle.r2dbc.impl;
 
 import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.R2dbcException;
 import io.r2dbc.spi.Result;
 import io.r2dbc.spi.Result.Message;
 import io.r2dbc.spi.Result.RowSegment;
@@ -638,7 +639,7 @@ public void testOracleR2dbcWarning() {
     Connection connection = awaitOne(sharedConnection());
     try {
 
-      // Expect a warning for forcing a view that references a non-existant
+      // Expect a warning for forcing a view that references a non-existent
       // table
       String sql = "CREATE OR REPLACE FORCE VIEW testOracleR2dbcWarning AS" +
           " SELECT x FROM thisdoesnotexist";
@@ -649,31 +650,29 @@ public void testOracleR2dbcWarning() {
         awaitMany(Flux.from(warningStatement.execute())
           .flatMap(result -> result.flatMap(Mono::just)));
 
-      // Expect the update count segment first. Warnings are always emitted
-      // after any result values.
-      Result.Segment firstSegment = segments.get(0);
-      assertEquals(0,
-        assertInstanceOf(UpdateCount.class, firstSegment).value());
-      assertFalse(OracleR2dbcWarning.isWarning(firstSegment));
-      assertThrows(IllegalArgumentException.class, () ->
-        OracleR2dbcWarning.getWarning(firstSegment));
-
-      // Expect the message segment next. Expect it to have the fixed message
+      // Expect the warning segment first. Expect it to have the fixed message
       // and error number used by Oracle JDBC for all warnings.
-      Result.Segment secondSegment = segments.get(1);
-      Message message = assertInstanceOf(Message.class, secondSegment);
-      assertEquals(
-        message.message(), "Warning: execution completed with warning");
-      assertEquals(message.errorCode(), 17110);
-      assertEquals("", message.sqlState());
+      Result.Segment secondSegment = segments.get(0);
       OracleR2dbcWarning warning =
-        assertInstanceOf(OracleR2dbcWarning.class, message.exception());
-      assertEquals(message.message(), warning.getMessage());
-      assertEquals(message.errorCode(), warning.getErrorCode());
-      assertEquals(message.sqlState(), warning.getSqlState());
-      assertEquals(sql, warning.getSql());
-      assertTrue(OracleR2dbcWarning.isWarning(secondSegment));
-      assertSame(warning, OracleR2dbcWarning.getWarning(secondSegment));
+        assertInstanceOf(OracleR2dbcWarning.class, secondSegment);
+      assertEquals(
+        warning.message(), "Warning: execution completed with warning");
+      assertEquals(warning.errorCode(), 17110);
+      assertEquals("99999", warning.sqlState()); // Default SQL state
+      R2dbcException exception =
+        assertInstanceOf(R2dbcException.class, warning.exception());
+      assertEquals(warning.message(), exception.getMessage());
+      assertEquals(warning.errorCode(), exception.getErrorCode());
+      assertEquals(warning.sqlState(), exception.getSqlState());
+      assertEquals(sql, exception.getSql());
+
+
+      // Expect the update count segment last. Warnings are always emitted
+      // first.
+      Result.Segment firstSegment = segments.get(1);
+      assertEquals(0,
+        assertInstanceOf(UpdateCount.class, firstSegment).value());
+      assertFalse(firstSegment instanceof OracleR2dbcWarning);
 
       // Verify that there are not any more segments
       assertEquals(2, segments.size());