Treat Reactive Publishers the same as Kotlin Flows

This commit makes sure that CoroutinesUtils::invokeSuspendingFunction
treats Reactive Publisher instances the same as Kotlin Flows, i.e.
it flatmaps them.

Closes gh-27011

Author: poutsma

spring-core/src/main/java/org/springframework/core/CoroutinesUtils.java

@@ -22,6 +22,7 @@
 
 import kotlin.Unit;
 import kotlin.jvm.JvmClassMappingKt;
+import kotlin.reflect.KClass;
 import kotlin.reflect.KClassifier;
 import kotlin.reflect.KFunction;
 import kotlin.reflect.full.KCallables;
@@ -75,13 +76,23 @@ public static Publisher<?> invokeSuspendingFunction(Method method, Object target
 		if (method.isAccessible() && !KCallablesJvm.isAccessible(function)) {
 			KCallablesJvm.setAccessible(function, true);
 		}
-		KClassifier classifier = function.getReturnType().getClassifier();
 		Mono<Object> mono = MonoKt.mono(Dispatchers.getUnconfined(), (scope, continuation) ->
 					KCallables.callSuspend(function, getSuspendedFunctionArgs(target, args), continuation))
 				.filter(result -> !Objects.equals(result, Unit.INSTANCE))
 				.onErrorMap(InvocationTargetException.class, InvocationTargetException::getTargetException);
-		if (classifier != null && classifier.equals(JvmClassMappingKt.getKotlinClass(Flow.class))) {
-			return mono.flatMapMany(CoroutinesUtils::asFlux);
+
+		KClassifier returnType = function.getReturnType().getClassifier();
+		if (returnType != null) {
+			if (returnType.equals(JvmClassMappingKt.getKotlinClass(Flow.class))) {
+				return mono.flatMapMany(CoroutinesUtils::asFlux);
+			}
+			else if (returnType.equals(JvmClassMappingKt.getKotlinClass(Mono.class))) {
+				return mono.flatMap(o -> ((Mono<?>)o));
+			}
+			else if (returnType instanceof KClass<?> kClass &&
+					Publisher.class.isAssignableFrom(JvmClassMappingKt.getJavaClass(kClass))) {
+				return mono.flatMapMany(o -> ((Publisher<?>)o));
+			}
 		}
 		return mono;
 	}