Leave Client drainloop if current conversation has no demand

The drainloop is now left if the current conversation has no demand.

[#242]

Author: mp911de

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

@@ -761,7 +761,10 @@ public Flux<BackendMessage> addConversation(Predicate<BackendMessage> takeUntil,
 
         public void onRequest(Conversation conversation, long n) {
             conversation.incrementDemand(n);
-            drainLoop();
+
+            while (hasBufferedItems() && hasDownstreamDemand()) {
+                drainLoop();
+            }
         }
 
         private void demandMore() {
@@ -776,6 +779,13 @@ public void onSubscribe(Subscription s) {
             this.demandMore();
         }
 
+        private boolean hasDownstreamDemand() {
+
+            Conversation conversation = this.conversations.peek();
+
+            return conversation != null && conversation.hasDemand();
+        }
+
         @Override
         public void onNext(BackendMessage message) {
 
@@ -788,11 +798,15 @@ public void onNext(BackendMessage message) {
             this.demand.decrementAndGet();
 
             if (!this.buffer.offer(message)) {
+                ReferenceCountUtil.release(message);
+                Operators.onNextDropped(message, currentContext());
                 onError(new ResponseQueueException("Response queue is full"));
                 return;
             }
 
-            drainLoop();
+            while (hasBufferedItems() && hasDownstreamDemand()) {
+                this.drainLoop();
+            }
         }
 
         private void drainLoop() {
@@ -802,7 +816,7 @@ private void drainLoop() {
             if (this.drain.compareAndSet(false, true)) {
 
                 try {
-                    while (!this.buffer.isEmpty()) {
+                    while (hasBufferedItems()) {
 
                         Conversation conversation = this.conversations.peek();
                         lastConversation = conversation;
@@ -818,25 +832,38 @@ private void drainLoop() {
                                 break;
                             }
 
-                            if (conversation.canComplete(item)) {
-                                this.conversations.poll();
-                                conversation.complete(item);
-                            } else {
-                                conversation.emit(item);
-                            }
+                            emit(conversation, item);
+                        } else {
+                            break;
                         }
                     }
                 } finally {
                     this.drain.compareAndSet(true, false);
                 }
             }
 
+            potentiallyDemandMore(lastConversation);
+        }
 
+        private void potentiallyDemandMore(@Nullable Conversation lastConversation) {
             if (lastConversation == null || lastConversation.hasDemand() || lastConversation.isCancelled()) {
                 this.demandMore();
             }
         }
 
+        private void emit(Conversation conversation, BackendMessage item) {
+            if (conversation.canComplete(item)) {
+                this.conversations.poll();
+                conversation.complete(item);
+            } else {
+                conversation.emit(item);
+            }
+        }
+
+        private boolean hasBufferedItems() {
+            return !this.buffer.isEmpty();
+        }
+
         @Override
         public void onError(Throwable throwable) {