Adding OracleR2dbcWarning

Author: Michael-A-McMahon

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

@@ -0,0 +1,87 @@
+package oracle.r2dbc;
+
+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.
+ * </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)}:
+ * </p><pre>{@code
+ * result.flatMap(segment -> {
+ *   if (OracleR2dbcWarning.isWarning(segment)) {
+ *     logWarning(OracleR2dbcWarning.getWarning(segment));
+ *     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))
+ * }</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>
+ */
+public class OracleR2dbcWarning extends R2dbcException {
+
+  /**
+   * 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.
+   */
+  public OracleR2dbcWarning(String sql, SQLWarning sqlWarning) {
+    super(sqlWarning.getMessage(), sqlWarning.getSQLState(),
+      sqlWarning.getErrorCode(), sql, sqlWarning);
+  }
+
+  /**
+   * 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.
+   */
+  public static boolean isWarning(Segment segment) {
+    return segment instanceof Message
+      && ((Message)segment).exception() instanceof OracleR2dbcWarning;
+  }
+
+  /**
+   * 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}.
+   */
+  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();
+  }
+
+}

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

@@ -31,6 +31,7 @@
 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;
@@ -42,6 +43,7 @@
 import java.sql.SQLTransactionRollbackException;
 import java.sql.SQLTransientConnectionException;
 import java.sql.SQLTransientException;
+import java.sql.SQLWarning;
 import java.util.function.Supplier;
 
 /**
@@ -214,6 +216,9 @@ 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/test/java/oracle/r2dbc/impl/OracleResultImplTest.java

@@ -28,6 +28,8 @@
 import io.r2dbc.spi.Result.UpdateCount;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
+import io.r2dbc.spi.Statement;
+import oracle.r2dbc.OracleR2dbcWarning;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -55,6 +57,9 @@
 import static oracle.r2dbc.util.Awaits.tryAwaitExecution;
 import static oracle.r2dbc.util.Awaits.tryAwaitNone;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -624,4 +629,60 @@ else if (index == 1) {
     }
   }
 
+  /**
+   * Verifies that a warnings are emitted as {@code Message} segments with an
+   * {@link oracle.r2dbc.OracleR2dbcWarning}.
+   */
+  @Test
+  public void testOracleR2dbcWarning() {
+    Connection connection = awaitOne(sharedConnection());
+    try {
+
+      // Expect a warning for forcing a view that references a non-existant
+      // table
+      String sql = "CREATE OR REPLACE FORCE VIEW testOracleR2dbcWarning AS" +
+          " SELECT x FROM thisdoesnotexist";
+      Statement warningStatement = connection.createStatement(sql);
+
+      // Collect the segments
+      List<Result.Segment> segments =
+        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
+      // 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());
+      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));
+
+      // Verify that there are not any more segments
+      assertEquals(2, segments.size());
+    }
+    finally {
+      tryAwaitExecution(
+        connection.createStatement("DROP VIEW testOracleR2dbcWarning"));
+      tryAwaitNone(connection.close());
+    }
+  }
+
 }