This commit introduces a comprehensive suite of tests to cover all possible declarations of Kotlin functions, consumers, and suppliers within the spring-cloud-function-kotlin module.

Test: Add arity tests for Kotlin functions, consumers, and suppliers

The primary goal of these tests is to explore the various declaration combinations and identify potential areas for improvement and enhanced support in the framework.

Signed-off-by: Adrien Poupard <[email protected]>

Feat: Enhance Kotlin arity handling in function context

This commit refactors the `spring-cloud-function-context` to comprehensively handle all combinations of possible declarations for Kotlin functions, consumers, and suppliers. This improves the framework's ability to correctly interpret and manage different Kotlin functional styles.

Signed-off-by: Adrien Poupard <[email protected]>

Test: Add unit tests for Kotlin arity and wrapper functionalities

This commit adds new unit tests to verify the core logic for Kotlin arity handling and wrapper mechanisms. These tests cover various scenarios including catalogue registration and native type casting for consumers, functions, and suppliers.

Signed-off-by: Adrien Poupard <[email protected]>

Docs: Update programming model with enhanced Kotlin support details

Signed-off-by: Adrien Poupard <[email protected]>

Migrate kotlin `TypeUtils` methods to java `KotlinUtils` for improved Kotlin type handling and consistency.

Signed-off-by: Adrien Poupard <[email protected]>

Refactor: Replace FunctionUtils with KotlinUtils for consumer and function validation

Signed-off-by: Adrien Poupard <[email protected]>

Resolves #1278

Author: olegz

docs/modules/ROOT/pages/spring-cloud-function/programming-model.adoc

@@ -705,31 +705,177 @@ However, given that `org.springframework.cloud.function.json.JsonMapper` is also
 [[kotlin-lambda-support]]
 == Kotlin Lambda support
 
-We also provide support for Kotlin lambdas (since v2.0).
-Consider the following:
+Spring Cloud Function provides first-class support for Kotlin, allowing developers to leverage idiomatic Kotlin features, including coroutines and Flow, alongside imperative and Reactor-based programming models.
 
-[source, java]
+=== Defining Functions in Kotlin
+
+You can define Suppliers, Functions, and Consumers in Kotlin and register them as Spring beans using several approaches:
+
+* **Kotlin Lambdas:** Define functions directly as lambda expressions within `@Bean` definitions. This is concise for simple functions.
+[source, kotlin]
+----
+@Configuration
+class MyKotlinConfiguration {
+
+        @Bean
+        fun kotlinSupplier(): () -> String = { "Hello from Kotlin Lambda" }
+
+        @Bean
+        fun kotlinFunction(): (String) -> String = { it.uppercase() }
+
+        @Bean
+        fun kotlinConsumer(): (String) -> Unit = { println("Consumed by Kotlin Lambda: $it") }
+    }
+----
+
+* **Kotlin Classes implementing Kotlin Functional Types:** Define a class that directly implements the desired Kotlin functional type (e.g., `(String) -> String`, `suspend () -> Flow<Int>`).
+[source, kotlin]
+----
+@Component
+class UppercaseFunction : (String) -> String {
+override fun invoke(p1: String): String = p1.uppercase()
+}
+
+    // Can also be registered via @Bean
+----
+
+* **Kotlin Classes implementing `java.util.function` Interfaces:** Define a Kotlin class that implements the standard Java `Supplier`, `Function`, or `Consumer` interfaces.
+[source, kotlin]
+----
+@Component
+class ReverseFunction : Function<String, String> {
+override fun apply(t: String): String = t.reversed()
+}
+----
+
+Regardless of the definition style, beans of these types are registered with the `FunctionCatalog`, benefiting from features like type conversion and composition.
+
+=== Coroutine Support (`suspend` and `Flow`)
+
+A key feature is the seamless integration with Kotlin Coroutines. You can use `suspend` functions and `kotlinx.coroutines.flow.Flow` directly in your function signatures. The framework automatically handles the coroutine context and reactive stream conversions.
+
+* **`suspend` Functions:** Functions marked with `suspend` can perform non-blocking operations using coroutine delays or other suspending calls.
+[source, kotlin]
 ----
 @Bean
