Use UnicastProcessor in Extended Flow to avoid overflow of outbound writes.

We now use UnicastProcessor to buffer outbound writes when consuming a cursor. Previously, DirectProcessor would fail if netty operator doesn't have demand caused by channel writability switches.

[resolves #395]

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

Author: mp911de

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

@@ -43,11 +43,12 @@
 import io.r2dbc.postgresql.message.frontend.Parse;
 import io.r2dbc.postgresql.message.frontend.Sync;
 import io.r2dbc.postgresql.util.Operators;
-import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.SynchronousSink;
+import reactor.core.publisher.UnicastProcessor;
+import reactor.util.concurrent.Queues;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -163,7 +164,7 @@ private static Flux<BackendMessage> fetchAll(List<FrontendMessage.DirectEncoder>
      */
     private static Flux<BackendMessage> fetchCursoredWithSync(List<FrontendMessage.DirectEncoder> messagesToSend, Client client, String portal, int fetchSize) {
 
-        DirectProcessor<FrontendMessage> requestsProcessor = DirectProcessor.create();
+        UnicastProcessor<FrontendMessage> requestsProcessor = UnicastProcessor.create(Queues.<FrontendMessage>small().get());
         FluxSink<FrontendMessage> requestsSink = requestsProcessor.sink();
         AtomicBoolean isCanceled = new AtomicBoolean(false);
         AtomicBoolean done = new AtomicBoolean(false);
@@ -226,7 +227,7 @@ private static Flux<BackendMessage> fetchCursoredWithSync(List<FrontendMessage.D
      */
     private static Flux<BackendMessage> fetchCursoredWithFlush(List<FrontendMessage.DirectEncoder> messagesToSend, Client client, String portal, int fetchSize) {
 
-        DirectProcessor<FrontendMessage> requestsProcessor = DirectProcessor.create();
+        UnicastProcessor<FrontendMessage> requestsProcessor = UnicastProcessor.create(Queues.<FrontendMessage>small().get());
         FluxSink<FrontendMessage> requestsSink = requestsProcessor.sink();
         AtomicBoolean isCanceled = new AtomicBoolean(false);
 

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

@@ -21,7 +21,6 @@
 import io.netty.util.ReferenceCounted;
 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.CommandComplete;
 import io.r2dbc.postgresql.message.backend.ErrorResponse;
 import io.r2dbc.postgresql.message.backend.ParseComplete;
@@ -39,10 +38,11 @@
 import io.r2dbc.postgresql.message.frontend.Sync;
 import io.r2dbc.postgresql.util.Assert;
 import io.r2dbc.postgresql.util.Operators;
-import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.SynchronousSink;
+import reactor.core.publisher.UnicastProcessor;
+import reactor.util.concurrent.Queues;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -129,7 +129,7 @@ private static Flux<BackendMessage> fetchAll(Flux<FrontendMessage> bindFlow, Cli
      */
     private static Flux<BackendMessage> fetchCursored(Flux<FrontendMessage> bindFlow, Client client, String portal, int fetchSize) {
 
-        DirectProcessor<FrontendMessage> requestsProcessor = DirectProcessor.create();
+        UnicastProcessor<FrontendMessage> requestsProcessor = UnicastProcessor.create(Queues.<FrontendMessage>small().get());
         FluxSink<FrontendMessage> requestsSink = requestsProcessor.sink();
         AtomicBoolean isCanceled = new AtomicBoolean(false);