Add kotlinx.serialization JSON support to Spring WebFlux

Flow decoding is not supported yet since it depends on
kotlin/kotlinx.serialization#1073, but it will be
enabled when this issue will be fixed.

Closes gh-25771

spring-web/src/main/java/org/springframework/http/codec/CodecConfigurer.java

@@ -176,6 +176,22 @@ public interface CodecConfigurer {
 		 */
 		void jaxb2Encoder(Encoder<?> encoder);
 
+		/**
+		 * Override the default Kotlin Serialization JSON {@code Decoder}.
+		 * @param decoder the decoder instance to use
+		 * @since 5.3
+		 * @see org.springframework.http.codec.json.KotlinSerializationJsonDecoder
+		 */
+		void kotlinSerializationJsonDecoder(Decoder<?> decoder);
+
+		/**
+		 * Override the default Kotlin Serialization JSON {@code Encoder}.
+		 * @param encoder the encoder instance to use
+		 * @since 5.3
+		 * @see org.springframework.http.codec.json.KotlinSerializationJsonEncoder
+		 */
+		void kotlinSerializationJsonEncoder(Encoder<?> encoder);
+
 		/**
 		 * Configure a limit on the number of bytes that can be buffered whenever
 		 * the input stream needs to be aggregated. This can be a result of

spring-web/src/main/java/org/springframework/http/codec/json/KotlinSerializationJsonDecoder.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2002-2020 the original author or 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
+ *
+ *      https://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 org.springframework.http.codec.json;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+import kotlinx.serialization.KSerializer;
+import kotlinx.serialization.SerializersKt;
+import kotlinx.serialization.json.Json;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.AbstractDecoder;
+import org.springframework.core.codec.StringDecoder;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.http.MediaType;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.springframework.util.MimeType;
+
+/**
+ * Decode a byte stream into JSON and convert to Object's with
+ * <a href="https://github.com/Kotlin/kotlinx.serialization">kotlinx.serialization</a>.
+ *
+ * <p>This decoder can be used to bind {@code @Serializable} Kotlin classes.
+ * It supports {@code application/json} and {@code application/*+json} with
+ * various character sets, {@code UTF-8} being the default.
+ *
+ * <p>Decoding streams is not supported yet, see
+ * <a href="https://github.com/Kotlin/kotlinx.serialization/issues/1073">kotlinx.serialization/issues/1073</a>
+ * related issue.
+ *
+ * @author Sebastien Deleuze
+ * @since 5.3
+ */
+public class KotlinSerializationJsonDecoder extends AbstractDecoder<Object> {
+
+	private static final Map<Type, KSerializer<Object>> serializerCache = new ConcurrentReferenceHashMap<>();
+
+	private final Json json;
+
+	// String decoding needed for now, see https://github.com/Kotlin/kotlinx.serialization/issues/204 for more details
+	private final StringDecoder stringDecoder = StringDecoder.allMimeTypes(StringDecoder.DEFAULT_DELIMITERS, false);
+
+	public KotlinSerializationJsonDecoder() {
+		this(Json.Default);
+	}
+
+	public KotlinSerializationJsonDecoder(Json json) {
+		super(MediaType.APPLICATION_JSON, new MediaType("application", "*+json"));
+		this.json = json;
+	}
+
+	@Override
+	public boolean canDecode(ResolvableType elementType, MimeType mimeType) {
+		return super.canDecode(elementType, mimeType) && (!CharSequence.class.isAssignableFrom(elementType.toClass()));
+	}
+
+	@Override
+	public Flux<Object> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) {
+		return Flux.error(new UnsupportedOperationException());
+	}
+
+	@Override
+	public Mono<Object> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) {
+		return stringDecoder
+				.decodeToMono(inputStream, elementType, mimeType, hints)
+				.map(jsonText -> this.json.decodeFromString(serializer(elementType.getType()), jsonText));
+	}
+
+	/**
+	 * Tries to find a serializer that can marshall or unmarshall instances of the given type
+	 * using kotlinx.serialization. If no serializer can be found, an exception is thrown.
+	 * <p>Resolved serializers are cached and cached results are returned on successive calls.
+	 * @param type the type to find a serializer for
+	 * @return a resolved serializer for the given type
+	 * @throws RuntimeException if no serializer supporting the given type can be found
+	 */
+	private KSerializer<Object> serializer(Type type) {
+		KSerializer<Object> serializer = serializerCache.get(type);
+		if (serializer == null) {
+			serializer = SerializersKt.serializer(type);
+			serializerCache.put(type, serializer);
+		}
+		return serializer;
+	}
+}

