fixup

Author: ArjanSchouten

src/jmh/java/io/r2dbc/postgresql/CopyInBenchmarks.java

@@ -65,7 +65,7 @@ public class CopyInBenchmarks extends BenchmarkSettings {
     @State(Scope.Benchmark)
     public static class ConnectionHolder {
 
-        @Param({"1", "100", "1000000"})
+        @Param({"0", "1", "100", "1000000"})
         int rows;
 
         final PgConnection jdbc;

src/main/java/io/r2dbc/postgresql/PostgresqlCopyIn.java

@@ -23,14 +23,14 @@
 import io.r2dbc.postgresql.message.backend.BackendMessage;
 import io.r2dbc.postgresql.message.backend.CommandComplete;
 import io.r2dbc.postgresql.message.backend.CopyInResponse;
-import io.r2dbc.postgresql.message.backend.ErrorResponse;
 import io.r2dbc.postgresql.message.backend.ReadyForQuery;
 import io.r2dbc.postgresql.message.frontend.CopyData;
 import io.r2dbc.postgresql.message.frontend.CopyDone;
 import io.r2dbc.postgresql.message.frontend.CopyFail;
 import io.r2dbc.postgresql.message.frontend.FrontendMessage;
 import io.r2dbc.postgresql.message.frontend.Query;
 import io.r2dbc.postgresql.util.Assert;
+import io.r2dbc.postgresql.util.Operators;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -69,6 +69,7 @@ private Mono<Long> copyIn(String sql, Flux<FrontendMessage> frontendMessages) {
         return startCopy(sql)
             .concatWith(backendMessages)
             .doOnCancel(() -> sendCopyFail("Cancelled"))
+            .as(Operators::discardOnCancel)
             .as(messages -> toResult(context, messages, ExceptionFactory.INSTANCE).getRowsUpdated());
     }
 
@@ -80,13 +81,14 @@ private Flux<BackendMessage> startCopy(String sql) {
             )
             .doOnNext(message -> {
                 if (message instanceof CommandComplete) {
-                    throw new IllegalArgumentException("Copy from stdin query expected but was ['" + sql + "']");
+                    throw new IllegalArgumentException("Copy from stdin query expected, sql='" + sql + "', message=" + message);
                 }
             });
     }
 
     private void sendCopyFail(String message) {
-        context.getClient().exchange(m -> m instanceof CommandComplete || m instanceof ErrorResponse, Mono.just(new CopyFail("Copy operation failed: " + message)))
+        context.getClient().exchange(Mono.just(new CopyFail("Copy operation failed: " + message)))
+            .as(Operators::discardOnCancel)
             .subscribe();
     }
 

src/test/java/io/r2dbc/postgresql/PostgresqlCopyInIntegrationTests.java

@@ -118,7 +118,7 @@ void shouldHandleErrorOnValidNonCopyInQuery() {
             .as(StepVerifier::create)
             .consumeErrorWith(e -> assertThat(e)
                 .isInstanceOf(IllegalArgumentException.class)
-                .hasMessage("Copy from stdin query expected but was ['SELECT 1']")
+                .hasMessage("Copy from stdin query expected, sql='SELECT 1', message=CommandComplete{command=SELECT, rowId=null, rows=1}")
             )
             .verify();
     }

src/test/java/io/r2dbc/postgresql/PostgresqlCopyInUnitTests.java

@@ -79,7 +79,7 @@ void copyInInvalidQuery() {
             .as(StepVerifier::create)
             .consumeErrorWith(e -> assertThat(e)
                 .isInstanceOf(IllegalArgumentException.class)
-                .hasMessage("Copy from stdin query expected but was ['" + sql + "']")
+                .hasMessage("Copy from stdin query expected, sql='invalid-sql', message=CommandComplete{command=command, rowId=0, rows=9}")
             )
             .verify();
     }