* Add kotlinx.coroutines.flow.Flow support

The `Flow` is essentially a multi-value reactive `Publisher`,
so use `ReactiveAdapterRegistry` to convert any custom reactive streams result to `Flux` and `Mono`
which we already support as reply types

* Add docs for `Kotlin Coroutines`
Rearrange the doc a bit extracting Kotlin support to individual `kotlin-functions.adoc` file

* Fix missed link to `reactive-streams.adoc` from the `index-single.adoc`
* Fix unintended Javadocs formatting in the `AbstractMessageProducingHandler`

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java

@@ -120,8 +120,8 @@ public final void setAsync(boolean async) {
 
 	/**
 	 * @return true if this handler supports async replies.
-	 * @see #setAsync(boolean)
 	 * @since 4.3
+	 * @see #setAsync(boolean)
 	 */
 	protected boolean isAsync() {
 		return this.async;

src/reference/asciidoc/functions-support.adoc

@@ -110,32 +110,3 @@ public IntegrationFlow supplierFlow() {
 ====
 
 This function support is useful when used together with the https://cloud.spring.io/spring-cloud-function/[Spring Cloud Function] framework, where we have a function catalog and can refer to its member functions from an integration flow definition.
-
-[[kotlin-functions-support]]
-==== Kotlin Lambdas
-
-The Framework also has been improved to support Kotlin lambdas for functions, so now you can use a combination of the Kotlin language and Spring Integration flow definitions:
-
-====
-[source, java]
-----
-@Bean
-@Transformer(inputChannel = "functionServiceChannel")
-fun kotlinFunction(): (String) -> String {
-    return { it.toUpperCase() }
-}
-
-@Bean
-@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
-fun kotlinConsumer(): (Message<Any>) -> Unit {
-    return { print(it) }
-}
-
-@Bean
-@InboundChannelAdapter(value = "counterChannel",
-        poller = [Poller(fixedRate = "10", maxMessagesPerPoll = "1")])
-fun kotlinSupplier(): () -> String {
-    return { "baz" }
-}
-----
-====

src/reference/asciidoc/index-single.adoc

@@ -37,6 +37,8 @@ include::./kotlin-dsl.adoc[]
 
 include::./system-management.adoc[]
 
+include::./reactive-streams.adoc[]
+
 include::./endpoint-summary.adoc[]
 
 include::./amqp.adoc[]

src/reference/asciidoc/kotlin-functions.adoc

@@ -0,0 +1,53 @@
+[[kotlin-functions-support]]
+=== Kotlin Support
+
+The Framework also has been improved to support Kotlin lambdas for functions, so now you can use a combination of the Kotlin language and Spring Integration flow definitions:
+
+====
+[source, kotlin]
+----
+@Bean
+@Transformer(inputChannel = "functionServiceChannel")
+fun kotlinFunction(): (String) -> String {
+    return { it.toUpperCase() }
+}
+
+@Bean
+@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
+fun kotlinConsumer(): (Message<Any>) -> Unit {
+    return { print(it) }
+}
+
+@Bean
+@InboundChannelAdapter(value = "counterChannel",
+        poller = Poller(fixedRate = "10", maxMessagesPerPoll = "1"))
+fun kotlinSupplier(): () -> String {
+    return { "baz" }
+}
+----
+====
+
+[[kotlin-coroutines]]
+==== Kotlin Coroutines
+
+Starting with version 6.0, Spring Integration provides support for https://kotlinlang.org/docs/coroutines-guide.html[Kotlin Coroutines].
+Now the `suspend` functions and `kotlinx.coroutines.Deferred` & `kotlinx.coroutines.flow.Flow` return types can be used for service methods:
+
+====
+[source, kotlin]
+----
+@ServiceActivator(inputChannel = "suspendServiceChannel", outputChannel = "resultChannel")
+suspend fun suspendServiceFunction(payload: String) = payload.uppercase()
+
+@ServiceActivator(inputChannel = "flowServiceChannel", outputChannel = "resultChannel")
+fun flowServiceFunction(payload: String) =
+    flow {
+        for (i in 1..3) {
+            emit("$payload #$i")
+        }
+    }
+----
+====
+
+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.

src/reference/asciidoc/messaging-endpoints.adoc

@@ -1,22 +1,13 @@
 [[messaging-endpoints-chapter]]
 == Messaging Endpoints
 
-// BE SURE TO PRECEDE ALL include:: with a blank line - see https://asciidoctor.org/docs/user-manual/#include-partitioning
 include::./endpoint.adoc[]
-
 include::./gateway.adoc[]
-
 include::./service-activator.adoc[]
-
 include::./delayer.adoc[]
-
 include::./scripting.adoc[]
-
 include::./groovy.adoc[]
-
 include::./handler-advice.adoc[]
-
 include::./logging-adapter.adoc[]
-
 include::./functions-support.adoc[]
-// BE SURE TO PRECEDE ALL include:: with a blank line - see https://asciidoctor.org/docs/user-manual/#include-partitioning
+include::./kotlin-functions.adoc[]

src/reference/asciidoc/reactive-streams.adoc

@@ -49,6 +49,8 @@ With a `ReactiveStreamsSubscribableChannel` for the `outputChannel`, there is no
 
 See <<./service-activator.adoc#async-service-activator,Asynchronous Service Activator>> for more information.
 
+Also see <<./kotlin-functions.adoc#kotlin-coroutines,Kotlin Coroutines>> for more information.
+
 === `FluxMessageChannel` and `ReactiveStreamsConsumer`
 
 The `FluxMessageChannel` is a combined implementation of `MessageChannel` and `Publisher<Message<?>>`.

src/reference/asciidoc/whats-new.adoc

@@ -56,6 +56,14 @@ See <<./amqp.adoc#rmq-streams,RabbitMQ Stream Queue Support>> for more informati
 The SFTP modules has been fully reworked from outdated JCraft JSch library to more robust and modern `org.apache.sshd:sshd-sftp` module of the Apache MINA project.
 
 See <<./sftp.adoc#sftp,SFTP Adapters>> for more information.
+
+[[x6.0-kotlin-coroutines]]
+==== Kotlin Coroutines
+
+Kotlin Coroutines support has been introduced to the framework.
+
+See <<./kotlin-functions.adoc#kotlin-coroutines,Kotlin Coroutines>> for more information.
+
 [[x6.0-general]]
 === General Changes