spring-web/src/main/java/org/springframework/http/codec/json/KotlinSerializationJsonEncoder.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright 2002-2020 the original author or 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
+ *
+ *      https://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 org.springframework.http.codec.json;
+
+import java.lang.reflect.Type;
+import java.util.List;
+import java.util.Map;
+
+import kotlinx.serialization.KSerializer;
+import kotlinx.serialization.SerializersKt;
+import kotlinx.serialization.json.Json;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.AbstractEncoder;
+import org.springframework.core.codec.CharSequenceEncoder;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.ServerSentEvent;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.springframework.util.MimeType;
+
+/**
+ * Encode from an {@code Object} stream to a byte stream of JSON objects using
+ * <a href="https://github.com/Kotlin/kotlinx.serialization">kotlinx.serialization</a>.
+ *
+ * <p>This encoder can be used to bind {@code @Serializable} Kotlin classes.
+ * It supports {@code application/json} and {@code application/*+json} with
+ * various character sets, {@code UTF-8} being the default.
+ *
+ * @author Sebastien Deleuze
+ * @since 5.3
+ */
+public class KotlinSerializationJsonEncoder extends AbstractEncoder<Object> {
+
+	private static final Map<Type, KSerializer<Object>> serializerCache = new ConcurrentReferenceHashMap<>();
+
+	private final Json json;
+
+	// CharSequence encoding needed for now, see https://github.com/Kotlin/kotlinx.serialization/issues/204 for more details
+	private final CharSequenceEncoder charSequenceEncoder = CharSequenceEncoder.allMimeTypes();
+
+	public KotlinSerializationJsonEncoder() {
+		this(Json.Default);
+	}
+
+	public KotlinSerializationJsonEncoder(Json json) {
+		super(MediaType.APPLICATION_JSON, new MediaType("application", "*+json"));
+		this.json = json;
+	}
+
+	@Override
+	public boolean canEncode(ResolvableType elementType, MimeType mimeType) {
+		return super.canEncode(elementType, mimeType)
+				&& (!String.class.isAssignableFrom(elementType.toClass()))
+				&& (!ServerSentEvent.class.isAssignableFrom(elementType.toClass()));
+	}
+
+	@Override
+	public Flux<DataBuffer> encode(Publisher<?> inputStream, DataBufferFactory bufferFactory, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) {
+		if (inputStream instanceof Mono) {
+			return Mono.from(inputStream)
+					.map(value -> encodeValue(value, bufferFactory, elementType, mimeType, hints))
+					.flux();
+		}
+		else {
+			ResolvableType listType = ResolvableType.forClassWithGenerics(List.class, elementType);
+			return Flux.from(inputStream)
+					.collectList()
+					.map(list -> encodeValue(list, bufferFactory, listType, mimeType, hints))
+					.flux();
+		}
+	}
+
+	@Override
+	public DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory, ResolvableType valueType, MimeType mimeType, Map<String, Object> hints) {
+		String json = this.json.encodeToString(serializer(valueType.getType()), value);
+		return charSequenceEncoder.encodeValue(json, bufferFactory, valueType, mimeType, null);
+	}
+
+	/**
+	 * Tries to find a serializer that can marshall or unmarshall instances of the given type
+	 * using kotlinx.serialization. If no serializer can be found, an exception is thrown.
+	 * <p>Resolved serializers are cached and cached results are returned on successive calls.
+	 * @param type the type to find a serializer for
+	 * @return a resolved serializer for the given type
+	 * @throws RuntimeException if no serializer supporting the given type can be found
+	 */
+	private KSerializer<Object> serializer(Type type) {
+		KSerializer<Object> serializer = serializerCache.get(type);
+		if (serializer == null) {
+			serializer = SerializersKt.serializer(type);
+			serializerCache.put(type, serializer);
+		}
+		return serializer;
+	}
+}

