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

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.

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 <[email protected]>

Author: Adrien

spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/arity/KotlinArityApplication.kt

@@ -0,0 +1,12 @@
+package org.springframework.cloud.function.kotlin.arity
+
+import org.springframework.boot.SpringApplication
+import org.springframework.boot.autoconfigure.SpringBootApplication
+import org.springframework.boot.runApplication
+
+@SpringBootApplication
+open class KotlinArityApplication
+
+fun main(args: Array<String>) {
+	SpringApplication.run(KotlinArityApplication::class.java, *args)
+}

spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/arity/KotlinAritySupplierComponent.kt

@@ -0,0 +1,188 @@
+/*
+ * Copyright 2025-2025 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.cloud.function.kotlin.arity
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
+import org.springframework.messaging.Message
+import org.springframework.messaging.support.MessageBuilder
+import org.springframework.stereotype.Component
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import java.time.Duration
+import java.util.UUID
+
+/**
+ * Examples of implementing suppliers using Kotlin's function type.
+ * 
+ * ## List of Combinations Implemented:
+ * --- Coroutine ---
+ * 1. () -> R                      -> supplierKotlinPlain
+ * 2. () -> Flow<R>               -> supplierKotlinFlow
+ * 3. suspend () -> R             -> supplierKotlinSuspendPlain
+ * 4. suspend () -> Flow<R>       -> supplierKotlinSuspendFlow
+ * --- Reactor ---
+ * 5. () -> Mono<R>               -> supplierKotlinMono
+ * 6. () -> Flux<R>               -> supplierKotlinFlux
+ * --- Message<T> ---
+ * 7. () -> Message<R>            -> supplierKotlinMessage
+ * 8. () -> Mono<Message<R>>     -> supplierKotlinMonoMessage
+ * 9. suspend () -> Message<R>    -> supplierKotlinSuspendMessage
+ * 10. () -> Flux<Message<R>>     -> supplierKotlinFluxMessage
+ * 11. () -> Flow<Message<R>>     -> supplierKotlinFlowMessage
+ * 12. suspend () -> Flow<Message<R>> -> supplierKotlinSuspendFlowMessage
+ *
+ * @author Adrien Poupard
+ */
+class KotlinSupplierKotlinExamples
+
+/** 1) () -> R */
+@Component
+class SupplierKotlinPlain : () -> Int {
+	override fun invoke(): Int {
+		return 42
+	}
+}
+
+/** 2) () -> Flow<R> */
+@Component
+class SupplierKotlinFlow : () -> Flow<String> {
+	override fun invoke(): Flow<String> {
+		return flow {
+			emit("A")
+			emit("B")
+			emit("C")
+		}
+	}
+}
+
+/** 3) suspend () -> R */
+@Component
+class SupplierKotlinSuspendPlain : suspend () -> String {
+	override suspend fun invoke(): String {
+		return "Hello from suspend"
+	}
+}
+
+/** 4) suspend () -> Flow<R> */
+@Component
+class SupplierKotlinSuspendFlow : suspend () -> Flow<String> {
+	override suspend fun invoke(): Flow<String> {
+		return flow {
+			emit("x")
+			emit("y")
+			emit("z")
+		}
+	}
+}
+
+/** 5) () -> Mono<R> */
+@Component
+class SupplierKotlinMono : () -> Mono<String> {
+	override fun invoke(): Mono<String> {
+		return Mono.just("Hello from Mono").delayElement(Duration.ofMillis(50))
+	}
+}
+
+/** 6) () -> Flux<R> */
+@Component
+class SupplierKotlinFlux : () -> Flux<String> {
+	override fun invoke(): Flux<String> {
+		return Flux.just("Alpha", "Beta", "Gamma").delayElements(Duration.ofMillis(20))
+	}
+}
+
+/** 7) () -> Message<R> */
+@Component
+class SupplierKotlinMessage : () -> Message<String> {
+	override fun invoke(): Message<String> {
+		return MessageBuilder.withPayload("Hello from Message")
+			.setHeader("messageId", UUID.randomUUID().toString())
+			.build()
+	}
+}
+
+/** 8) () -> Mono<Message<R>> */
+@Component
+class SupplierKotlinMonoMessage : () -> Mono<Message<String>> {
+	override fun invoke(): Mono<Message<String>> {
+		return Mono.just(
+			MessageBuilder.withPayload("Hello from Mono Message")
+				.setHeader("monoMessageId", UUID.randomUUID().toString())
+				.setHeader("source", "mono")
+				.build()
+		).delayElement(Duration.ofMillis(40))
+	}
+}
+
+/** 9) suspend () -> Message<R> */
+@Component
+class SupplierKotlinSuspendMessage : suspend () -> Message<String> {
+	override suspend fun invoke(): Message<String> {
+		return MessageBuilder.withPayload("Hello from Suspend Message")
+			.setHeader("suspendMessageId", UUID.randomUUID().toString())
+			.setHeader("wasSuspended", true)
+			.build()
+	}
+}
+
+/** 10) () -> Flux<Message<R>> */
+@Component
+class SupplierKotlinFluxMessage : () -> Flux<Message<String>> {
+	override fun invoke(): Flux<Message<String>> {
+		return Flux.just("Msg1", "Msg2")
+			.delayElements(Duration.ofMillis(30))
+			.map { payload ->
+				MessageBuilder.withPayload(payload)
+					.setHeader("fluxMessageId", UUID.randomUUID().toString())
+					.build()
+			}
+	}
+}
+
+/** 11) () -> Flow<Message<R>> */
+@Component
+class SupplierKotlinFlowMessage : () -> Flow<Message<String>> {
+	override fun invoke(): Flow<Message<String>> {
+		return flow {
+			listOf("FlowMsg1", "FlowMsg2").forEach { payload ->
+				emit(
+					MessageBuilder.withPayload(payload)
+						.setHeader("flowMessageId", UUID.randomUUID().toString())
+						.build()
+				)
+			}
+		}
+	}
+}
+
+/** 12) suspend () -> Flow<Message<R>> */
+@Component
+class SupplierKotlinSuspendFlowMessage : suspend () -> Flow<Message<String>> {
+	override suspend fun invoke(): Flow<Message<String>> {
+		return flow {
+			listOf("SuspendFlowMsg1", "SuspendFlowMsg2").forEach { payload ->
+				emit(
+					MessageBuilder.withPayload(payload)
+						.setHeader("suspendFlowMessageId", UUID.randomUUID().toString())
+						.build()
+				)
+			}
+		}
+	}
+}

