Refactor decode pipeline to use Reactor operators.

ReactorNettyClient's inbound decoder now decodes items into Iterables and uses concatMapIterable(…) to correctly determine demand vs. the number of produced elements.

Previously, the decoder used a onNext hook that propagated items into the result processor. That could lead to demand mismatch as the inbound channel could had emitted chunked buffers that couldn't be decoded into messages. That would lead to a mismatch between demand and emitted items because the emitter byte buffer decremented the demand by 1 while the receiving side didn't yield any results. Eventually, the publisher stopped emission because it had emitted buffers that couldn't be decoded while the pipeline didn't account for requesting additional buffers to fully decode the buffers.

Thanks to @gvisokinskas who helped a lot to investigate on this issue.

[resolves #216]

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

Author: mp911de

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

@@ -68,6 +68,7 @@
 import javax.annotation.Nullable;
 import java.security.GeneralSecurityException;
 import java.time.Duration;
+import java.util.Collections;
 import java.util.Optional;
 import java.util.Queue;
 import java.util.UUID;
@@ -279,24 +280,22 @@ public void next(Message message) {
         };
 
         connection.inbound().receiveObject() //
-            .doOnNext(it -> {
+            .concatMapIterable(it -> {
 
                 if (it instanceof ByteBuf) {
 
                     ByteBuf buffer = (ByteBuf) it;
-                    decoder.decode(buffer, this.decodeFunction, sink);
-                    return;
+                    return decoder.decode(buffer, this.decodeFunction);
                 }
 
                 if (it instanceof Message) {
-                    sink.next((Message) it);
-                    return;
+                    return Collections.singleton((Message) it);
                 }
 
                 throw ProtocolException.unsupported(String.format("Unexpected protocol message: [%s]", it));
             })
             .onErrorResume(this::resumeError)
-            .subscribe(new CoreSubscriber<Object>() {
+            .subscribe(new CoreSubscriber<Message>() {
 
                 @Override
                 public Context currentContext() {
@@ -309,7 +308,8 @@ public void onSubscribe(Subscription s) {
                 }
 
                 @Override
-                public void onNext(Object message) {
+                public void onNext(Message message) {
+                    sink.next(message);
                 }
 
                 @Override

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

@@ -34,7 +34,7 @@
  * <p/>
  * TDS messages consist of a header ({@link Header#LENGTH 8 byte length}) and a body. Messages can be either self-contained ({@link Status.StatusBit#EOM}) or chunked.  This decoder attempts to
  * decode messages from a {@link ByteBuf stream} by emitting zero, one or many {@link Message}s. Data buffers are aggregated and de-chunked until reaching a message boundary, then adaptive decoding
- * attempts to decode the aggregated and de-chunked body as far as possible. Remaining (undecoded) data buffers are aggregated until the next attempt.
+ * attempts to decode the aggregated and de-chunked body as far as possible. Remaining (non-decodable) data buffers are aggregated until the next attempt.
  * <p/>
  * This decoder is stateful and should be used in a try-to-decode fashion.
  *
@@ -59,30 +59,9 @@ public List<Message> decode(ByteBuf in, MessageDecoder messageDecoder) {
         Assert.requireNonNull(in, "in must not be null");
         Assert.requireNonNull(messageDecoder, "MessageDecoder must not be null");
 
-        List<Message> result = new ArrayList<>();
+        ListSink<Message> result = new ListSink<>();
 
-        decode(in, messageDecoder, new SynchronousSink<Message>() {
-
-            @Override
-            public void complete() {
-                throw new UnsupportedOperationException();
-            }
-
-            @Override
-            public Context currentContext() {
-                throw new UnsupportedOperationException();
-            }
-
-            @Override
-            public void error(Throwable e) {
-                throw new RuntimeException(e);
-            }
-
-            @Override
-            public void next(Message message) {
-                result.add(message);
-            }
-        });
+        decode(in, messageDecoder, result);
 
         return result;
     }
@@ -350,4 +329,32 @@ int getChunkLength() {
 
     }
 
+    static class ListSink<T> extends ArrayList<T> implements SynchronousSink<T> {
+
+        public ListSink() {
+            super(2);
+        }
+
+        @Override
+        public void complete() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Context currentContext() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void error(Throwable e) {
+            throw new RuntimeException(e);
+        }
+
+        @Override
+        public void next(T message) {
+            add(message);
+        }
+
+    }
+
 }

src/test/java/io/r2dbc/mssql/LobIntegrationTests.java

@@ -28,8 +28,10 @@
 
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
 import java.util.Optional;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -116,6 +118,36 @@ void testBigBlob() {
             .verifyComplete();
     }
 
+    @Test
+    void testBigBlobs() {
+
+        int count = 512000;
+        createTable(connection, "NVARCHAR(MAX)");
+
+        CharBuffer chars = CharBuffer.allocate(count);
+        IntStream.range(0, count).forEach(i -> chars.put((char) ((i % 26) + 'a')));
+        chars.flip();
+        String data = chars.toString();
+
+        for (int i = 0; i < 30; i++) {
+            Flux.from(connection.createStatement("INSERT INTO lob_test values(@P0)")
+                .bind("P0", data)
+                .execute())
+                .flatMap(Result::getRowsUpdated)
+                .as(StepVerifier::create)
+                .expectNext(1)
+                .verifyComplete();
+
+        }
+
+        connection.createStatement("SELECT my_col FROM lob_test")
+            .execute()
+            .concatMap(it -> it.map((row, rowMetadata) -> row.get("my_col")))
+            .as(StepVerifier::create)
+            .expectNextCount(30)
+            .verifyComplete();
+    }
+
     @Test
     void testByteArrayBlob() {