spring-web/src/main/java/org/springframework/http/codec/support/BaseDefaultCodecs.java

@@ -51,6 +51,8 @@ import org.springframework.http.codec.json.Jackson2JsonDecoder;
 import org.springframework.http.codec.json.Jackson2JsonEncoder;
 import org.springframework.http.codec.json.Jackson2SmileDecoder;
 import org.springframework.http.codec.json.Jackson2SmileEncoder;
+import org.springframework.http.codec.json.KotlinSerializationJsonDecoder;
+import org.springframework.http.codec.json.KotlinSerializationJsonEncoder;
 import org.springframework.http.codec.multipart.DefaultPartHttpMessageReader;
 import org.springframework.http.codec.multipart.MultipartHttpMessageReader;
 import org.springframework.http.codec.multipart.MultipartHttpMessageWriter;
@@ -91,6 +93,8 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs, CodecConfigure
 
 	static final boolean nettyByteBufPresent;
 
+	static final boolean kotlinSerializationJsonPresent;
+
 	static {
 		ClassLoader classLoader = BaseCodecConfigurer.class.getClassLoader();
 		jackson2Present = ClassUtils.isPresent("com.fasterxml.jackson.databind.ObjectMapper", classLoader) &&
@@ -100,6 +104,7 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs, CodecConfigure
 		protobufPresent = ClassUtils.isPresent("com.google.protobuf.Message", classLoader);
 		synchronossMultipartPresent = ClassUtils.isPresent("org.synchronoss.cloud.nio.multipart.NioMultipartParser", classLoader);
 		nettyByteBufPresent = ClassUtils.isPresent("io.netty.buffer.ByteBuf", classLoader);
+		kotlinSerializationJsonPresent = ClassUtils.isPresent("kotlinx.serialization.json.Json", classLoader);
 	}
 
 
@@ -127,6 +132,12 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs, CodecConfigure
 	@Nullable
 	private Encoder<?> jaxb2Encoder;
 
+	@Nullable
+	private Decoder<?> kotlinSerializationJsonDecoder;
+
+	@Nullable
+	private Encoder<?> kotlinSerializationJsonEncoder;
+
 	@Nullable
 	private Integer maxInMemorySize;
 
@@ -151,6 +162,8 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs, CodecConfigure
 		this.protobufEncoder = other.protobufEncoder;
 		this.jaxb2Decoder = other.jaxb2Decoder;
 		this.jaxb2Encoder = other.jaxb2Encoder;
+		this.kotlinSerializationJsonDecoder = other.kotlinSerializationJsonDecoder;
+		this.kotlinSerializationJsonEncoder = other.kotlinSerializationJsonEncoder;
 		this.maxInMemorySize = other.maxInMemorySize;
 		this.enableLoggingRequestDetails = other.enableLoggingRequestDetails;
 		this.registerDefaults = other.registerDefaults;
@@ -196,6 +209,16 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs, CodecConfigure
 		this.jaxb2Encoder = encoder;
 	}
 
