Query never completes fix

The problem: when our receive buffer fills and conversation demand drops to zero and last item is `CommandComplete` we never request more frames and will never get `ReadyForQuery`

Signed-off-by: Mark Paluch <[email protected]>

[closes #401][#426]

Author: Squiry

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

@@ -23,20 +23,13 @@
 import io.r2dbc.postgresql.PostgresqlConnectionFactory;
 import io.r2dbc.postgresql.api.PostgresqlConnection;
 import io.r2dbc.postgresql.authentication.PasswordAuthenticationHandler;
-import io.r2dbc.postgresql.message.Format;
 import io.r2dbc.postgresql.message.backend.BackendMessage;
-import io.r2dbc.postgresql.message.backend.BindComplete;
 import io.r2dbc.postgresql.message.backend.CommandComplete;
 import io.r2dbc.postgresql.message.backend.DataRow;
 import io.r2dbc.postgresql.message.backend.NotificationResponse;
 import io.r2dbc.postgresql.message.backend.RowDescription;
-import io.r2dbc.postgresql.message.frontend.Bind;
-import io.r2dbc.postgresql.message.frontend.Describe;
-import io.r2dbc.postgresql.message.frontend.Execute;
-import io.r2dbc.postgresql.message.frontend.ExecutionType;
 import io.r2dbc.postgresql.message.frontend.FrontendMessage;
 import io.r2dbc.postgresql.message.frontend.Query;
-import io.r2dbc.postgresql.message.frontend.Sync;
 import io.r2dbc.postgresql.util.PostgresqlServerExtension;
 import io.r2dbc.spi.R2dbcNonTransientResourceException;
 import io.r2dbc.spi.R2dbcPermissionDeniedException;
@@ -60,6 +53,7 @@
 import java.io.File;
 import java.lang.reflect.Field;
 import java.time.Duration;
+import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -70,7 +64,6 @@
 import java.util.function.Function;
 import java.util.stream.IntStream;
 
-import static io.r2dbc.postgresql.codec.PostgresqlObjectId.INT4;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.assertj.core.api.Assertions.fail;
@@ -342,6 +335,33 @@ void unixDomainSocketTest() {
             .verifyComplete();
     }
 
+    @Test
+    @Timeout(10)
+    void queryNeverCompletes() {
+        PostgresqlConnectionFactory connectionFactory = new PostgresqlConnectionFactory(PostgresqlConnectionConfiguration.builder()
+            .host(SERVER.getHost())
+            .port(SERVER.getPort())
+            .username(SERVER.getUsername())
+            .password(SERVER.getPassword())
+            .database(SERVER.getDatabase())
+            .applicationName(ReactorNettyClientIntegrationTests.class.getName())
+            .build());
+        Flux.usingWhen(
+            connectionFactory.create(),
+            connection -> connection.createStatement("SELECT floor(random() * 100) FROM generate_series(1, 256 + 15)").execute()
+                .limitRate(1, 1)
+                .flatMap(r -> r.map((row, meta) -> row.get(0)).limitRate(1, 1))
+                .limitRate(1, 1)
+                .delayElements(Duration.of(1, ChronoUnit.MILLIS))
+                .limitRate(1, 1),
+            io.r2dbc.spi.Connection::close
+        )
+            .as(StepVerifier::create)
+            .thenAwait()
+            .expectNextCount(256 + 15)
+            .verifyComplete();
+    }
+
     public static class FailedVerification implements HostnameVerifier {
 
         @Override

src/test/java/io/r2dbc/postgresql/util/BlockhoundExceptions.java

@@ -29,6 +29,7 @@ public class BlockhoundExceptions implements BlockHoundIntegration {
     @Override
     public void applyTo(BlockHound.Builder builder) {
         builder.allowBlockingCallsInside(SecureRandom.class.getName(), "next");
+        builder.allowBlockingCallsInside("io.r2dbc.postgresql.client.ReactorNettyClientIntegrationTests", "queryNeverCompletes");
     }
 
 }