* Verify sync and async Flow processing

* Mention default sync behavior in the docs

Author: artembilan

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

@@ -66,6 +66,7 @@ public static boolean isContinuationType(Class<?> candidate) {
 		return KOTLIN_CONTINUATION_CLASS != null && KOTLIN_CONTINUATION_CLASS.isAssignableFrom(candidate);
 	}
 
+	@Nullable
 	@SuppressWarnings("unchecked")
 	public static <T> T monoAwaitSingleOrNull(Mono<? extends T> source, Object continuation) {
 		Assert.notNull(KOTLIN_CONTINUATION_CLASS, "Kotlin Coroutines library is not present in classpath");

spring-integration-core/src/test/kotlin/org/springframework/integration/function/FunctionsTests.kt

@@ -18,7 +18,7 @@ package org.springframework.integration.function
 
 import assertk.assertThat
 import assertk.assertions.*
-import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.runBlocking
 import org.junit.jupiter.api.Test
 import org.springframework.beans.factory.annotation.Autowired
@@ -190,6 +190,34 @@ class FunctionsTests {
 		stepVerifier.verify(Duration.ofSeconds(10))
 	}
 
+	@Autowired
+	private lateinit var syncFlowServiceChannel: MessageChannel
+
+	@Test
+	fun `verify sync flow function reply`() {
+		val replyChannel = QueueChannel()
+		val testPayload = "test flow"
+
+		syncFlowServiceChannel.send(
+			MessageBuilder.withPayload(testPayload)
+				.setReplyChannel(replyChannel)
+				.build()
+		)
+
+		val receive = replyChannel.receive(10_000)
+
+		val payload = receive?.payload
+
+		assertThat(payload)
+			.isNotNull()
+			.isInstanceOf(Flow::class)
+
+		runBlocking {
+			val strings = (payload as Flow<String>).toList()
+			assertThat(strings).containsExactly("Sync $testPayload #1", "Sync $testPayload #2", "Sync $testPayload #3")
+		}
+	}
+
 	@Autowired
 	private lateinit var suspendRequestChannel: DirectChannel
 
@@ -256,14 +284,19 @@ class FunctionsTests {
 		@ServiceActivator(inputChannel = "suspendServiceChannel")
 		suspend fun suspendServiceFunction(payload: String) = payload.uppercase()
 
-		@ServiceActivator(inputChannel = "flowServiceChannel")
+		@ServiceActivator(inputChannel = "flowServiceChannel", async = "true")
 		fun flowServiceFunction(payload: String) =
 			flow {
 				for (i in 1..3) {
 					emit("$payload #$i")
 				}
 			}
 
+		@ServiceActivator(inputChannel = "syncFlowServiceChannel")
+		fun syncFlowServiceFunction(payload: String) =
+			(1..3).asFlow()
+				.map { "Sync $payload #$it" }
+
 		@Bean
 		fun suspendRequestChannel() = DirectChannel()
 

src/reference/asciidoc/kotlin-functions.adoc

@@ -39,7 +39,7 @@ Now the `suspend` functions and `kotlinx.coroutines.Deferred` & `kotlinx.corouti
 @ServiceActivator(inputChannel = "suspendServiceChannel", outputChannel = "resultChannel")
 suspend fun suspendServiceFunction(payload: String) = payload.uppercase()
 
-@ServiceActivator(inputChannel = "flowServiceChannel", outputChannel = "resultChannel")
+@ServiceActivator(inputChannel = "flowServiceChannel", outputChannel = "resultChannel", async = "true")
 fun flowServiceFunction(payload: String) =
     flow {
         for (i in 1..3) {
@@ -52,6 +52,9 @@ fun flowServiceFunction(payload: String) =
 The framework treats them as Reactive Streams interactions and uses `ReactiveAdapterRegistry` to convert to respective `Mono` and `Flux` reactor types.
 Such a function reply is processed then in the reply channel, if it is a `ReactiveStreamsSubscribableChannel`, or as a result of `CompletableFuture` in the respective callback.
 
+NOTE: The functions with `Flow` result are not `async` by default on the `@ServiceActivator`, so `Flow` instance is produced as a reply message payload.
+It is already target application to process this object as a coroutine or convert it to `Flux`, respectively.
+
 The `@MessagingGateway` interface methods also can be marked with a `suspend` modifier when declared in Kotlin.
 The framework utilizes a `Mono` logic internally to perform request-reply for downstream flow.
 Such a `Mono` result is processed by the `MonoKt.awaitSingleOrNull()` API internally to fulfil a `kotlin.coroutines.Continuation` argument fo the called `suspend` function of the gateway: