Ensure demand propagation from onRequest(…)

We now make sure to call demandMore() regardless of the drain loop to ensure that demand is propagated upstream even if the drain conditions are not met. Previously, concurrent consumption lead to an empty buffer without demand propagation and so emission got eventually stuck.

[resolves pgjdbc#292]

Author: mp911de

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

@@ -781,7 +781,7 @@ public Flux<BackendMessage> addConversation(Predicate<BackendMessage> takeUntil,
 
         public void onRequest(Conversation conversation, long n) {
             conversation.incrementDemand(n);
-
+            demandMore();
             tryDrainLoop();
         }
 

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

@@ -45,12 +45,14 @@
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.springframework.util.ReflectionUtils;
 import reactor.core.publisher.EmitterProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.UnicastProcessor;
+import reactor.core.scheduler.Schedulers;
 import reactor.netty.Connection;
 import reactor.test.StepVerifier;
 
@@ -75,6 +77,9 @@
 import static org.assertj.core.api.Assertions.fail;
 import static org.assertj.core.api.Assumptions.assumeThat;
 
+/**
+ * Integration tests for {@link ReactorNettyClient}.
+ */
 final class ReactorNettyClientIntegrationTests {
 
     @RegisterExtension
@@ -92,6 +97,37 @@ final class ReactorNettyClientIntegrationTests {
                 Collections.emptyMap()))
         .block();
 
+    @BeforeEach
+    void before() {
+        SERVER.getJdbcOperations().execute("DROP TABLE IF EXISTS test");
+        SERVER.getJdbcOperations().execute("CREATE TABLE test ( value INTEGER )");
+    }
+
+    @AfterEach
+    void after() {
+        SERVER.getJdbcOperations().execute("DROP TABLE IF EXISTS test");
+        this.client.close()
+            .block();
+    }
+
+
+    @Test
+    @Timeout(50)
+    void concurrentConsumptionShouldComplete() {
+
+        IntStream.range(0, 1000)
+            .forEach(i -> SERVER.getJdbcOperations().update("INSERT INTO test VALUES(?),(?),(?),(?),(?),(?),(?),(?),(?),(?)", i, i, i, i, i, i, i, i, i, i));
+
+        this.client
+            .exchange(Mono.just(new Query("SELECT value FROM test")))
+            .limitRate(1)
+            .publishOn(Schedulers.elastic())
+            .doOnNext(ReferenceCountUtil::release)
+            .as(StepVerifier::create)
+            .thenConsumeWhile((t) -> true)
+            .verifyComplete();
+    }
+
     @Test
     void close() {
         this.client.close()
@@ -158,30 +194,13 @@ void shouldCancelExchangeOnCloseInFlight() throws Exception {
         }
     }
 
-    @AfterEach
-    void closeClient() {
-        SERVER.getJdbcOperations().execute("DROP TABLE IF EXISTS test");
-        this.client.close()
-            .block();
-    }
 
     @Test
     void constructorNoHost() {
         assertThatIllegalArgumentException().isThrownBy(() -> ReactorNettyClient.connect(null, SERVER.getPort()))
             .withMessage("host must not be null");
     }
 
-    @BeforeEach
-    void createTable() {
-        dropTable();
-        SERVER.getJdbcOperations().execute("CREATE TABLE test ( value INTEGER )");
-    }
-
-    @AfterEach
-    void dropTable() {
-        SERVER.getJdbcOperations().execute("DROP TABLE IF EXISTS test");
-    }
-
     @Test
     void exchange() {
         SERVER.getJdbcOperations().execute("INSERT INTO test VALUES (100)");