Leave drainLoop if loop cannot be entered

We now exit the drainLoop depending on whether the loop could be actually progress. Previously, the loop attempted to enter a guarded block that can be processed by a single thread only. If the drain loop block was already invoked by a different thread, then the parent caller of drainLoop would re-attempt to run the drainLoop until the subscriber had no demand or the buffer became empty. In case of recursion, a reentrant call to drainLoop could never succeed as the drain loop guard was engaged with the parent call.

[resolves #285]

Author: mp911de

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

@@ -799,9 +799,7 @@ public Flux<BackendMessage> addConversation(Predicate<BackendMessage> takeUntil,
         public void onRequest(Conversation conversation, long n) {
             conversation.incrementDemand(n);
 
-            while (hasBufferedItems() && hasDownstreamDemand()) {
-                drainLoop();
-            }
+            tryDrainLoop();
         }
 
         private void demandMore() {
@@ -852,45 +850,57 @@ public void onNext(BackendMessage message) {
                 return;
             }
 
+            tryDrainLoop();
+        }
+
+        private void tryDrainLoop() {
             while (hasBufferedItems() && hasDownstreamDemand()) {
-                this.drainLoop();
+                if (!this.drainLoop()) {
+                    return;
+                }
             }
         }
 
-        private void drainLoop() {
+        private boolean drainLoop() {
 
-            Conversation lastConversation = null;
+            if (!this.drain.compareAndSet(false, true)) {
+                return false;
+            }
 
-            if (this.drain.compareAndSet(false, true)) {
+            Conversation lastConversation = null;
 
-                try {
-                    while (hasBufferedItems()) {
+            try {
 
-                        Conversation conversation = this.conversations.peek();
-                        lastConversation = conversation;
-                        if (conversation == null) {
-                            break;
-                        }
+                while (hasBufferedItems()) {
 
-                        if (conversation.hasDemand()) {
+                    Conversation conversation = this.conversations.peek();
+                    lastConversation = conversation;
+                    if (conversation == null) {
+                        break;
+                    }
 
-                            BackendMessage item = this.buffer.poll();
+                    if (conversation.hasDemand()) {
 
-                            if (item == null) {
-                                break;
-                            }
+                        BackendMessage item = this.buffer.poll();
 
-                            emit(conversation, item);
-                        } else {
+                        if (item == null) {
                             break;
                         }
+
+                        emit(conversation, item);
+                    } else {
+                        break;
                     }
-                } finally {
-                    this.drain.compareAndSet(true, false);
                 }
+
+            } finally {
+                this.drain.compareAndSet(true, false);
             }
 
             potentiallyDemandMore(lastConversation);
+
+            return true;
+
         }
 
         private void potentiallyDemandMore(@Nullable Conversation lastConversation) {