spring-cloud-function-kotlin/src/test/kotlin/org/springframework/cloud/function/kotlin/arity/KotlinConsumerArityBean.kt

@@ -0,0 +1,150 @@
+/*
+ * Copyright 2025-2025 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.cloud.function.kotlin.arity
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.messaging.Message
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+/**
+ * ## List of Combinations Tested (in requested order):
+ * --- Coroutine ---
+ * 1. (T) -> Unit                 -> consumerPlain
+ * 2. (Flow<T>) -> Unit           -> consumerFlow
+ * 3. suspend (T) -> Unit         -> consumerSuspendPlain
+ * 4. suspend (Flow<T>) -> Unit   -> consumerSuspendFlow
+ * --- Reactor ---
+ * 5. (T) -> Mono<Void>           -> consumerMonoInput
+ * 6. (Mono<T>) -> Mono<Void>     -> consumerMono
+ * 7. (Flux<T>) -> Mono<Void>     -> consumerFlux
+ * --- Message<T> ---
+ * 8. (Message<T>) -> Unit        -> consumerMessage
+ * 9. (Mono<Message<T>>) -> Mono<Void> -> consumerMonoMessage
+ * 10. suspend (Message<T>) -> Unit -> consumerSuspendMessage
+ * 11. (Flux<Message<T>>) -> Unit  -> consumerFluxMessage
+ * 12. (Flow<Message<T>>) -> Unit  -> consumerFlowMessage
+ * 13. suspend (Flow<Message<T>>) -> Unit -> consumerSuspendFlowMessage
+ *
+ *  @author Adrien Poupard
+ */
+@Configuration
+open class KotlinConsumerArityBean {
+
+	/** 1) (T) -> Unit */
+	@Bean
+	open fun consumerPlain(): (String) -> Unit = { input ->
+		println("Consumed: $input")
+	}
+
+	/** 2) (Flow<T>) -> Unit */
+	@Bean
+	open fun consumerFlow(): (Flow<String>) -> Unit = { flowInput ->
+		println("Received flow: $flowInput (would collect in coroutine)")
+	}
+
+	/** 3) suspend (T) -> Unit */
+	@Bean
+	open fun consumerSuspendPlain(): suspend (String) -> Unit = { input ->
+		println("Suspend consumed: $input")
+	}
+
+	/** 4) suspend (Flow<T>) -> Unit */
+	@Bean
+	open fun consumerSuspendFlow(): suspend (Flow<String>) -> Unit = { flowInput ->
+		flowInput.collect { item ->
+			println("Flow item consumed: $item")
+		}
+	}
+
+	/** 5) (T) -> Mono<Void> */
+	@Bean
+	open fun consumerMonoInput(): (String) -> Mono<Void> = { input ->
+		Mono.fromRunnable<Void> {
+			println("[Reactor] Consumed T: $input")
+		}
+	}
+
+	/** 6) (Mono<T>) -> Mono<Void> */
+	@Bean
+	open fun consumerMono(): (Mono<String>) -> Mono<Void> = { monoInput ->
+		monoInput.doOnNext { item ->
+			println("[Reactor] Consumed Mono item: $item")
+		}.then()
+	}
+
+	/** 7) (Flux<T>) -> Mono<Void> */
+	@Bean
+	open fun consumerFlux(): (Flux<String>) -> Mono<Void> = { fluxInput ->
+		fluxInput.doOnNext { item ->
+			println("[Reactor] Consumed Flux item: $item")
+		}.then()
+	}
+
+	/** 8) (Message<T>) -> Unit */
+	@Bean
+	open fun consumerMessage(): (Message<String>) -> Unit = { message ->
+		println("[Message] Consumed payload: ${message.payload}, Headers: ${message.headers}")
+	}
+
+	/** 9) (Mono<Message<T>>) -> Mono<Void> */
+	@Bean
+	open fun consumerMonoMessage(): (Mono<Message<String>>) -> Mono<Void> = { monoMsgInput ->
+		monoMsgInput
+			.doOnNext { message ->
+				println("[Message][Mono] Consumed payload: ${message.payload}, Header id: ${message.headers.id}")
+			}
+			.then()
+	}
+
+	/** 10) suspend (Message<T>) -> Unit */
+	@Bean
+	open fun consumerSuspendMessage(): suspend (Message<String>) -> Unit = { message ->
+		println("[Message][Suspend] Consumed payload: ${message.payload}, Header count: ${message.headers.size}")
+	}
+
+	/** 11) (Flux<Message<T>>) -> Unit */
+	@Bean
+	open fun consumerFluxMessage(): (Flux<Message<String>>) -> Unit = { fluxMsgInput ->
+		// Explicit subscription needed here because the lambda itself returns Unit
+		fluxMsgInput.subscribe { message ->
+			println("[Message] Consumed Flux payload: ${message.payload}, Headers: ${message.headers}")
+		}
+	}
+
+	/** 12) (Flow<Message<T>>) -> Unit */
+	@Bean
+	open fun consumerFlowMessage(): (Flow<Message<String>>) -> Unit = { flowMsgInput ->
+		// Similar to Flux consumer returning Unit, explicit collection might be needed depending on context.
+		println("[Message] Received Flow: $flowMsgInput (would need explicit collection if signature returns Unit)")
+		// Example:
+		// CoroutineScope(Dispatchers.IO).launch {
+		//     flowMsgInput.collect { message -> println(...) }
+		// }
+	}
+
+	/** 13) suspend (Flow<Message<T>>) -> Unit */
+	@Bean
+	open fun consumerSuspendFlowMessage(): suspend (Flow<Message<String>>) -> Unit = { flowMsgInput ->
+		flowMsgInput.collect { message ->
+			println("[Message] Consumed Suspend Flow payload: ${message.payload}, Headers: ${message.headers}")
+		}
+	}
+}