+	@Override
+	public void kotlinSerializationJsonDecoder(Decoder<?> decoder) {
+		this.kotlinSerializationJsonDecoder = decoder;
+	}
+
+	@Override
+	public void kotlinSerializationJsonEncoder(Encoder<?> encoder) {
+		this.kotlinSerializationJsonEncoder = encoder;
+	}
+
 	@Override
 	public void maxInMemorySize(int byteCount) {
 		this.maxInMemorySize = byteCount;
@@ -365,6 +388,9 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs, CodecConfigure
 		if (jackson2Present) {
 			addCodec(readers, new DecoderHttpMessageReader<>(getJackson2JsonDecoder()));
 		}
+		else if (kotlinSerializationJsonPresent) {
+			addCodec(readers, new DecoderHttpMessageReader<>(getKotlinSerializationJsonDecoder()));
+		}
 		if (jackson2SmilePresent) {
 			addCodec(readers, new DecoderHttpMessageReader<>(this.jackson2SmileDecoder != null ?
 					(Jackson2SmileDecoder) this.jackson2SmileDecoder : new Jackson2SmileDecoder()));
@@ -461,6 +487,9 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs, CodecConfigure
 		if (jackson2Present) {
 			writers.add(new EncoderHttpMessageWriter<>(getJackson2JsonEncoder()));
 		}
+		else if (kotlinSerializationJsonPresent) {
+			writers.add(new EncoderHttpMessageWriter<>(getKotlinSerializationJsonEncoder()));
+		}
 		if (jackson2SmilePresent) {
 			writers.add(new EncoderHttpMessageWriter<>(this.jackson2SmileEncoder != null ?
 					(Jackson2SmileEncoder) this.jackson2SmileEncoder : new Jackson2SmileEncoder()));
@@ -522,4 +551,18 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs, CodecConfigure
 		return this.jackson2JsonEncoder;
 	}
 
+	protected Decoder<?> getKotlinSerializationJsonDecoder() {
+		if (this.kotlinSerializationJsonDecoder == null) {
+			this.kotlinSerializationJsonDecoder = new KotlinSerializationJsonDecoder();
+		}
+		return this.kotlinSerializationJsonDecoder;
+	}
+
+	protected Encoder<?> getKotlinSerializationJsonEncoder() {
+		if (this.kotlinSerializationJsonEncoder == null) {
+			this.kotlinSerializationJsonEncoder = new KotlinSerializationJsonEncoder();
+		}
+		return this.kotlinSerializationJsonEncoder;
+	}
+
 }

spring-web/src/main/java/org/springframework/http/codec/support/ClientDefaultCodecsImpl.java

@@ -96,9 +96,10 @@ class ClientDefaultCodecsImpl extends BaseDefaultCodecs implements ClientCodecCo
 	@Override
 	protected void extendObjectReaders(List<HttpMessageReader<?>> objectReaders) {
 
-		Decoder<?> decoder = (this.sseDecoder != null ?
-				this.sseDecoder :
-				jackson2Present ? getJackson2JsonDecoder() : null);
+		Decoder<?> decoder = (this.sseDecoder != null ? this.sseDecoder :
+				jackson2Present ? getJackson2JsonDecoder() :
+				kotlinSerializationJsonPresent ? getKotlinSerializationJsonDecoder() :
+				null);
 
 		addCodec(objectReaders, new ServerSentEventHttpMessageReader(decoder));
 	}

spring-web/src/main/java/org/springframework/http/codec/support/ServerDefaultCodecsImpl.java

@@ -86,7 +86,10 @@ class ServerDefaultCodecsImpl extends BaseDefaultCodecs implements ServerCodecCo
 
 	@Nullable
 	private Encoder<?> getSseEncoder() {
-		return this.sseEncoder != null ? this.sseEncoder : jackson2Present ? getJackson2JsonEncoder() : null;
+		return this.sseEncoder != null ? this.sseEncoder :
+				jackson2Present ? getJackson2JsonEncoder() :
+				kotlinSerializationJsonPresent ? getKotlinSerializationJsonEncoder() :
+				null;
 	}
 
 }