Let Result.getRowsUpdated() return Long.

mp911de · mp911de · commit 1277a522fe40 · 2022-01-27T10:07:17.000+01:00
[resolves #484] Signed-off-by: Mark Paluch <mpaluch@vmware.com>
diff --git a/pom.xml b/pom.xml
@@ -47,7 +47,7 @@
         <postgresql.version>42.3.1</postgresql.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
-        <r2dbc-spi.version>0.9.0.RELEASE</r2dbc-spi.version>
+        <r2dbc-spi.version>1.0.0.BUILD-SNAPSHOT</r2dbc-spi.version>
         <reactor.version>2020.0.15</reactor.version>
         <scram-client.version>2.1</scram-client.version>
         <spring-framework.version>5.3.14</spring-framework.version>
diff --git a/src/main/java/io/r2dbc/postgresql/PostgresqlConnection.java b/src/main/java/io/r2dbc/postgresql/PostgresqlConnection.java
@@ -71,7 +71,7 @@ final class PostgresqlConnection implements io.r2dbc.postgresql.api.PostgresqlCo
 
     private final Codecs codecs;
 
-    private final Flux<Integer> validationQuery;
+    private final Flux<Long> validationQuery;
 
     private final AtomicReference<NotificationAdapter> notificationAdapter = new AtomicReference<>();
 
@@ -380,15 +380,15 @@ public Mono<Boolean> validate(ValidationDepth depth) {
                 return;
             }
 
-            this.validationQuery.subscribe(new CoreSubscriber<Integer>() {
+            this.validationQuery.subscribe(new CoreSubscriber<Long>() {
 
                 @Override
                 public void onSubscribe(Subscription s) {
                     s.request(Integer.MAX_VALUE);
                 }
 
                 @Override
-                public void onNext(Integer integer) {
+                public void onNext(Long integer) {
 
                 }
 
diff --git a/src/main/java/io/r2dbc/postgresql/PostgresqlResult.java b/src/main/java/io/r2dbc/postgresql/PostgresqlResult.java
@@ -61,7 +61,7 @@ final class PostgresqlResult extends AbstractReferenceCounted implements io.r2db
 
     @Override
     @SuppressWarnings({"rawtypes", "unchecked"})
-    public Mono<Integer> getRowsUpdated() {
+    public Mono<Long> getRowsUpdated() {
 
         return this.messages
             .<Integer>handle((message, sink) -> {
@@ -88,7 +88,7 @@ public Mono<Integer> getRowsUpdated() {
                     return;
                 }
 
-                int sum = 0;
+                long sum = 0;
 
                 for (Integer integer : list) {
                     sum += integer;
diff --git a/src/main/java/io/r2dbc/postgresql/PostgresqlSegmentResult.java b/src/main/java/io/r2dbc/postgresql/PostgresqlSegmentResult.java
@@ -114,7 +114,7 @@ private PostgresqlSegmentResult(Flux<Segment> segments) {
     }
 
     @Override
-    public Mono<Integer> getRowsUpdated() {
+    public Mono<Long> getRowsUpdated() {
         return this.segments
             .<Integer>handle((segment, sink) -> {
 
@@ -137,7 +137,7 @@ public Mono<Integer> getRowsUpdated() {
                     return;
                 }
 
-                int sum = 0;
+                long sum = 0;
 
                 for (Integer integer : list) {
                     sum += integer;
diff --git a/src/main/java/io/r2dbc/postgresql/api/PostgresqlResult.java b/src/main/java/io/r2dbc/postgresql/api/PostgresqlResult.java
@@ -33,7 +33,7 @@ public interface PostgresqlResult extends Result {
      * {@inheritDoc}
      */
     @Override
-    Mono<Integer> getRowsUpdated();
+    Mono<Long> getRowsUpdated();
 
     /**
      * {@inheritDoc}
diff --git a/src/test/java/io/r2dbc/postgresql/AbstractCodecIntegrationTests.java b/src/test/java/io/r2dbc/postgresql/AbstractCodecIntegrationTests.java
@@ -738,7 +738,7 @@ private <IN, OUT> void testCodec(Class<IN> javaType, IN value, Class<OUT> outTyp
 
                         .concatWith(close(connection)))
                     .as(StepVerifier::create)
-                    .expectNext(1)
+                    .expectNext(1L)
                     .verifyComplete();
 
                 SERVER.getJdbcOperations().execute("DELETE FROM test");
@@ -754,7 +754,7 @@ private <IN, OUT> void testCodec(Class<IN> javaType, IN value, Class<OUT> outTyp
 
                         .concatWith(close(connection)))
                     .as(StepVerifier::create)
-                    .expectNext(1)
+                    .expectNext(1L)
                     .verifyComplete();
             } else {
 
@@ -769,7 +769,7 @@ private <IN, OUT> void testCodec(Class<IN> javaType, IN value, Class<OUT> outTyp
 
                         .concatWith(close(connection)))
                     .as(StepVerifier::create)
-                    .expectNext(1)
+                    .expectNext(1L)
                     .verifyComplete();
 
                 SERVER.getJdbcOperations().execute("DELETE FROM test");
@@ -785,7 +785,7 @@ private <IN, OUT> void testCodec(Class<IN> javaType, IN value, Class<OUT> outTyp
 
                         .concatWith(close(connection)))
                     .as(StepVerifier::create)
-                    .expectNext(1)
+                    .expectNext(1L)
                     .verifyComplete();
             }
 
@@ -856,7 +856,7 @@ private <W, R> void testCodecReadAs(W toWrite, Class<R> javaTypeToRead, Consumer
 
                     .concatWith(close(connection)))
                 .as(StepVerifier::create)
-                .expectNext(1)
+                .expectNext(1L)
                 .verifyComplete();
 
             this.connectionFactory.create()
diff --git a/src/test/java/io/r2dbc/postgresql/PostgresqlResultUnitTests.java b/src/test/java/io/r2dbc/postgresql/PostgresqlResultUnitTests.java
@@ -43,7 +43,7 @@ void toResultCommandComplete() {
 
         result.getRowsUpdated()
             .as(StepVerifier::create)
-            .expectNext(1)
+            .expectNext(1L)
             .verifyComplete();
     }
 
@@ -57,7 +57,7 @@ void toResultCommandCompleteUsingSegments() {
 
         result.getRowsUpdated()
             .as(StepVerifier::create)
-            .expectNext(1)
+            .expectNext(1L)
             .verifyComplete();
     }
 
diff --git a/src/test/java/io/r2dbc/postgresql/PostgresqlSegmentResultUnitTests.java b/src/test/java/io/r2dbc/postgresql/PostgresqlSegmentResultUnitTests.java
@@ -93,7 +93,7 @@ void shouldConsumeRowsUpdated() {
 
         result.getRowsUpdated()
             .as(StepVerifier::create)
-            .expectNext(42)
+            .expectNext(42L)
             .verifyComplete();
     }
 
@@ -105,7 +105,7 @@ void filterShouldRetainUpdateCount() {
 
         result.filter(Result.UpdateCount.class::isInstance).getRowsUpdated()
             .as(StepVerifier::create)
-            .expectNext(42)
+            .expectNext(42L)
             .verifyComplete();
     }
 
diff --git a/src/test/java/io/r2dbc/postgresql/PostgresqlStatementErrorsIntegrationTests.java b/src/test/java/io/r2dbc/postgresql/PostgresqlStatementErrorsIntegrationTests.java
@@ -47,7 +47,7 @@ void shouldReportDataIntegrityViolationUsingSimpleFlow() {
 
         SERVER.getJdbcOperations().execute("CREATE TABLE test (id SERIAL PRIMARY KEY)");
 
-        Flux<Integer> insert = Flux.from(this.connection.createStatement("INSERT INTO test (id) VALUES (1) RETURNING *").execute()).flatMap(Result::getRowsUpdated);
+        Flux<?> insert = Flux.from(this.connection.createStatement("INSERT INTO test (id) VALUES (1) RETURNING *").execute()).flatMap(Result::getRowsUpdated);
 
         insert.thenMany(insert).as(StepVerifier::create).verifyError(R2dbcDataIntegrityViolationException.class);
     }
@@ -57,7 +57,7 @@ void shouldReportDataIntegrityViolationUsingExtendedFlow() {
 
         SERVER.getJdbcOperations().execute("CREATE TABLE test (id SERIAL PRIMARY KEY)");
 
-        Flux<Integer> insert = Flux.from(this.connection.createStatement("INSERT INTO test (id) VALUES ($1) RETURNING *").bind("$1", 1).execute()).flatMap(Result::getRowsUpdated);
+        Flux<?> insert = Flux.from(this.connection.createStatement("INSERT INTO test (id) VALUES ($1) RETURNING *").bind("$1", 1).execute()).flatMap(Result::getRowsUpdated);
 
         insert.thenMany(insert).as(StepVerifier::create).verifyError(R2dbcDataIntegrityViolationException.class);
     }
diff --git a/src/test/java/io/r2dbc/postgresql/api/MockPostgresqlResult.java b/src/test/java/io/r2dbc/postgresql/api/MockPostgresqlResult.java
@@ -42,7 +42,7 @@ public static Builder builder() {
     }
 
     @Override
-    public Mono<Integer> getRowsUpdated() {
+    public Mono<Long> getRowsUpdated() {
         return Mono.empty();
     }
 
diff --git a/src/test/java/io/r2dbc/postgresql/codec/EnumCodecIntegrationTests.java b/src/test/java/io/r2dbc/postgresql/codec/EnumCodecIntegrationTests.java
@@ -86,7 +86,7 @@ void shouldBindEnumTypeAsString() {
             .execute()
             .flatMap(PostgresqlResult::getRowsUpdated)
             .as(StepVerifier::create)
-            .expectNext(1)
+            .expectNext(1L)
             .verifyComplete();
 
         String result = SERVER.getJdbcOperations().queryForObject("SELECT the_value FROM enum_test", String.class);
@@ -107,7 +107,7 @@ void shouldBindEnumArrayTypeAsString() {
             .execute()
             .flatMap(PostgresqlResult::getRowsUpdated)
             .as(StepVerifier::create)
-            .expectNext(1)
+            .expectNext(1L)
             .verifyComplete();
 
         String result = SERVER.getJdbcOperations().queryForObject("SELECT the_value FROM enum_test", String.class);
@@ -125,7 +125,7 @@ void shouldBindEnumArrayType() {
             .execute()
             .flatMap(PostgresqlResult::getRowsUpdated)
             .as(StepVerifier::create)
-            .expectNext(1)
+            .expectNext(1L)
             .verifyComplete();
 
         this.connection.createStatement("SELECT * FROM enum_test")
diff --git a/src/test/java/io/r2dbc/postgresql/replication/LogicalDecodeIntegrationTests.java b/src/test/java/io/r2dbc/postgresql/replication/LogicalDecodeIntegrationTests.java
@@ -102,7 +102,7 @@ void shouldReceiveReplication() {
 
         ReplicationStream replicationStream = replicationConnection.startReplication(replicationRequest).block(Duration.ofSeconds(10));
 
-        connection.createStatement("INSERT INTO logical_decode_test VALUES('Hello World')").execute().flatMap(PostgresqlResult::getRowsUpdated).as(StepVerifier::create).expectNext(1).verifyComplete();
+        connection.createStatement("INSERT INTO logical_decode_test VALUES('Hello World')").execute().flatMap(PostgresqlResult::getRowsUpdated).as(StepVerifier::create).expectNext(1L).verifyComplete();
 
         replicationStream.map(byteBuf -> byteBuf.toString(StandardCharsets.UTF_8))
             .as(StepVerifier::create)

Original file line number	Diff line number	Diff line change
`@@ -71,7 +71,7 @@ final class PostgresqlConnection implements io.r2dbc.postgresql.api.PostgresqlCo`
`71`	`71`
`72`	`72`	`private final Codecs codecs;`
`73`	`73`
`74`		`- private final Flux<Integer> validationQuery;`
	`74`	`+ private final Flux<Long> validationQuery;`
`75`	`75`
`76`	`76`	`private final AtomicReference<NotificationAdapter> notificationAdapter = new AtomicReference<>();`
`77`	`77`
`@@ -380,15 +380,15 @@ public Mono<Boolean> validate(ValidationDepth depth) {`
`380`	`380`	`return;`
`381`	`381`	`}`
`382`	`382`
`383`		`- this.validationQuery.subscribe(new CoreSubscriber<Integer>() {`
	`383`	`+ this.validationQuery.subscribe(new CoreSubscriber<Long>() {`
`384`	`384`
`385`	`385`	`@Override`
`386`	`386`	`public void onSubscribe(Subscription s) {`
`387`	`387`	`s.request(Integer.MAX_VALUE);`
`388`	`388`	`}`
`389`	`389`
`390`	`390`	`@Override`
`391`		`- public void onNext(Integer integer) {`
	`391`	`+ public void onNext(Long integer) {`
`392`	`392`
`393`	`393`	`}`
`394`	`394`
Original file line number	Diff line number	Diff line change
`@@ -114,7 +114,7 @@ private PostgresqlSegmentResult(Flux<Segment> segments) {`
`114`	`114`	`}`
`115`	`115`
`116`	`116`	`@Override`
`117`		`- public Mono<Integer> getRowsUpdated() {`
	`117`	`+ public Mono<Long> getRowsUpdated() {`
`118`	`118`	`return this.segments`
`119`	`119`	`.<Integer>handle((segment, sink) -> {`
`120`	`120`
`@@ -137,7 +137,7 @@ public Mono<Integer> getRowsUpdated() {`
`137`	`137`	`return;`
`138`	`138`	`}`
`139`	`139`
`140`		`- int sum = 0;`
	`140`	`+ long sum = 0;`
`141`	`141`
`142`	`142`	`for (Integer integer : list) {`
`143`	`143`	`sum += integer;`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ void toResultCommandComplete() {`
`43`	`43`
`44`	`44`	`result.getRowsUpdated()`
`45`	`45`	`.as(StepVerifier::create)`
`46`		`- .expectNext(1)`
	`46`	`+ .expectNext(1L)`
`47`	`47`	`.verifyComplete();`
`48`	`48`	`}`
`49`	`49`
`@@ -57,7 +57,7 @@ void toResultCommandCompleteUsingSegments() {`
`57`	`57`
`58`	`58`	`result.getRowsUpdated()`
`59`	`59`	`.as(StepVerifier::create)`
`60`		`- .expectNext(1)`
	`60`	`+ .expectNext(1L)`
`61`	`61`	`.verifyComplete();`
`62`	`62`	`}`
`63`	`63`
Original file line number	Diff line number	Diff line change
`@@ -93,7 +93,7 @@ void shouldConsumeRowsUpdated() {`
`93`	`93`
`94`	`94`	`result.getRowsUpdated()`
`95`	`95`	`.as(StepVerifier::create)`
`96`		`- .expectNext(42)`
	`96`	`+ .expectNext(42L)`
`97`	`97`	`.verifyComplete();`
`98`	`98`	`}`
`99`	`99`
`@@ -105,7 +105,7 @@ void filterShouldRetainUpdateCount() {`
`105`	`105`
`106`	`106`	`result.filter(Result.UpdateCount.class::isInstance).getRowsUpdated()`
`107`	`107`	`.as(StepVerifier::create)`
`108`		`- .expectNext(42)`
	`108`	`+ .expectNext(42L)`
`109`	`109`	`.verifyComplete();`
`110`	`110`	`}`
`111`	`111`
Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@ void shouldReportDataIntegrityViolationUsingSimpleFlow() {`
`47`	`47`
`48`	`48`	`SERVER.getJdbcOperations().execute("CREATE TABLE test (id SERIAL PRIMARY KEY)");`
`49`	`49`
`50`		`- Flux<Integer> insert = Flux.from(this.connection.createStatement("INSERT INTO test (id) VALUES (1) RETURNING *").execute()).flatMap(Result::getRowsUpdated);`
	`50`	`+ Flux<?> insert = Flux.from(this.connection.createStatement("INSERT INTO test (id) VALUES (1) RETURNING *").execute()).flatMap(Result::getRowsUpdated);`
`51`	`51`
`52`	`52`	`insert.thenMany(insert).as(StepVerifier::create).verifyError(R2dbcDataIntegrityViolationException.class);`
`53`	`53`	`}`
`@@ -57,7 +57,7 @@ void shouldReportDataIntegrityViolationUsingExtendedFlow() {`
`57`	`57`
`58`	`58`	`SERVER.getJdbcOperations().execute("CREATE TABLE test (id SERIAL PRIMARY KEY)");`
`59`	`59`
`60`		`- Flux<Integer> insert = Flux.from(this.connection.createStatement("INSERT INTO test (id) VALUES ($1) RETURNING *").bind("$1", 1).execute()).flatMap(Result::getRowsUpdated);`
	`60`	`+ Flux<?> insert = Flux.from(this.connection.createStatement("INSERT INTO test (id) VALUES ($1) RETURNING *").bind("$1", 1).execute()).flatMap(Result::getRowsUpdated);`
`61`	`61`
`62`	`62`	`insert.thenMany(insert).as(StepVerifier::create).verifyError(R2dbcDataIntegrityViolationException.class);`
`63`	`63`	`}`
Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,7 @@ public static Builder builder() {`
`42`	`42`	`}`
`43`	`43`
`44`	`44`	`@Override`
`45`		`- public Mono<Integer> getRowsUpdated() {`
	`45`	`+ public Mono<Long> getRowsUpdated() {`
`46`	`46`	`return Mono.empty();`
`47`	`47`	`}`
`48`	`48`