GH-939 & GH-956 Fix Kotlin function registration regression

Resolves #939
Resolves #956

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionRegistration.java

@@ -31,6 +31,8 @@ import reactor.core.publisher.Flux;
 
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
+import org.springframework.cloud.function.context.config.KotlinLambdaToFunctionAutoConfiguration;
+import org.springframework.core.KotlinDetector;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 
@@ -105,11 +107,14 @@ public class FunctionRegistration<T> implements BeanNameAware {
 	}
 
 	public FunctionRegistration<T> type(Type type) {
+		this.type = type;
+		if (KotlinDetector.isKotlinPresent() && this.target instanceof KotlinLambdaToFunctionAutoConfiguration.KotlinFunctionWrapper) {
+			return this;
+		}
 		Type discoveredFunctionType = FunctionTypeUtils.discoverFunctionTypeFromClass(this.target.getClass());
 		if (discoveredFunctionType == null) { // only valid for Kafka Stream KStream[] return type.
 			return null;
 		}
-		this.type = type;
 
 		Class<?> inputType = FunctionTypeUtils.getRawType(FunctionTypeUtils.getInputType(discoveredFunctionType));
 		Class<?> outputType = FunctionTypeUtils.getRawType(FunctionTypeUtils.getOutputType(discoveredFunctionType));
@@ -169,44 +174,10 @@ public class FunctionRegistration<T> implements BeanNameAware {
 	 *
 	 */
 
-//	@SuppressWarnings({ "unchecked", "rawtypes" })
-//	public <S> FunctionRegistration<S> wrap() {
-//		this.isFunctionSignatureSupported();
-//		FunctionRegistration<S> result;
-//		if (this.type == null) {
-//			result = (FunctionRegistration<S>) this;
-//		}
-//		else if (this.target instanceof RoutingFunction) {
-//			S target = (S) this.target;
-//			result = new FunctionRegistration<S>(target);
-//			result.type(this.type.getType());
-//			result = result.target(target).names(this.names)
-//					.type(result.type.wrap(Flux.class)).properties(this.properties);
-//		}
-//		else {
-//			S target = (S) this.target;
-//			result = new FunctionRegistration<S>(target);
-//			result.type(this.type.getType());
-//			result = result.target(target).names(this.names)
-//					.type(result.type.wrap(Flux.class)).properties(this.properties);
-//		}
-//
-//		return result;
-//	}
-
 	@Override
 	public void setBeanName(String name) {
 		if (CollectionUtils.isEmpty(this.names)) {
 			this.name(name);
 		}
 	}
-
-//	private void isFunctionSignatureSupported() {
-//		if (type != null) {
-//			Assert.isTrue(!(Mono.class.isAssignableFrom(this.type.getOutputWrapper())
-//					&& Mono.class.isAssignableFrom(this.type.getInputWrapper())),
-//					"Function<Mono, Mono> is not supported.");
-//		}
-//	}
-
 }

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java

@@ -32,6 +32,8 @@ import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import kotlin.jvm.functions.Function0;
+import kotlin.jvm.functions.Function1;
 import net.jodah.typetools.TypeResolver;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -46,6 +48,7 @@ import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry
 import org.springframework.cloud.function.context.config.FunctionContextUtils;
 import org.springframework.cloud.function.context.config.RoutingFunction;
 import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.core.KotlinDetector;
 import org.springframework.core.ResolvableType;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
@@ -187,6 +190,14 @@ public final class FunctionTypeUtils {
 
 	@SuppressWarnings("unchecked")
 	public static Type discoverFunctionTypeFromClass(Class<?> functionalClass) {
+		if (KotlinDetector.isKotlinPresent()) {
+			if (Function1.class.isAssignableFrom(functionalClass)) {
+				return TypeResolver.reify(Function1.class, (Class<Function1<?, ?>>) functionalClass);
+			}
+			else if (Function0.class.isAssignableFrom(functionalClass)) {
+				return TypeResolver.reify(Function0.class, (Class<Function0<?>>) functionalClass);
+			}
+		}
 		if (Function.class.isAssignableFrom(functionalClass)) {
 			for (Type superInterface : functionalClass.getGenericInterfaces()) {
 				if (superInterface != null && !superInterface.equals(Object.class)) {
@@ -203,7 +214,7 @@ public final class FunctionTypeUtils {
 		else if (Supplier.class.isAssignableFrom(functionalClass)) {
 			return TypeResolver.reify(Supplier.class, (Class<Supplier<?>>) functionalClass);
 		}
-		return null;
+		return TypeResolver.reify(functionalClass);
 	}
 
 	/**