Add support for Jackson's new ByteBufferFeeder (apple#1711)

Motivation:

In 2.14.0, Jackson added support for feeding ByteBuffers directly
into the streaming parser engine (where previously only byte arrays
would be supported). This changeset incorporates this enhancement
into the JacksonStreamingSerializer infrastructure.

Modifications:

Before this changeset, the ByteArrayParser (and corresponding feeder)
were the only choice when using the streaming deserialization
infrastructure in Jackson, so the code was able to eaglerly initialize
the parser and feeder.

Now there is a choice that can be made at runtime which parser might
be the best one based on the incoming buffer in the stream. As such,
the code now checks the first Buffer and then decides if either the
array-backed or the bytebuffer-backed parser should be initalized.

This also has the advantage that if no items are emitted on an empty
stream, no parser needs to be initialized at all.

Note that the code still preserves the runtime "backed by type" checks
since (while likely not common but possible) buffers with different
backing types can arrive one after another and need to be handled. In
this case it is possible that a sub-optimal parser is chosen, but the
code optimizes for the likely scenario that all buffers on one stream
are backed by the same type.

Result:

Added support for Jackson's new ByteBufferFeeder and choosing the
best strategy at runtime depending on the first buffer type that
arrives.

Author: daschl

gradle.properties

@@ -48,7 +48,7 @@ jerseyVersion=2.35
 
 reactiveStreamsVersion=1.0.4
 jcToolsVersion=3.3.1-ea
-jacksonVersion=2.13.4.20221013
+jacksonVersion=2.14.0-rc3
 # backward compatible with jackson 2.9+, we do not depend on any new features from later versions.
 
 openTracingVersion=0.33.0

servicetalk-data-jackson/src/main/java/io/servicetalk/data/jackson/JacksonStreamingSerializer.java

@@ -15,6 +15,7 @@
  */
 package io.servicetalk.data.jackson;
 
+import com.fasterxml.jackson.core.async.NonBlockingInputFeeder;
 import io.servicetalk.buffer.api.Buffer;
 import io.servicetalk.buffer.api.BufferAllocator;
 import io.servicetalk.concurrent.PublisherSource.Subscriber;
@@ -29,7 +30,6 @@
 import com.fasterxml.jackson.core.JsonToken;
 import com.fasterxml.jackson.core.async.ByteArrayFeeder;
 import com.fasterxml.jackson.core.async.ByteBufferFeeder;
