Refine Kotlin serialization contribution

This commit moves the JSON specific code to
KotlinSerializationJsonDecoder, uses switchOnFirst operator to keep the
existing behavior and derives the list serializer from the element one.

Closes gh-36597

spring-test/src/test/kotlin/org/springframework/test/web/reactive/server/WebTestClientKotlinTests.kt

@@ -1,3 +1,19 @@
+/*
+ * Copyright 2002-present 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.test.web.reactive.server
 
 import kotlinx.serialization.Serializable
@@ -34,4 +50,4 @@ class WebTestClientKotlinTests {
 		@GetMapping("test")
 		fun test(): List<Response> = listOf(Response("Hello"), Response("World"))
 	}
-}
+}

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

@@ -55,7 +55,7 @@ public abstract class KotlinSerializationStringDecoder<T extends StringFormat> e
 	implements Decoder<Object> {
 
 	// 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);
+	protected final StringDecoder stringDecoder = StringDecoder.allMimeTypes(StringDecoder.DEFAULT_DELIMITERS, false);
 
 
 	/**
@@ -116,21 +116,23 @@ public abstract class KotlinSerializationStringDecoder<T extends StringFormat> e
 	}
 
 	@Override
-	@SuppressWarnings("unchecked")
 	public Flux<Object> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 		return Flux.defer(() -> {
 			KSerializer<Object> serializer = serializer(elementType);
-			KSerializer<Object> listSerializer = serializer(ResolvableType.forClassWithGenerics(List.class, elementType));
-			if (serializer == null || listSerializer == null) {
+			if (serializer == null) {
 				return Mono.error(new DecodingException("Could not find KSerializer for " + elementType));
 			}
 			return this.stringDecoder
 					.decode(inputStream, elementType, mimeType, hints)
-					.flatMapIterable(string -> string.startsWith("[") ?
-							(List<Object>) format().decodeFromString(listSerializer, string) :
-							List.of(format().decodeFromString(serializer, string)))
-					.onErrorMap(IllegalArgumentException.class, this::processException);
+					.handle((string, sink) -> {
+						try {
+							sink.next(format().decodeFromString(serializer, string));
+						}
+						catch (IllegalArgumentException ex) {
+							sink.error(processException(ex));
+						}
+					});
 		});
 	}
 
@@ -156,7 +158,7 @@ public abstract class KotlinSerializationStringDecoder<T extends StringFormat> e
 		});
 	}
 
-	private CodecException processException(IllegalArgumentException ex) {
+	protected CodecException processException(IllegalArgumentException ex) {
 		return new DecodingException("Decoding error: " + ex.getMessage(), ex);
 	}
 

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

@@ -16,11 +16,22 @@
 
 package org.springframework.http.codec.json;
 
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.function.Predicate;
 
+import kotlinx.serialization.KSerializer;
+import kotlinx.serialization.builtins.BuiltinSerializersKt;
 import kotlinx.serialization.json.Json;
+import org.jspecify.annotations.Nullable;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.DecodingException;
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.KotlinSerializationStringDecoder;
 import org.springframework.util.MimeType;
@@ -95,4 +106,37 @@ public class KotlinSerializationJsonDecoder extends KotlinSerializationStringDec
 		super(json, typePredicate, DEFAULT_JSON_MIME_TYPES);
 	}
 
+	@Override
+	public Flux<Object> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+		return Flux.defer(() -> {
+			KSerializer<Object> serializer = serializer(elementType);
+			if (serializer == null) {
+				return Mono.error(new DecodingException("Could not find KSerializer for " + elementType));
+			}
+			return this.stringDecoder
+					.decode(inputStream, elementType, mimeType, hints)
+					.switchOnFirst((signal, flux) -> {
+						if (signal.hasValue()) {
+							String value = Objects.requireNonNull(signal.get());
+							if (value.stripLeading().startsWith("[") && !List.class.isAssignableFrom(elementType.toClass())) {
+								KSerializer<List<Object>> listSerializer = BuiltinSerializersKt.ListSerializer(serializer);
+								return flux
+										.flatMapIterable(string -> format().decodeFromString(listSerializer, string))
+										.onErrorMap(IllegalArgumentException.class, this::processException);
+							}
+							return flux.handle((string, sink) -> {
+								try {
+									sink.next(format().decodeFromString(serializer, string));
+								}
+								catch (IllegalArgumentException ex) {
+									sink.error(processException(ex));
+								}
+							});
+						}
+						return flux;
+					});
+		});
+	}
+
 }

spring-web/src/test/kotlin/org/springframework/http/codec/json/KotlinSerializationJsonDecoderTests.kt

@@ -211,6 +211,20 @@ class KotlinSerializationJsonDecoderTests : AbstractDecoderTests<KotlinSerializa
 		}, null, null)
 	}
 
+	@Test
+	fun decodeJsonArrayToFluxOfList() {
+		val input = Flux.concat(
+			stringBuffer("[{\"bar\":\"b1\",\"foo\":\"f1\"},{\"bar\":\"b2\",\"foo\":\"f2\"}]\n"),
+			stringBuffer("[{\"bar\":\"b3\",\"foo\":\"f3\"},{\"bar\":\"b4\",\"foo\":\"f4\"}]"))
+
+		testDecodeAll(input, ResolvableType.forClassWithGenerics(List::class.java, Pojo::class.java), {
+			it.expectNext(listOf(Pojo("f1", "b1"),Pojo("f2", "b2")))
+				.expectNext(listOf(Pojo("f3", "b3"),Pojo("f4", "b4")))
+				.expectComplete()
+				.verify()
+		}, null, null)
+	}
+
 	@Test
 	override fun decodeToMono() {
 		val input = Flux.concat(