New Serializer APIs, consolidation of ContentCodec, and gRPC MethodDescriptor (#1673)

Motivation:
Supporting a custom transport for gRPC currently requires
reflection and special knowledge of types for serialization. The
generated code has information related to types and serialization that
can be exposed to the runtime which also allows supporting custom
transports.

This task requires exposing serializer APIs into the public
API. Our existing serializer APIs were heavily influced by streaming use
cases and JSON serialization which can serialize any type and generally
doesn't have type restrictions. This lead to a general
SerializationProvider API which is impossible to constrain types due to
the generics being at the method level, which doesn't account for
implementing serialization for type constrained approached (protobuf
MessageLite). Also the serialization APIs couple scalar and streaming
use cases which in practice are more commonly decoupled into scalar
(individual objects) and streaming (which applies some framing around
individual objects). These concepts are coupled which leads to confusing
specializations (deserializeAggregatedSingle, deserializeAggregated)
which are easy to misuse and lead to invalid serialization in practice
(streaming {form url, string} encoding are not properly framed and may
not be deserialized correctly). The existing Serializer APIs also do not
promote composability in that if you can serialize one object you should
be able to re-use that serialization for a stream of objects with some
frameing applied on top (required for gRPC).

Modifications:
- Introduce a new servicetalk-serializer-api package with new
  Serializer APIs.
- Introduce a new servicetalk-serializer-utils package with common
  utilities for serialization such as streaming framing (fixed and
  varint length prefixed)
- Introduce BufferEncoder APIs in servicetalk-encoding-api for
  compression and decompression. This also comes with
  NettyBufferEncoders and NettyCompression to create Netty backed
  implementations.
- Introduce HttpStreamingSerializer and new HttpSerializer APIs in
  servicetalk-http-api, and modify all request/response APIs to use the
  new serializer APIs. Introduce new methods to deserialize and process
  trailers.
- Introduce HttpSerializers in servicetalk-http-api for implementations
  of the new APIs.
- Introduce JacksonSerializerCache in servicetalk-http-api as the new
  entry point to create/cache serializers for JSON.
- Introduce ProtobufSerializerCache in servicetalk-data-protobuf as the
  new entry point to create/cache serializers for protobuf.
- Introduce MethodDescriptor API into servicetalk-grpc-api and update
  code generation to use this new API.
- Update all examples avoid usage of deprecated APIs and leverage new
  APIs.
- Deprecate the following APIs and related methods in favor of the new
  types descrbied above:
  - SerializationProvider, JacksonSerializationProvider,
    ProtobufSerializationProvider, GrpcSerializationProvider,
    HttpSerializationProvider
  - ContentCodec, CodecDecodingException, CodecEncodingException,
    ContentCodings
  - GrpcMetadata#path(), code generated types that extend
    DefaultGrpcClientMetadata

Result:
gRPC API supports MethodDescriptor, which uses new Serializer APIs, and
the entire code base has been updated to use the new APIs consistently.

Author: Scottmitch

servicetalk-buffer-api/src/main/java/io/servicetalk/buffer/api/Buffer.java

@@ -1657,6 +1657,21 @@ default boolean tryEnsureWritable(int minWritableBytes, boolean force) {
      */
     Buffer writeUtf8(CharSequence seq, int ensureWritable);
 
+    /**
+     * Encode a {@link CharSequence} encoded in {@link Charset} and write it to this buffer starting at
+     * {@code writerIndex} and increases the {@code writerIndex} by the number of the transferred bytes.
+     *
+     * @param seq the source of the data.
+     * @param charset the charset used for encoding.
+     * @return self.
+     * @throws ReadOnlyBufferException if this buffer is read-only
+     */
+    default Buffer writeCharSequence(CharSequence seq, Charset charset) {
+        byte[] bytes = seq.toString().getBytes(charset);
+        writeBytes(bytes);
+        return this;
+    }
+
     /**
      * Locates the first occurrence of the specified {@code value} in this
      * buffer.  The search takes place from the specified {@code fromIndex}

servicetalk-buffer-api/src/main/java/io/servicetalk/buffer/api/EmptyBuffer.java

@@ -644,6 +644,11 @@ public Buffer writeUtf8(CharSequence seq, int ensureWritable) {
         throw new IndexOutOfBoundsException();
     }
 
+    @Override
+    public Buffer writeCharSequence(CharSequence seq, Charset charset) {
+        throw new IndexOutOfBoundsException();
+    }
+
     @Override
     public int indexOf(int fromIndex, int toIndex, byte value) {
         checkIndex(fromIndex);

servicetalk-buffer-api/src/main/java/io/servicetalk/buffer/api/ReadOnlyByteBuffer.java

@@ -375,6 +375,11 @@ public Buffer writeUtf8(CharSequence seq, int ensureWritable) {
         throw new ReadOnlyBufferException();
     }
 
+    @Override
+    public Buffer writeCharSequence(CharSequence seq, Charset charset) {
+        throw new ReadOnlyBufferException();
+    }
+
     @Override
     public Buffer readSlice(int length) {
         checkReadableBytes0(length);

servicetalk-buffer-netty/src/main/java/io/servicetalk/buffer/netty/NettyBuffer.java

@@ -710,6 +710,12 @@ public Buffer writeUtf8(CharSequence seq, int ensureWritable) {
         return this;
     }
 
+    @Override
+    public Buffer writeCharSequence(CharSequence seq, Charset charset) {
+        buffer.writeCharSequence(seq, charset);
+        return this;
+    }
+
     @Override
     public int indexOf(int fromIndex, int toIndex, byte value) {
         return buffer.indexOf(fromIndex, toIndex, value);

servicetalk-buffer-netty/src/main/java/io/servicetalk/buffer/netty/NettyCompositeBuffer.java

@@ -21,6 +21,7 @@
 import io.netty.buffer.CompositeByteBuf;
 
 import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
 
 final class NettyCompositeBuffer extends NettyBuffer<CompositeByteBuf> implements CompositeBuffer {
 
@@ -363,4 +364,10 @@ public CompositeBuffer writeUtf8(CharSequence seq, int ensureWritable) {
         super.writeUtf8(seq, ensureWritable);
         return this;
     }
+
+    @Override
+    public CompositeBuffer writeCharSequence(CharSequence seq, Charset charset) {
+        super.writeCharSequence(seq, charset);
+        return this;
+    }
 }

servicetalk-buffer-netty/src/main/java/io/servicetalk/buffer/netty/ReadOnlyBuffer.java

@@ -20,6 +20,7 @@
 import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.nio.ReadOnlyBufferException;
+import java.nio.charset.Charset;
 
 final class ReadOnlyBuffer extends WrappedBuffer {
 
@@ -302,6 +303,11 @@ public Buffer writeUtf8(CharSequence seq, int ensureWritable) {
         throw new ReadOnlyBufferException();
     }
 
+    @Override
+    public Buffer writeCharSequence(CharSequence seq, Charset charset) {
+        throw new ReadOnlyBufferException();
+    }
+
     @Override
     public Buffer readSlice(int length) {
         return buffer.readSlice(length).asReadOnly();

servicetalk-buffer-netty/src/main/java/io/servicetalk/buffer/netty/WrappedBuffer.java

@@ -658,6 +658,12 @@ public Buffer writeUtf8(CharSequence seq, int ensureWritable) {
         return this;
     }
 
+    @Override
+    public Buffer writeCharSequence(CharSequence seq, Charset charset) {
+        buffer.writeCharSequence(seq, charset);
+        return this;
+    }
+
     @Override
     public int indexOf(int fromIndex, int toIndex, byte value) {
         return buffer.indexOf(fromIndex, toIndex, value);

servicetalk-concurrent-reactivestreams/build.gradle

@@ -21,6 +21,8 @@ dependencies {
   api "org.reactivestreams:reactive-streams:$reactiveStreamsVersion"
 
   implementation project(":servicetalk-annotations")
+  implementation project(":servicetalk-serializer-utils")
+  implementation project(":servicetalk-buffer-netty")
   implementation "com.google.code.findbugs:jsr305:$jsr305Version"
   implementation "org.slf4j:slf4j-api:$slf4jVersion"
 

servicetalk-concurrent-reactivestreams/src/test/java/io/servicetalk/concurrent/reactivestreams/tck/FramedDeserializerOperatorTckTest.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright © 2021 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.servicetalk.concurrent.reactivestreams.tck;
+
+import io.servicetalk.concurrent.api.Publisher;
+import io.servicetalk.serializer.utils.FramedDeserializerOperator;
+
+import org.testng.annotations.Test;
+
+import static io.servicetalk.buffer.netty.BufferAllocators.DEFAULT_ALLOCATOR;
+import static java.util.function.Function.identity;
+
+@Test
+public class FramedDeserializerOperatorTckTest extends AbstractPublisherTckTest<Integer> {
+    @Override
+    public Publisher<Integer> createServiceTalkPublisher(final long elements) {
+        return Publisher.range(0, TckUtils.requestNToInt(elements))
+                .map(i -> DEFAULT_ALLOCATOR.newBuffer().writeInt(i))
+                .liftSync(new FramedDeserializerOperator<>(
+                        (serializedData, allocator) -> serializedData.readInt(),
+                        () -> (buffer, bufferAllocator) ->
+                                buffer.readableBytes() < Integer.BYTES ? null : buffer.readBytes(Integer.BYTES),
+                        DEFAULT_ALLOCATOR))
+                .flatMapConcatIterable(identity());
+    }
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return TckUtils.maxElementsFromPublisher();
+    }
+}

servicetalk-data-jackson-jersey/src/main/java/io/servicetalk/data/jackson/jersey/JacksonSerializationProviderContextResolver.java

@@ -20,6 +20,10 @@
 import javax.annotation.Nullable;
 import javax.ws.rs.ext.ContextResolver;
 
+/**
+ * @deprecated Use {@link JacksonSerializerFactoryContextResolver}.
+ */
+@Deprecated
 final class JacksonSerializationProviderContextResolver implements ContextResolver<JacksonSerializationProvider> {
     private final JacksonSerializationProvider jacksonSerializationProvider;
 

servicetalk-data-jackson-jersey/src/main/java/io/servicetalk/data/jackson/jersey/JacksonSerializerFactoryContextResolver.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright © 2021 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.servicetalk.data.jackson.jersey;
+
+import io.servicetalk.data.jackson.JacksonSerializerFactory;
+
+import javax.annotation.Nullable;
+import javax.ws.rs.ext.ContextResolver;
+
+import static java.util.Objects.requireNonNull;
+
+final class JacksonSerializerFactoryContextResolver implements ContextResolver<JacksonSerializerFactory> {
+    private final JacksonSerializerFactory factory;
+
+    JacksonSerializerFactoryContextResolver(final JacksonSerializerFactory factory) {
+        this.factory = requireNonNull(factory);
+    }
+
+    @Nullable
+    @Override
+    public JacksonSerializerFactory getContext(final Class<?> aClass) {
+        if (!JacksonSerializerFactory.class.isAssignableFrom(aClass)) {
+            return null;
+        }
+
+        return factory;
+    }
+}

servicetalk-data-jackson-jersey/src/main/java/io/servicetalk/data/jackson/jersey/JacksonSerializerMessageBodyReaderWriter.java

@@ -19,12 +19,13 @@
 import io.servicetalk.buffer.api.BufferAllocator;
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.concurrent.api.Single;
-import io.servicetalk.data.jackson.JacksonSerializationProvider;
+import io.servicetalk.data.jackson.JacksonSerializerFactory;
 import io.servicetalk.http.router.jersey.internal.SourceWrappers.PublisherSource;
 import io.servicetalk.http.router.jersey.internal.SourceWrappers.SingleSource;
-import io.servicetalk.serialization.api.DefaultSerializer;
-import io.servicetalk.serialization.api.SerializationException;
-import io.servicetalk.serialization.api.Serializer;
+import io.servicetalk.serializer.api.Deserializer;
+import io.servicetalk.serializer.api.SerializationException;
+import io.servicetalk.serializer.api.Serializer;
+import io.servicetalk.serializer.api.StreamingSerializer;
 import io.servicetalk.transport.api.ConnectionContext;
 import io.servicetalk.transport.api.ExecutionContext;
 
@@ -68,9 +69,6 @@
 @Consumes(WILDCARD)
 @Produces(WILDCARD)
 final class JacksonSerializerMessageBodyReaderWriter implements MessageBodyReader<Object>, MessageBodyWriter<Object> {
-    private static final JacksonSerializationProvider DEFAULT_JACKSON_SERIALIZATION_PROVIDER =
-            new JacksonSerializationProvider();
-
     // We can not use `@Context ConnectionContext` directly because we would not see the latest version
     // in case it has been rebound as part of offloading.
     @Context
@@ -95,86 +93,81 @@ public boolean isReadable(final Class<?> type, final Type genericType, final Ann
     public Object readFrom(final Class<Object> type, final Type genericType, final Annotation[] annotations,
                            final MediaType mediaType, final MultivaluedMap<String, String> httpHeaders,
                            final InputStream entityStream) throws WebApplicationException {
-
-        final Serializer serializer = getSerializer(mediaType);
+        final JacksonSerializerFactory serializerFactory = getJacksonSerializerFactory(mediaType);
         final ExecutionContext executionContext = ctxRefProvider.get().get().executionContext();
         final BufferAllocator allocator = executionContext.bufferAllocator();
         final int contentLength = requestCtxProvider.get().getLength();
 
         if (Single.class.isAssignableFrom(type)) {
             return handleEntityStream(entityStream, allocator,
-                    (p, a) -> deserialize(p, serializer, getSourceClass(genericType), contentLength, a),
-                    (is, a) -> new SingleSource<>(deserialize(toBufferPublisher(is, a), serializer,
-                            getSourceClass(genericType), contentLength, a)));
+                    (p, a) -> deserialize(p, serializerFactory.serializerDeserializer(getSourceClass(genericType)),
+                            contentLength, a),
+                    (is, a) -> new SingleSource<>(deserialize(toBufferPublisher(is, a),
+                            serializerFactory.serializerDeserializer(getSourceClass(genericType)), contentLength, a)));
         } else if (Publisher.class.isAssignableFrom(type)) {
             return handleEntityStream(entityStream, allocator,
-                    (p, a) -> serializer.deserialize(p, getSourceClass(genericType)),
-                    (is, a) -> new PublisherSource<>(serializer.deserialize(toBufferPublisher(is, a),
-                            getSourceClass(genericType))));
+                    (p, a) -> serializerFactory.streamingSerializerDeserializer(
+                            getSourceClass(genericType)).deserialize(p, a),
+                    (is, a) -> new PublisherSource<>(serializerFactory.streamingSerializerDeserializer(
+                            getSourceClass(genericType)).deserialize(toBufferPublisher(is, a), a)));
         }
 
         return handleEntityStream(entityStream, allocator,
-                (p, a) -> deserializeObject(p, serializer, type, contentLength, a),
-                (is, a) -> deserializeObject(toBufferPublisher(is, a), serializer, type, contentLength, a));
+                (p, a) -> deserializeObject(p, serializerFactory.serializerDeserializer(type), contentLength, a),
+                (is, a) -> deserializeObject(toBufferPublisher(is, a), serializerFactory.serializerDeserializer(type),
+                        contentLength, a));
     }
 
     @Override
     public boolean isWriteable(final Class<?> type, final Type genericType, final Annotation[] annotations,
                                final MediaType mediaType) {
-
         return !isSse(requestCtxProvider.get()) && isSupportedMediaType(mediaType);
     }
 
+    @SuppressWarnings({"rawtypes", "unchecked"})
     @Override
     public void writeTo(final Object o, final Class<?> type, final Type genericType, final Annotation[] annotations,
                         final MediaType mediaType, final MultivaluedMap<String, Object> httpHeaders,
                         final OutputStream entityStream) throws WebApplicationException {
-
+        final BufferAllocator allocator = ctxRefProvider.get().get().executionContext().bufferAllocator();
         final Publisher<Buffer> bufferPublisher;
         if (o instanceof Single) {
-            bufferPublisher = getResponseBufferPublisher(((Single) o).toPublisher(), genericType, mediaType);
+            final Class<?> clazz = genericType instanceof Class ? (Class) genericType : getSourceClass(genericType);
+            Serializer serializer = getJacksonSerializerFactory(mediaType).serializerDeserializer(clazz);
+            bufferPublisher = ((Single) o).map(t -> serializer.serialize(t, allocator)).toPublisher();
         } else if (o instanceof Publisher) {
-            bufferPublisher = getResponseBufferPublisher((Publisher) o, genericType, mediaType);
+            final Class<?> clazz = genericType instanceof Class ? (Class) genericType : getSourceClass(genericType);
+            StreamingSerializer serializer = getJacksonSerializerFactory(mediaType)
+                    .streamingSerializerDeserializer(clazz);
+            bufferPublisher = serializer.serialize((Publisher) o, allocator);
         } else {
-            bufferPublisher = getResponseBufferPublisher(Publisher.from(o), o.getClass(), mediaType);
+            Serializer serializer = getJacksonSerializerFactory(mediaType).serializerDeserializer(o.getClass());
+            bufferPublisher = Publisher.from(serializer.serialize(o, allocator));
         }
 
         setResponseBufferPublisher(bufferPublisher, requestCtxProvider.get());
     }
 
-    @SuppressWarnings("unchecked")
-    private Publisher<Buffer> getResponseBufferPublisher(final Publisher publisher, final Type type,
-                                                         final MediaType mediaType) {
-        final BufferAllocator allocator = ctxRefProvider.get().get().executionContext().bufferAllocator();
-        return getSerializer(mediaType).serialize(publisher, allocator,
-                type instanceof Class ? (Class) type : getSourceClass(type));
-    }
-
-    private Serializer getSerializer(final MediaType mediaType) {
-        return new DefaultSerializer(getJacksonSerializationProvider(mediaType));
-    }
+    private JacksonSerializerFactory getJacksonSerializerFactory(final MediaType mediaType) {
+        final ContextResolver<JacksonSerializerFactory> contextResolver =
+                providers.getContextResolver(JacksonSerializerFactory.class, mediaType);
 
-    private JacksonSerializationProvider getJacksonSerializationProvider(final MediaType mediaType) {
-        final ContextResolver<JacksonSerializationProvider> contextResolver =
-                providers.getContextResolver(JacksonSerializationProvider.class, mediaType);
-
-        return contextResolver != null ? contextResolver.getContext(JacksonSerializationProvider.class) :
-                DEFAULT_JACKSON_SERIALIZATION_PROVIDER;
+        return contextResolver != null ? contextResolver.getContext(JacksonSerializerFactory.class) :
+                JacksonSerializerFactory.JACKSON;
     }
 
     private static Publisher<Buffer> toBufferPublisher(final InputStream is, final BufferAllocator a) {
         return fromInputStream(is).map(a::wrap);
     }
 
-    private static <T> Single<T> deserialize(final Publisher<Buffer> bufferPublisher, final Serializer ser,
-                                             final Class<T> type, final int contentLength,
-                                             final BufferAllocator allocator) {
-
+    private static <T> Single<T> deserialize(
+            final Publisher<Buffer> bufferPublisher, final Deserializer<T> deserializer, final int contentLength,
+            final BufferAllocator allocator) {
         return bufferPublisher
                 .collect(() -> newBufferForRequestContent(contentLength, allocator), Buffer::writeBytes)
                 .map(buf -> {
                     try {
-                        return ser.deserializeAggregatedSingle(buf, type);
+                        return deserializer.deserialize(buf, allocator);
                     } catch (final NoSuchElementException e) {
                         throw new BadRequestException("No deserializable JSON content", e);
                     } catch (final SerializationException e) {
@@ -192,10 +185,9 @@ static Buffer newBufferForRequestContent(final int contentLength,
     }
 
     // visible for testing
-    static <T> T deserializeObject(final Publisher<Buffer> bufferPublisher, final Serializer ser,
-                                   final Class<T> type, final int contentLength,
-                                   final BufferAllocator allocator) {
-        return awaitResult(deserialize(bufferPublisher, ser, type, contentLength, allocator).toFuture());
+    static <T> T deserializeObject(final Publisher<Buffer> bufferPublisher, final Deserializer<T> deserializer,
+                                   final int contentLength, final BufferAllocator allocator) {
+        return awaitResult(deserialize(bufferPublisher, deserializer, contentLength, allocator).toFuture());
     }
 
     private static boolean isSse(ContainerRequestContext requestCtx) {