-import com.fasterxml.jackson.core.async.NonBlockingInputFeeder;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.JsonNode;
@@ -101,37 +101,58 @@ private DeserializeOperator(ObjectReader reader) {
 
         @Override
         public Subscriber<? super Buffer> apply(final Subscriber<? super Iterable<T>> subscriber) {
-            final JsonParser parser;
-            try {
-                // TODO(scott): ByteBufferFeeder is currently not supported by jackson, and the current API throws
-                // UnsupportedOperationException if not supported. When jackson does support two NonBlockingInputFeeder
-                // types we need an approach which doesn't involve catching UnsupportedOperationException to try to get
-                // ByteBufferFeeder and then ByteArrayFeeder.
-                parser = reader.getFactory().createNonBlockingByteArrayParser();
-            } catch (IOException e) {
-                throw new SerializationException(e);
-            }
-            NonBlockingInputFeeder feeder = parser.getNonBlockingInputFeeder();
-            if (feeder instanceof ByteBufferFeeder) {
-                return new ByteBufferDeserializeSubscriber<>(subscriber, reader, parser, (ByteBufferFeeder) feeder);
-            } else if (feeder instanceof ByteArrayFeeder) {
-                return new ByteArrayDeserializeSubscriber<>(subscriber, reader, parser, (ByteArrayFeeder) feeder);
-            }
-            return new FailedSubscriber<>(subscriber, new SerializationException("unsupported feeder type: " + feeder));
+            return new DeserializeSubscriber<>(subscriber, reader);
         }
 
-        private static final class ByteArrayDeserializeSubscriber<T> extends DeserializeSubscriber<T> {
-            private final ByteArrayFeeder feeder;
+        private static class DeserializeSubscriber<T> implements Subscriber<Buffer> {
+            private final ObjectReader reader;
+            private final Deque<JsonNode> tokenStack = new ArrayDeque<>(8);
+            private final Subscriber<? super Iterable<T>> subscriber;
+            @Nullable
+            private Subscription subscription;
+            @Nullable
+            private String fieldName;
+
+            @Nullable
+            private JsonParser parser;
+            @Nullable
+            private NonBlockingInputFeeder feeder;
+
+            private boolean isByteArrayParser;
+
+            private DeserializeSubscriber(final Subscriber<? super Iterable<T>> subscriber, final ObjectReader reader) {
+                this.reader = reader;
+                this.subscriber = subscriber;
+            }
+
+            /**
+             * Consumer the buffer from {@link #onNext(Buffer)}.
+             * @param buffer The bytes to append.
+             * @return {@code true} if more data is required to parse an object. {@code false} if object(s) should be
+             * parsed after this method returns.
+             * @throws IOException If an exception occurs while appending {@link Buffer}.
+             */
+            private boolean consumeOnNext(final Buffer buffer) throws IOException {
+                if (feeder == null) {
+                    throw new NullPointerException("The NonBlockingInputFeeder is null when it should not be " +
+                      "- this is a bug.");
+                }
 
-            private ByteArrayDeserializeSubscriber(final Subscriber<? super Iterable<T>> subscriber,
-                                                   final ObjectReader reader, final JsonParser parser,
-                                                   final ByteArrayFeeder feeder) {
-                super(subscriber, reader, parser);
-                this.feeder = feeder;
+                if (isByteArrayParser) {
+                    feedByteArrayParser(buffer, (ByteArrayFeeder) feeder);
+                } else {
+                    feedByteBufferParser(buffer, (ByteBufferFeeder) feeder);
+                }
+                return feeder.needMoreInput();
             }
 
-            @Override
-            boolean consumeOnNext(final Buffer buffer) throws IOException {
+            /**
+             * Feeds the buffer into the {@link ByteArrayFeeder}.
+             * @param buffer the buffer to feed into the streaming JSON parser.
+             * @param feeder the feeder into which the buffer should be fed.
+             * @throws IOException if feeding the buffer failed.
+             */
+            private void feedByteArrayParser(final Buffer buffer, final ByteArrayFeeder feeder) throws IOException {
                 if (buffer.hasArray()) {
                     final int start = buffer.arrayOffset() + buffer.readerIndex();
                     feeder.feedInput(buffer.array(), start, start + buffer.readableBytes());
@@ -143,53 +164,33 @@ boolean consumeOnNext(final Buffer buffer) throws IOException {
                         feeder.feedInput(copy, 0, copy.length);
                     }
                 }
-                return feeder.needMoreInput();
-            }
-        }
-
-        private static final class ByteBufferDeserializeSubscriber<T> extends DeserializeSubscriber<T> {
-            private final ByteBufferFeeder feeder;
-
-            private ByteBufferDeserializeSubscriber(final Subscriber<? super Iterable<T>> subscriber,
-                                                    final ObjectReader reader, final JsonParser parser,
-                                                    final ByteBufferFeeder feeder) {
-                super(subscriber, reader, parser);
-                this.feeder = feeder;
             }
 
-            @Override
-            boolean consumeOnNext(final Buffer buffer) throws IOException {
+            /**
+             * Feeds the buffer into the {@link ByteBufferFeeder}.
+             * @param buffer the buffer to feed into the streaming JSON parser.
+             * @param feeder the feeder into which the buffer should be fed.
+             * @throws IOException if feeding the buffer failed.
+             */
+            private void feedByteBufferParser(final Buffer buffer, final ByteBufferFeeder feeder) throws IOException {
                 feeder.feedInput(buffer.toNioBuffer());
-                return feeder.needMoreInput();
-            }
-        }
-
-        private abstract static class DeserializeSubscriber<T> implements Subscriber<Buffer> {
-            private final JsonParser parser;
-            private final ObjectReader reader;
-            private final Deque<JsonNode> tokenStack = new ArrayDeque<>(8);
-            private final Subscriber<? super Iterable<T>> subscriber;
-            @Nullable
-            private Subscription subscription;
-            @Nullable
-            private String fieldName;
-
-            private DeserializeSubscriber(final Subscriber<? super Iterable<T>> subscriber,
-                                          final ObjectReader reader,
-                                          final JsonParser parser) {
-                this.reader = reader;
-                this.parser = parser;
-                this.subscriber = subscriber;
             }
 
             /**
-             * Consumer the buffer from {@link #onNext(Buffer)}.
-             * @param buffer The bytes to append.
-             * @return {@code true} if more data is required to parse an object. {@code false} if object(s) should be
-             * parsed after this method returns.
-             * @throws IOException If an exception occurs while appending {@link Buffer}.
+             * Initializes the JSON streaming parser and feeder based on the buffer type.
+             * @param arrayBackedBuffer if the buffer on the operator is backed by an array.
+             * @throws IOException if creating the non-blocking {@link JsonParser} failed.
              */
-            abstract boolean consumeOnNext(Buffer buffer) throws IOException;
+            private void initParserAndFeeder(final boolean arrayBackedBuffer) throws IOException {
+                if (arrayBackedBuffer) {
+                    parser = reader.getFactory().createNonBlockingByteArrayParser();
+                    isByteArrayParser = true;
+                } else {
+                    parser = reader.getFactory().createNonBlockingByteBufferParser();
+                    isByteArrayParser = false;
+                }
+                feeder = parser.getNonBlockingInputFeeder();
+            }
 
             @Override
             public final void onSubscribe(final Subscription subscription) {
@@ -201,6 +202,10 @@ public final void onSubscribe(final Subscription subscription) {
             public final void onNext(@Nullable final Buffer buffer) {
                 assert subscription != null;
                 try {
+                    if (parser == null && buffer != null) {
+                        initParserAndFeeder(buffer.hasArray());
+                    }
+
                     if (buffer == null || consumeOnNext(buffer)) {
                         subscription.request(1);
                     } else {
@@ -238,11 +243,17 @@ public final void onNext(@Nullable final Buffer buffer) {
 
             @Override
             public final void onError(final Throwable t) {
+                if (feeder != null) {
+                    feeder.endOfInput();
+                }
                 subscriber.onError(t);
             }
 
             @Override
             public final void onComplete() {
+                if (feeder != null) {
+                    feeder.endOfInput();
+                }
                 if (tokenStack.isEmpty()) {
                     subscriber.onComplete();
                 } else {

servicetalk-data-jackson/src/test/java/io/servicetalk/data/jackson/JacksonSerializerFactoryTest.java

@@ -22,10 +22,12 @@
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import java.util.Arrays;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.stream.Stream;