Request more items when dropping tokens.

Signed-off-by: Mark Paluch <mpaluch@vmware.com>

Author: mp911de

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

@@ -75,6 +75,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -210,6 +211,7 @@ private ReactorNettyClient(Connection connection, TdsEncoder tdsEncoder, Connect
                 }
             };
 
+        AtomicReference<Subscription> subscriptionRef = new AtomicReference<>();
         SynchronousSink<Message> sink = new SynchronousSink<Message>() {
 
             @Override
@@ -250,6 +252,7 @@ public void next(Message message) {
                     ReactorNettyClient.this.featureAckChange.accept((FeatureExtAckToken) message);
                 }
 
+                Subscription subscription = subscriptionRef.get();
                 if (AbstractDoneToken.isAttentionAck(message)) {
 
                     long current;
@@ -260,6 +263,11 @@ public void next(Message message) {
                             if (DEBUG_ENABLED) {
                                 logger.debug(ReactorNettyClient.this.context.getMessage("Swallowing attention acknowledged, no pending requests: {}. "), message);
                             }
+
+                            // update demand for dropped next signal
+                            if (subscription != null) {
+                                subscription.request(1);
+                            }
                             return;
                         }
 
@@ -272,6 +280,10 @@ public void next(Message message) {
                     if (DEBUG_ENABLED) {
                         logger.debug(ReactorNettyClient.this.context.getMessage("Discard message {}. Draining frames until attention acknowledgement."), message);
                     }
+                    // update demand for dropped next signal
+                    if (subscription != null) {
+                        subscription.request(1);
+                    }
                     return;
                 }
 
@@ -304,6 +316,7 @@ public Context currentContext() {
 
                 @Override
                 public void onSubscribe(Subscription s) {
+                    subscriptionRef.set(s);
                     ReactorNettyClient.this.responseProcessor.onSubscribe(s);
                 }