Support JSON list deserialization in KotlinSerializationStringDecoder

See gh-36597
Signed-off-by: Dmitry Sulman <dmitry.sulman@gmail.com>

spring-test/spring-test.gradle

@@ -1,6 +1,7 @@
 description = "Spring TestContext Framework"
 
 apply plugin: "kotlin"
+apply plugin: "kotlinx-serialization"
 
 dependencies {
 	api(project(":spring-core"))
@@ -81,6 +82,7 @@ dependencies {
 	testImplementation("org.hibernate.orm:hibernate-core")
 	testImplementation("org.hibernate.validator:hibernate-validator")
 	testImplementation("org.hsqldb:hsqldb")
+    testImplementation("org.jetbrains.kotlinx:kotlinx-serialization-json")
 	testImplementation("org.junit.platform:junit-platform-testkit")
 	testImplementation("tools.jackson.core:jackson-databind")
 	testRuntimeOnly("com.sun.xml.bind:jaxb-core")

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

@@ -0,0 +1,37 @@
+package org.springframework.test.web.reactive.server
+
+import kotlinx.serialization.Serializable
+import org.junit.jupiter.api.Test
+import org.springframework.http.MediaType
+import org.springframework.http.codec.json.KotlinSerializationJsonDecoder
+import org.springframework.web.bind.annotation.GetMapping
+import org.springframework.web.bind.annotation.RestController
+
+class WebTestClientKotlinTests {
+
+	@Test
+	fun expectBodyListKotlinSerialization() {
+		val client = WebTestClient.bindToController(TestController::class.java)
+			.configureClient()
+			.codecs {
+				it.registerDefaults(false)
+				it.customCodecs().register(KotlinSerializationJsonDecoder())
+			}.build()
+
+		client.get().uri("/test")
+			.accept(MediaType.APPLICATION_JSON)
+			.exchangeSuccessfully()
+			.expectBodyList<Response>()
+			.hasSize(2)
+			.contains(Response("Hello"), Response("World"))
+	}
+
+	@Serializable
+	data class Response(val message: String)
+
+	@RestController
+	class TestController {
+		@GetMapping("test")
+		fun test(): List<Response> = listOf(Response("Hello"), Response("World"))
+	}
+}

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

@@ -116,23 +116,21 @@ 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);
-			if (serializer == null) {
+			KSerializer<Object> listSerializer = serializer(ResolvableType.forClassWithGenerics(List.class, elementType));
+			if (serializer == null || listSerializer == null) {
 				return Mono.error(new DecodingException("Could not find KSerializer for " + elementType));
 			}
 			return this.stringDecoder
 					.decode(inputStream, elementType, mimeType, hints)
-					.handle((string, sink) -> {
-						try {
-							sink.next(format().decodeFromString(serializer, string));
-						}
-						catch (IllegalArgumentException ex) {
-							sink.error(processException(ex));
-						}
-					});
+					.flatMapIterable(string -> string.startsWith("[") ?
+							(List<Object>) format().decodeFromString(listSerializer, string) :
+							List.of(format().decodeFromString(serializer, string)))
+					.onErrorMap(IllegalArgumentException.class, this::processException);
 		});
 	}
 

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

@@ -197,6 +197,20 @@ class KotlinSerializationJsonDecoderTests : AbstractDecoderTests<KotlinSerializa
 		}, null, null)
 	}
 
+	@Test
+	fun decodeJsonArrayToFlux() {
+		val input = Flux.concat(
+			stringBuffer("[{\"bar\":\"b1\",\"foo\":\"f1\"},"),
+			stringBuffer("{\"bar\":\"b2\",\"foo\":\"f2\"}]"))
+
+		testDecodeAll(input, ResolvableType.forClass(Pojo::class.java), {
+			it.expectNext(Pojo("f1", "b1"))
+				.expectNext(Pojo("f2", "b2"))
+				.expectComplete()
+				.verify()
+		}, null, null)
+	}
+
 	@Test
 	override fun decodeToMono() {
 		val input = Flux.concat(