Use Long update counts in test code

Author: Michael-A-McMahon

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

@@ -952,7 +952,7 @@ public void testSetAutoCommit() {
             "Unexpected value returned by isAutoCommit() before subscribing to"
               + " setAutoCommit(true) publisher");
           awaitMany(
-            List.of(1, 1),
+            List.of(1L, 1L),
             Flux.from(sessionA.createBatch()
               .add("INSERT INTO testSetAutoCommit VALUES ('C')")
               .add("INSERT INTO testSetAutoCommit VALUES ('C')")

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

@@ -89,9 +89,9 @@ public void testGetRowsUpdated() {
           .toIterable()
           .iterator();
       Result insertResult0 = insertResults.next();
-      Publisher<Integer> insertCountPublisher0 =
+      Publisher<Long> insertCountPublisher0 =
         insertResult0.getRowsUpdated();
-      awaitOne(1, insertCountPublisher0);
+      awaitOne(1L, insertCountPublisher0);
 
       // Expect IllegalStateException from multiple Result consumptions.
       assertThrows(IllegalStateException.class,
@@ -100,12 +100,12 @@ public void testGetRowsUpdated() {
         () -> insertResult0.map((row, metadata) -> "unexpected"));
 
       // Expect update count publisher to support multiple subscribers
-      awaitOne(1, insertCountPublisher0);
+      awaitOne(1L, insertCountPublisher0);
 
       Result insertResult1 = insertResults.next();
-      Publisher<Integer> insertCountPublisher1 =
+      Publisher<Long> insertCountPublisher1 =
         insertResult1.getRowsUpdated();
-      awaitOne(1, insertCountPublisher1);
+      awaitOne(1L, insertCountPublisher1);
 
       // Expect IllegalStateException from multiple Result consumptions.
       assertThrows(IllegalStateException.class,
@@ -114,53 +114,53 @@ public void testGetRowsUpdated() {
         () -> insertResult1.map((row, metadata) -> "unexpected"));
 
       // Expect update count publisher to support multiple subscribers
-      awaitOne(1, insertCountPublisher1);
+      awaitOne(1L, insertCountPublisher1);
 
       // Expect an update count of zero from UPDATE of zero rows
       consumeOne(connection.createStatement(
         "UPDATE testGetRowsUpdated SET y = 99 WHERE x = 99")
         .execute(),
         noUpdateResult -> {
-          Publisher<Integer> noUpdateCountPublisher =
+          Publisher<Long> noUpdateCountPublisher =
             noUpdateResult.getRowsUpdated();
-          awaitOne(0, noUpdateCountPublisher);
+          awaitOne(0L, noUpdateCountPublisher);
 
           // Expect IllegalStateException from multiple Result consumptions.
           assertThrows(IllegalStateException.class,
             () -> noUpdateResult.map((row, metadata) -> "unexpected"));
           assertThrows(IllegalStateException.class, noUpdateResult::getRowsUpdated);
 
           // Expect update count publisher to support multiple subscribers
-          awaitOne(0, noUpdateCountPublisher);
+          awaitOne(0L, noUpdateCountPublisher);
         });
 
       // Expect update count of 2 from UPDATE of 2 rows
       consumeOne(connection.createStatement(
         "UPDATE testGetRowsUpdated SET y = 2 WHERE x = 0")
         .execute(),
         updateResult -> {
-        Publisher<Integer> updateCountPublisher = updateResult.getRowsUpdated();
-        awaitOne(2, updateCountPublisher);
+        Publisher<Long> updateCountPublisher = updateResult.getRowsUpdated();
+        awaitOne(2L, updateCountPublisher);
 
         // Expect IllegalStateException from multiple Result consumptions.
         assertThrows(IllegalStateException.class,
           () -> updateResult.map((row, metadata) -> "unexpected"));
         assertThrows(IllegalStateException.class, updateResult::getRowsUpdated);
 
         // Expect update count publisher to support multiple subscribers
-        awaitOne(2, updateCountPublisher);
+        awaitOne(2L, updateCountPublisher);
       });
 
       // Expect no update count from SELECT
       awaitNone(Mono.from(connection.createStatement(
         "SELECT x,y FROM testGetRowsUpdated")
         .execute())
         .flatMapMany(selectResult -> {
-          Publisher<Integer> selectCountPublisher =
+          Publisher<Long> selectCountPublisher =
             selectResult.getRowsUpdated();
 
           // Expect update count publisher to support multiple subscribers
-          Publisher<Integer> result = Flux.concat(
+          Publisher<Long> result = Flux.concat(
             Mono.from(selectCountPublisher).cache(),
             Mono.from(selectCountPublisher).cache());
 
@@ -178,16 +178,16 @@ public void testGetRowsUpdated() {
         .bind("x", 0)
         .execute(),
         deleteResult -> {
-          Publisher<Integer> deleteCountPublisher = deleteResult.getRowsUpdated();
-          awaitOne(2, deleteCountPublisher);
+          Publisher<Long> deleteCountPublisher = deleteResult.getRowsUpdated();
+          awaitOne(2L, deleteCountPublisher);
 
           // Expect IllegalStateException from multiple Result consumptions.
           assertThrows(IllegalStateException.class,
             () -> deleteResult.map((row, metadata) -> "unexpected"));
           assertThrows(IllegalStateException.class, deleteResult::getRowsUpdated);
 
           // Expect update count publisher to support multiple subscribers
-          awaitOne(2, deleteCountPublisher);
+          awaitOne(2L, deleteCountPublisher);
         });
     }
     finally {
@@ -473,7 +473,7 @@ public void testFilter() {
       // UpdateCount segment to be published by getRowsUpdated
       AtomicReference<UpdateCount> unfilteredUpdateCount =
         new AtomicReference<>(null);
-      awaitOne(1, Flux.from(connection.createStatement(
+      awaitOne(1L, Flux.from(connection.createStatement(
         "INSERT INTO testFilter VALUES (1)")
         .execute())
         .map(result ->
@@ -529,7 +529,7 @@ public void testFilter() {
         .execute())
         .block(sqlTimeout());
       Result filteredResult = unfilteredResult.filter(segment -> false);
-      Publisher<Integer> filteredUpdateCounts = filteredResult.getRowsUpdated();
+      Publisher<Long> filteredUpdateCounts = filteredResult.getRowsUpdated();
       assertThrows(
         IllegalStateException.class, unfilteredResult::getRowsUpdated);
       assertThrows(
@@ -545,13 +545,13 @@ public void testFilter() {
         .block(sqlTimeout());
       Result filteredResult2 = unfilteredResult2.filter(segment ->
         fail("Unexpected invocation"));
-      Publisher<Integer> unfilteredUpdateCounts =
+      Publisher<Long> unfilteredUpdateCounts =
         unfilteredResult2.getRowsUpdated();
       assertThrows(
         IllegalStateException.class, filteredResult2::getRowsUpdated);
       assertThrows(
         IllegalStateException.class, unfilteredResult2::getRowsUpdated);
-      awaitOne(1, unfilteredUpdateCounts);
+      awaitOne(1L, unfilteredUpdateCounts);
 
       // Execute an INSERT that fails, and filter Message type segments.
       // Expect the Result to not emit {@code onError} when consumed.

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

@@ -45,8 +45,10 @@
 import java.util.NoSuchElementException;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.LongStream;
 import java.util.stream.Stream;
 
 import static java.util.Arrays.asList;
@@ -811,7 +813,7 @@ public void testAdd() {
 
       // Expect the statement to execute with previously added binds, and
       // then emit an error if binds are missing in the final set of binds.
-      List<Signal<Integer>> signals =
+      List<Signal<Long>> signals =
         awaitOne(Flux.from(connection.createStatement(
           "INSERT INTO testAdd VALUES (:x, :y)")
           .bind("x", 0).bind("y", 1).add()
@@ -911,7 +913,7 @@ public void testExecute() {
         selectStatement::execute);
 
       // Expect update to execute when a subscriber subscribes
-      awaitOne(1,
+      awaitOne(1L,
         Flux.from(updatePublisher)
           .flatMap(result -> result.getRowsUpdated()));
       awaitQuery(
@@ -1767,9 +1769,9 @@ public void testNoOutImplicitResult() {
       IntStream.range(0, 100)
         .forEach(i -> insert.bind(0, i).add());
       insert.bind(0, 100);
-      awaitOne(101, Flux.from(insert.execute())
+      awaitOne(101L, Flux.from(insert.execute())
         .flatMap(Result::getRowsUpdated)
-        .reduce(0, (total, updateCount) -> total + updateCount));
+        .reduce(0L, (total, updateCount) -> total + updateCount));
 
       // Create a procedure that returns a cursor
       awaitExecution(connection.createStatement(
@@ -1844,8 +1846,8 @@ public void testNoOutImplicitResult() {
               .collectList()));
 
       // Expect Implicit Results to have no update counts
-      AtomicInteger count = new AtomicInteger(-9);
-      awaitMany(asList(-9, -10),
+      AtomicLong count = new AtomicLong(-9);
+      awaitMany(asList(-9L, -10L),
         Flux.from(connection.createStatement("BEGIN countDown; END;")
           .execute())
           .concatMap(result ->
@@ -1878,9 +1880,9 @@ public void testOutAndImplicitResult() {
       IntStream.range(0, 100)
         .forEach(i -> insert.bind(0, i).add());
       insert.bind(0, 100);
-      awaitOne(101, Flux.from(insert.execute())
+      awaitOne(101L, Flux.from(insert.execute())
         .flatMap(Result::getRowsUpdated)
-        .reduce(0, (total, updateCount) -> total + updateCount));
+        .reduce(0L, (total, updateCount) -> total + updateCount));
 
       // Create a procedure that returns a cursor
       awaitExecution(connection.createStatement(
@@ -1961,8 +1963,8 @@ public void testOutAndImplicitResult() {
               .collectList()));
 
       // Expect Implicit Results to have no update counts
-      AtomicInteger count = new AtomicInteger(-8);
-      awaitMany(asList(-8, -9, -10),
+      AtomicLong count = new AtomicLong(-8);
+      awaitMany(asList(-8L, -9L, -10L),
         Flux.from(connection.createStatement("BEGIN countDown(?); END;")
           .bind(0, Parameters.out(R2dbcType.VARCHAR))
           .execute())
@@ -2083,7 +2085,7 @@ public void testConcurrentFetch() {
 
       // Create many statements and execute them in parallel. "Many" should
       // be enough to exhaust the common ForkJoinPool if any thread gets blocked
-      Publisher<Integer>[] publishers =
+      Publisher<Long>[] publishers =
         new Publisher[ForkJoinPool.getCommonPoolParallelism() * 4];
 
       for (int i = 0; i < publishers.length; i++) {
@@ -2098,9 +2100,9 @@ public void testConcurrentFetch() {
             statement.add().bind(0, value);
           });
 
-        Mono<Integer> mono = Flux.from(statement.execute())
+        Mono<Long> mono = Flux.from(statement.execute())
           .flatMap(Result::getRowsUpdated)
-          .collect(Collectors.summingInt(Integer::intValue))
+          .collect(Collectors.summingLong(Long::longValue))
           .cache();
 
         // Execute in parallel, and retain the result for verification later
@@ -2110,7 +2112,7 @@ public void testConcurrentFetch() {
 
       // Expect each publisher to emit an update count of 100
       awaitMany(
-        IntStream.range(0, publishers.length)
+        LongStream.range(0, publishers.length)
           .map(i -> 100)
           .boxed()
           .collect(Collectors.toList()),

src/test/java/oracle/r2dbc/test/OracleTestKit.java

@@ -213,9 +213,9 @@ private Object extractColumn(String name, Row row) {
    * </p>
    */
   @Override
-  public Mono<Integer> extractRowsUpdated(Result result) {
+  public Mono<Long> extractRowsUpdated(Result result) {
     return Flux.from(result.getRowsUpdated())
-      .reduce(0, (total, updateCount) -> total + updateCount);
+      .reduce(0L, (total, updateCount) -> total + updateCount);
   }
 
   @Override

src/test/java/oracle/r2dbc/util/Awaits.java

@@ -269,6 +269,7 @@ public static void awaitUpdate(
       expectedCounts,
       Flux.from(statement.execute())
         .flatMap(result -> Flux.from(result.getRowsUpdated()))
+        .map(Math::toIntExact)
         .collectList()
         .block(sqlTimeout()),
       "Unexpected update counts");