Handling errors in parse part of extended query flow

Author: Squiry

src/main/java/io/r2dbc/postgresql/client/Client.java

@@ -71,6 +71,14 @@ default Flux<BackendMessage> exchange(Publisher<FrontendMessage> requests) {
      */
     Flux<BackendMessage> exchange(Predicate<BackendMessage> takeUntil, Publisher<FrontendMessage> requests);
 
+    /**
+     * Send one message without waiting for response.
+     *
+     * @param message outbound message
+     * @throws IllegalArgumentException if {@code message} is {@code null}
+     */
+    void send(FrontendMessage message);
+
     /**
      * Returns the {@link ByteBufAllocator}.
      *

src/main/java/io/r2dbc/postgresql/client/ExtendedQueryMessageFlow.java

@@ -20,7 +20,9 @@
 import io.r2dbc.postgresql.message.Format;
 import io.r2dbc.postgresql.message.backend.BackendMessage;
 import io.r2dbc.postgresql.message.backend.CloseComplete;
+import io.r2dbc.postgresql.message.backend.ErrorResponse;
 import io.r2dbc.postgresql.message.backend.ParseComplete;
+import io.r2dbc.postgresql.message.backend.ReadyForQuery;
 import io.r2dbc.postgresql.message.frontend.Bind;
 import io.r2dbc.postgresql.message.frontend.Close;
 import io.r2dbc.postgresql.message.frontend.Describe;
@@ -95,7 +97,20 @@ public static Flux<BackendMessage> parse(Client client, String name, String quer
         Assert.requireNonNull(query, "query must not be null");
         Assert.requireNonNull(types, "types must not be null");
 
-        return client.exchange(ParseComplete.class::isInstance, Flux.just(new Parse(name, types, query), Flush.INSTANCE));
+        /*
+         ParseComplete will be received if parse was successful
+         ReadyForQuery will be received as a response to Sync, which was send in case of error in parsing
+         */
+        return client.exchange(message -> message instanceof ParseComplete || message instanceof ReadyForQuery, Flux.just(new Parse(name, types, query), Flush.INSTANCE))
+            .doOnNext(message -> {
+                if (message instanceof ErrorResponse) {
+                    /*
+                    When an error is detected while processing any extended-query message, the backend issues ErrorResponse, then reads and discards messages until a Sync is reached.
+                    So we have to provide Sync message to continue.
+                    */
+                    client.send(Sync.INSTANCE);
+                }
+            });
     }
 
     /**

src/main/java/io/r2dbc/postgresql/client/ReactorNettyClient.java

@@ -209,6 +209,13 @@ public Flux<BackendMessage> exchange(Predicate<BackendMessage> takeUntil, Publis
             });
     }
 
+    @Override
+    public void send(FrontendMessage message) {
+        Assert.requireNonNull(message, "requests must not be null");
+
+        this.requests.next(message);
+    }
+
     private Mono<Void> resumeError(Throwable throwable) {
 
         handleConnectionError(throwable);

src/test/java/io/r2dbc/postgresql/client/ExtendedQueryMessageFlowTest.java

@@ -19,8 +19,10 @@
 import io.r2dbc.postgresql.message.backend.BindComplete;
 import io.r2dbc.postgresql.message.backend.CloseComplete;
 import io.r2dbc.postgresql.message.backend.CommandComplete;
+import io.r2dbc.postgresql.message.backend.ErrorResponse;
 import io.r2dbc.postgresql.message.backend.NoData;
 import io.r2dbc.postgresql.message.backend.ParseComplete;
+import io.r2dbc.postgresql.message.backend.ReadyForQuery;
 import io.r2dbc.postgresql.message.frontend.Bind;
 import io.r2dbc.postgresql.message.frontend.Close;
 import io.r2dbc.postgresql.message.frontend.Describe;
@@ -112,7 +114,22 @@ void parse() {
         ExtendedQueryMessageFlow
             .parse(client, "test-name", "test-query", Collections.singletonList(100))
             .as(StepVerifier::create)
-            .expectNext(ParseComplete.INSTANCE)
+            .verifyComplete();
+    }
+
+    @Test
+    void parseWithError() {
+        Client client = TestClient.builder()
+            .expectRequest(new Parse("test-name", Collections.singletonList(100), "test-query"), Flush.INSTANCE)
+            .thenRespond(new ErrorResponse(Collections.emptyList()))
+            .expectRequest(Sync.INSTANCE)
+            .thenRespond(new ReadyForQuery(ReadyForQuery.TransactionStatus.IDLE))
+            .build();
+
+        ExtendedQueryMessageFlow
+            .parse(client, "test-name", "test-query", Collections.singletonList(100))
+            .as(StepVerifier::create)
+            .expectNext(new ErrorResponse(Collections.emptyList()))
             .verifyComplete();
     }
 

src/test/java/io/r2dbc/postgresql/client/TestClient.java

@@ -119,7 +119,8 @@ public Flux<BackendMessage> exchange(Predicate<BackendMessage> takeUntil, Publis
                 Flux.from(requests)
                     .subscribe(this.requests::next, this.requests::error))
             .next()
-            .flatMapMany(Function.identity());
+            .flatMapMany(Function.identity())
+            .takeWhile(takeUntil.negate());
     }
 
     @Override
@@ -152,6 +153,11 @@ public boolean isConnected() {
         return this.connected;
     }
 
+    @Override
+    public void send(FrontendMessage message) {
+        this.requests.next(message);
+    }
+
     @Override
     public Disposable addNotificationListener(Consumer<NotificationResponse> consumer) {
         return this.notificationProcessor.subscribe(consumer);