-open fun kotlinSupplier(): () -> String {
-    return  { "Hello from Kotlin" }
+fun suspendingFunction(): suspend (String) -> Int = {
+delay(100) // Non-blocking delay
+it.length
 }
 
 @Bean
-open fun kotlinFunction(): (String) -> String {
-    return  { it.toUpperCase() }
+fun suspendingSupplier(): suspend () -> String = {
+	delay(50)
+	"Data from suspend"
 }
 
 @Bean
-open fun kotlinConsumer(): (String) -> Unit {
-    return  { println(it) }
+fun suspendingConsumer(): suspend (String) -> Unit = {
+	delay(20)
+	println("Suspend consumed: $it")
 }
+----
 
+* **`Flow` Integration:** Kotlin `Flow` can be used for reactive stream processing, similar to Reactor's `Flux`.
+[source, kotlin]
 ----
-The above represents Kotlin lambdas configured as Spring beans. The signature of each maps to a Java equivalent of `Supplier`, `Function` and `Consumer`, and thus supported/recognized signatures by the framework.
-While mechanics of Kotlin-to-Java mapping are outside of the scope of this documentation, it is important to understand that the same rules for signature transformation outlined in "Java 8 function support" section are applied here as well.
+@Bean
+fun flowFunction(): (Flow<String>) -> Flow<Int> = { flow ->
+flow.map { it.length } // Process the stream reactively
+}
+
+@Bean
+fun flowSupplier(): () -> Flow<String> = {
+	flow { // kotlinx.coroutines.flow.flow builder
+		emit("a")
+		delay(10)
+		emit("b")
+	}
+}
+
+// Consumer example taking a Flow
+@Bean
+fun flowConsumer(): suspend (Flow<String>) -> Unit = { flow ->
+	 flow.collect { item -> // Collect must happen within a coroutine scope
+		 println("Flow consumed: $item")
+	 }
+}
+----
+
+* **Combining `suspend` and `Flow`:** You can combine `suspend` and `Flow` for complex asynchronous and streaming logic.
+[source, kotlin]
+----
+ @Bean
+ fun suspendingFlowFunction(): suspend (Flow<String>) -> Flow<String> = { incoming ->
+	 flow {
+		 delay(50) // Initial suspend
+		 incoming.collect {
+			 emit(it.uppercase()) // Process and emit
+		 }
+	 }
+ }
+
+ @Bean
+ fun suspendingFlowSupplier(): suspend () -> Flow<Int> = {
+	 flow {
+		 repeat(3) {
+			 delay(100)
+			 emit(it)
+		 }
+	 }
+ }
+----
+
+=== Reactive Types (`Mono`/`Flux`)
+
+Kotlin functions can seamlessly use Reactor's `Mono` and `Flux` types, just like Java functions.
+[source, kotlin]
+----
+@Bean
+fun reactorFunction(): (Flux<String>) -> Mono<Int> = { flux ->
+    flux.map { it.length }.reduce(0) { acc, i -> acc + i }
+}
+
+@Bean
+fun monoSupplier(): () -> Mono<String> = {
+	Mono.just("Reactive Hello")
+}
+----
+
+=== `Message<T>` Support
+
+Kotlin functions can also operate directly on `org.springframework.messaging.Message<T>` to access headers, including combinations with `suspend` and `Flow`.
+[source, kotlin]
+----
+@Bean
+fun messageFunction(): (Message<String>) -> Message<Int> = { msg ->
+    MessageBuilder.withPayload(msg.payload.length)
+        .copyHeaders(msg.headers)
+        .setHeader("processed", true)
+        .build()
+}
+
+@Bean
+fun suspendMessageFunction(): suspend (Message<String>) -> Message<String> = { msg ->
+    delay(100)
+    MessageBuilder.withPayload(msg.payload.reversed())
+        .copyHeaders(msg.headers)
+        .build()
+}
+
+@Bean
+fun flowMessageFunction(): (Flow<Message<String>>) -> Flow<Message<Int>> = { flow ->
+    flow.map { msg ->
+         MessageBuilder.withPayload(msg.payload.hashCode())
+            .copyHeaders(msg.headers)
+            .build()
+    }
+}
+----
+
+=== Kotlin Sample Project
 
-To enable Kotlin support all you need is to add Kotlin SDK libraries on the classpath which will trigger appropriate autoconfiguration and supporting classes.
+For a comprehensive set of runnable examples showcasing these Kotlin features, please refer to the `src/test/kotlin/org/springframework/cloud/function/kotlin/arity` directory within the Spring Cloud Function repository. These examples demonstrate a wide range of function signatures with different arities, including regular functions, suspend functions (coroutines), and various reactive types (Flow, Mono, Flux).
 
 [[function-component-scan]]
 == Function Component Scan

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

@@ -31,7 +31,7 @@
 
 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.cloud.function.context.wrapper.KotlinFunctionWrapper;
 import org.springframework.core.KotlinDetector;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -118,7 +118,7 @@ public FunctionRegistration<T> properties(Map<String, String> properties) {
 
 	public FunctionRegistration<T> type(Type type) {
 		this.type = type;
-		if (KotlinDetector.isKotlinPresent() && this.target instanceof KotlinLambdaToFunctionAutoConfiguration.KotlinFunctionWrapper) {
+		if (KotlinDetector.isKotlinPresent() && this.target instanceof KotlinFunctionWrapper) {
 			return this;
 		}
 		Type discoveredFunctionType = type; //FunctionTypeUtils.discoverFunctionTypeFromClass(this.target.getClass());
@@ -174,15 +174,6 @@ public FunctionRegistration<T> names(String... names) {
 		return this.names(Arrays.asList(names));
 	}
 
-	/**
-	 * Transforms (wraps) function identified by the 'target' to its {@code Flux}
-	 * equivalent unless it already is. For example, {@code Function<String, String>}
-	 * becomes {@code Function<Flux<String>, Flux<String>>}
-	 * @param <S> the expected target type of the function (e.g., FluxFunction)
-	 * @return {@code FunctionRegistration} with the appropriately wrapped target.
-	 *
-	 */
-
 	@Override
 	public void setBeanName(String name) {
 		if (CollectionUtils.isEmpty(this.names)) {

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

@@ -36,11 +36,12 @@
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.BeanFactoryAnnotationUtils;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.cloud.function.context.FunctionProperties;
 import org.springframework.cloud.function.context.FunctionRegistration;
 import org.springframework.cloud.function.context.FunctionRegistry;
 import org.springframework.cloud.function.context.config.FunctionContextUtils;
-import org.springframework.cloud.function.context.config.KotlinLambdaToFunctionAutoConfiguration;
+import org.springframework.cloud.function.context.config.KotlinLambdaToFunctionFactory;
 import org.springframework.cloud.function.context.config.RoutingFunction;
 import org.springframework.cloud.function.core.FunctionInvocationHelper;
 import org.springframework.cloud.function.json.JsonMapper;
@@ -120,7 +121,8 @@ public <T> T lookup(Class<?> type, String functionDefinition, String... expected
 		functionDefinition = StringUtils.hasText(functionDefinition)
 				? functionDefinition
 						: this.applicationContext.getEnvironment().getProperty(FunctionProperties.FUNCTION_DEFINITION, "");
-		if (!this.applicationContext.containsBean(functionDefinition) || !KotlinUtils.isKotlinType(this.applicationContext.getBean(functionDefinition))) {
+
+		if (!this.applicationContext.containsBean(functionDefinition) || !isKotlinType(functionDefinition)) {
 			functionDefinition = this.normalizeFunctionDefinition(functionDefinition);
 		}
 		if (!isFunctionDefinitionEligible(functionDefinition)) {
@@ -160,12 +162,9 @@ public <T> T lookup(Class<?> type, String functionDefinition, String... expected
 							else if (functionCandidate instanceof BiFunction || functionCandidate instanceof BiConsumer) {
 								functionRegistration = this.registerMessagingBiFunction(functionCandidate, functionName);
 							}
-							else if (KotlinUtils.isKotlinType(functionCandidate)) {
-								KotlinLambdaToFunctionAutoConfiguration.KotlinFunctionWrapper wrapper =
-									new KotlinLambdaToFunctionAutoConfiguration.KotlinFunctionWrapper(functionCandidate);
-								wrapper.setName(functionName);
-								wrapper.setBeanFactory(this.applicationContext.getBeanFactory());
-								functionRegistration = wrapper.getFunctionRegistration();
+							else if (isKotlinType(functionName, functionCandidate)) {
+								KotlinLambdaToFunctionFactory kotlinFactory = new KotlinLambdaToFunctionFactory(functionCandidate, this.applicationContext.getBeanFactory());
+								functionRegistration = kotlinFactory.getFunctionRegistration(functionName);
 							}
 							else if (this.isFunctionPojo(functionCandidate, functionName)) {
 								Method functionalMethod = FunctionTypeUtils.discoverFunctionalMethod(functionCandidate.getClass());
@@ -203,6 +202,17 @@ else if (this.isSpecialFunctionRegistration(functionNames, functionName)) {
 		return (T) function;
 	}
 
+	private boolean isKotlinType(String functionDefinition) {
+		Object fonctionBean = this.applicationContext.getBean(functionDefinition);
+		return isKotlinType(functionDefinition, fonctionBean);
+	}
+
+	private boolean isKotlinType(String functionDefinition, Object fonctionBean) {
+		ConfigurableListableBeanFactory beanFactory = this.applicationContext.getBeanFactory();
+		Type functionType = FunctionContextUtils.findType(functionDefinition, beanFactory);
+		return KotlinUtils.isKotlinType(fonctionBean, functionType);
+	}
+
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private FunctionRegistration registerMessagingBiFunction(Object userFunction, String functionName) {
 		Type biFunctionType = FunctionContextUtils.findType(this.applicationContext.getBeanFactory(), functionName);

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

@@ -48,6 +48,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import kotlin.jvm.functions.Function0;
 import kotlin.jvm.functions.Function1;
+import kotlin.jvm.functions.Function2;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
@@ -234,6 +235,10 @@ else if (Function0.class.isAssignableFrom(functionalClass)) {
 				ResolvableType kotlinType = ResolvableType.forClass(functionalClass).as(Function0.class);
 				return GenericTypeResolver.resolveType(kotlinType.getType(), functionalClass);
 			}
+			else if (Function2.class.isAssignableFrom(functionalClass)) {
+				ResolvableType kotlinType = ResolvableType.forClass(functionalClass).as(Function2.class);
+				return GenericTypeResolver.resolveType(kotlinType.getType(), functionalClass);
+			}
 		}
 		Type typeToReturn = null;
 		if (Function.class.isAssignableFrom(functionalClass)) {