* 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

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.
-	 * @since 4.3
 	 * @see #setAsync(boolean)
+	 * @since 4.3
 	 */
 	protected boolean isAsync() {
 		return this.async;
@@ -133,8 +133,8 @@ protected boolean isAsync() {
 	 * {@link #shouldCopyRequestHeaders() shouldCopyRequestHeaaders} is true.
 	 * At least one pattern as "*" means do not copy headers at all.
 	 * @param headers the headers to not propagate from the inbound message.
-	 * @since 4.3.10
 	 * @see org.springframework.util.PatternMatchUtils
+	 * @since 4.3.10
 	 */
 	@Override
 	public void setNotPropagatedHeaders(String... headers) {
@@ -190,8 +190,7 @@ public Collection<String> getNotPropagatedHeaders() {
 	/**
 	 * Add header patterns ("xxx*", "*xxx", "*xxx*" or "xxx*yyy")
 	 * that will NOT be copied from the inbound message if
-	 * {@link #shouldCopyRequestHeaders()} is true, instead of overwriting the existing
-	 * set.
+	 * {@link #shouldCopyRequestHeaders()} is true, instead of overwriting the existing set.
 	 * @param headers the headers to not propagate from the inbound message.
 	 * @since 4.3.10
 	 * @see #setNotPropagatedHeaders(String...)
@@ -310,28 +309,68 @@ private void doProduceOutput(Message<?> requestMessage, MessageHeaders requestHe
 			replyChannel = getOutputChannel();
 		}
 
-		if (this.async && (reply instanceof org.springframework.util.concurrent.ListenableFuture<?>
-				|| reply instanceof CompletableFuture<?>
-				|| reply instanceof Publisher<?>)) {
+		ReactiveAdapter reactiveAdapter = null;
 
-			if (reply instanceof Publisher<?> &&
-					replyChannel instanceof ReactiveStreamsSubscribableChannel) {
+		if (this.async &&
+				(reply instanceof org.springframework.util.concurrent.ListenableFuture<?>
+						|| reply instanceof CompletableFuture<?>
+						|| (reactiveAdapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(null, reply)) != null)) {
 
-				((ReactiveStreamsSubscribableChannel) replyChannel)
+			if (replyChannel instanceof ReactiveStreamsSubscribableChannel reactiveStreamsSubscribableChannel) {
+				Publisher<?> reactiveReply = toPublisherReply(reply, reactiveAdapter);
+				reactiveStreamsSubscribableChannel
 						.subscribeTo(
-								Flux.from((Publisher<?>) reply)
+								Flux.from(reactiveReply)
 										.doOnError((ex) -> sendErrorMessage(requestMessage, ex))
 										.map(result -> createOutputMessage(result, requestHeaders)));
 			}
 			else {
-				asyncNonReactiveReply(requestMessage, reply, replyChannel);
+				CompletableFuture<?> futureReply = toFutureReply(reply, reactiveAdapter);
+				futureReply.whenComplete(new ReplyFutureCallback(requestMessage, replyChannel));
 			}
 		}
 		else {
 			sendOutput(createOutputMessage(reply, requestHeaders), replyChannel, false);
 		}
 	}
 
+	private static Publisher<?> toPublisherReply(Object reply, @Nullable ReactiveAdapter reactiveAdapter) {
+		if (reactiveAdapter != null) {
+			return reactiveAdapter.toPublisher(reply);
+		}
+		else {
+			return Mono.fromFuture(toCompletableFuture(reply));
+		}
+	}
+
+	private static CompletableFuture<?> toFutureReply(Object reply, @Nullable ReactiveAdapter reactiveAdapter) {
+		if (reactiveAdapter != null) {
+			Mono<?> reactiveReply;
+			Publisher<?> publisher = reactiveAdapter.toPublisher(reply);
+			if (reactiveAdapter.isMultiValue()) {
+				reactiveReply = Mono.just(publisher);
+			}
+			else {
+				reactiveReply = Mono.from(publisher);
+			}
+
+			return reactiveReply.publishOn(Schedulers.boundedElastic()).toFuture();
+		}
+		else {
+			return toCompletableFuture(reply);
+		}
+	}
+
+	@SuppressWarnings("deprecation")
+	private static CompletableFuture<?> toCompletableFuture(Object reply) {
+		if (reply instanceof CompletableFuture<?>) {
+			return (CompletableFuture<?>) reply;
+		}
+		else {
+			return ((org.springframework.util.concurrent.ListenableFuture<?>) reply).completable();
+		}
+	}
+
 	private AbstractIntegrationMessageBuilder<?> addRoutingSlipHeader(Object reply, List<?> routingSlip,
 			AtomicInteger routingSlipIndex) {
 
@@ -354,30 +393,6 @@ else if (reply instanceof AbstractIntegrationMessageBuilder) {
 		return builder;
 	}
 
-	@SuppressWarnings("deprecation")
-	private void asyncNonReactiveReply(Message<?> requestMessage, Object reply, @Nullable Object replyChannel) {
-		CompletableFuture<?> future;
-		if (reply instanceof CompletableFuture<?>) {
-			future = (CompletableFuture<?>) reply;
-		}
-		else if (reply instanceof org.springframework.util.concurrent.ListenableFuture<?>) {
-			future = ((org.springframework.util.concurrent.ListenableFuture<?>) reply).completable();
-		}
-		else {
-			Mono<?> reactiveReply;
-			ReactiveAdapter adapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(null, reply);
-			if (adapter != null && adapter.isMultiValue()) {
-				reactiveReply = Mono.just(reply);
-			}
-			else {
-				reactiveReply = Mono.from((Publisher<?>) reply);
-			}
-
-			future = reactiveReply.publishOn(Schedulers.boundedElastic()).toFuture();
-		}
-		future.whenComplete(new ReplyFutureCallback(requestMessage, replyChannel));
-	}
-
 	private Object getOutputChannelFromRoutingSlip(Object reply, Message<?> requestMessage, List<?> routingSlip,
 			AtomicInteger routingSlipIndex) {
 
@@ -446,7 +461,7 @@ else if (output instanceof AbstractIntegrationMessageBuilder) {
 	 * <code>null</code>, and it must be an instance of either String or {@link MessageChannel}.
 	 * @param output the output object to send
 	 * @param replyChannelArg the 'replyChannel' value from the original request
-	 * @param useArgChannel - use the replyChannel argument (must not be null), not
+	 * @param useArgChannel use the replyChannel argument (must not be null), not
 	 * the configured output channel.
 	 */
 	protected void sendOutput(Object output, @Nullable Object replyChannelArg, boolean useArgChannel) {

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

@@ -55,6 +55,7 @@
 import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ParameterNameDiscoverer;
+import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.core.annotation.AnnotationAttributes;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.core.convert.ConversionFailedException;
@@ -329,7 +330,6 @@ else if (targetObject instanceof Consumer) {
 	/**
 	 * A {@code boolean} flag to use SpEL Expression evaluation or {@link InvocableHandlerMethod}
 	 * for target method invocation.
-	 *
 	 * @param useSpelInvoker to use SpEL Expression evaluation or not.
 	 * @since 5.0
 	 */
@@ -520,7 +520,6 @@ private boolean isProvidedMessageHandlerFactoryBean() {
 	 * This should not be needed in production but we have many tests
 	 * that don't run in an application context.
 	 */
-
 	private void initializeHandler(HandlerMethod candidate) {
 		ExpressionParser parser;
 		if (candidate.useSpelInvoker == null) {
@@ -847,7 +846,7 @@ private void populateHandlerMethod(Map<Class<?>, HandlerMethod> candidateMethods
 			if (handlerMethod1.isMessageMethod()) {
 				if (fallbackMessageMethods.containsKey(targetParameterType)) {
 					// we need to check for duplicate type matches,
-					// but only if we end up falling back
+					// but only if we end up falling back,
 					// and we'll only keep track of the first one
 					ambiguousFallbackMessageGenericType.compareAndSet(null, targetParameterType);
 				}
@@ -887,7 +886,6 @@ private void findSingleSpecificMethodOnInterfacesIfProxy(Map<Class<?>, HandlerMe
 			Map<Class<?>, HandlerMethod> candidateMethods) {
 		if (AopUtils.isAopProxy(this.targetObject)) {
 			final AtomicReference<Method> targetMethod = new AtomicReference<>();
-			final AtomicReference<Class<?>> targetClass = new AtomicReference<>();
 			Class<?>[] interfaces = ((Advised) this.targetObject).getProxiedInterfaces();
 			for (Class<?> clazz : interfaces) {
 				ReflectionUtils.doWithMethods(clazz, method1 -> {
@@ -897,7 +895,6 @@ private void findSingleSpecificMethodOnInterfacesIfProxy(Map<Class<?>, HandlerMe
 					}
 					else {
 						targetMethod.set(method1);
-						targetClass.set(clazz);
 					}
 				}, method12 -> method12.getName().equals(this.methodName));
 			}
@@ -1029,7 +1026,8 @@ public boolean isAsync() {
 			Method methodToCheck = this.handlerMethodsList.get(0).values().iterator().next().method;
 			return Publisher.class.isAssignableFrom(methodToCheck.getReturnType())
 					|| CompletableFuture.class.isAssignableFrom(methodToCheck.getReturnType())
-					|| KotlinDetector.isSuspendingFunction(methodToCheck);
+					|| KotlinDetector.isSuspendingFunction(methodToCheck)
+					|| ReactiveAdapterRegistry.getSharedInstance().getAdapter(methodToCheck.getReturnType()) != null;
 		}
 		return false;
 	}
@@ -1066,7 +1064,7 @@ private static class HandlerMethod {
 
 		// The number of times InvocableHandlerMethod was attempted and failed - enables us to eventually
 		// give up trying to call it when it just doesn't seem to be possible.
-		// Switching to spelOnly afterwards forever.
+		// Switching to 'spelOnly' afterwards forever.
 		private volatile int failedAttempts = 0;
 
 		HandlerMethod(Method method, boolean canProcessMessageList) {
@@ -1359,7 +1357,6 @@ public static class ParametersWrapper {
 
 		/**
 		 * SpEL Function to retrieve a required header.
-		 *
 		 * @param headers the headers.
 		 * @param header  the header name
 		 * @return the header

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

@@ -17,21 +17,14 @@
 package org.springframework.integration.function
 
 import assertk.assertThat
-import assertk.assertions.containsAll
-import assertk.assertions.isEqualTo
-import assertk.assertions.isNotNull
-import assertk.assertions.isTrue
-import assertk.assertions.size
+import assertk.assertions.*
+import kotlinx.coroutines.flow.flow
 import org.junit.jupiter.api.Test
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
-import org.springframework.integration.annotation.EndpointId
-import org.springframework.integration.annotation.InboundChannelAdapter
-import org.springframework.integration.annotation.Poller
-import org.springframework.integration.annotation.ServiceActivator
-import org.springframework.integration.annotation.Transformer
+import org.springframework.integration.annotation.*
 import org.springframework.integration.channel.DirectChannel
 import org.springframework.integration.channel.FluxMessageChannel
 import org.springframework.integration.channel.QueueChannel
@@ -91,8 +84,8 @@ class FunctionsTests {
 		val replyChannel = QueueChannel()
 
 		val message = MessageBuilder.withPayload("foo")
-				.setReplyChannel(replyChannel)
-				.build()
+			.setReplyChannel(replyChannel)
+			.build()
 
 		this.functionServiceChannel.send(message)
 
@@ -101,8 +94,8 @@ class FunctionsTests {
 		val payload = receive?.payload
 
 		assertThat(payload)
-				.isNotNull()
-				.isEqualTo("FOO")
+			.isNotNull()
+			.isEqualTo("FOO")
 	}
 
 	@Test
@@ -142,8 +135,8 @@ class FunctionsTests {
 		val mono = this.monoFunction.apply("test")
 
 		StepVerifier.create(mono.map(Message<*>::getPayload).cast(String::class.java))
-				.expectNext("TEST")
-				.verifyComplete()
+			.expectNext("TEST")
+			.verifyComplete()
 
 		val gateways = this.monoFunctionGateway.gateways
 		assertThat(gateways).size().isEqualTo(3)
@@ -167,7 +160,30 @@ class FunctionsTests {
 		suspendServiceChannel.send(
 			MessageBuilder.withPayload(testPayload)
 				.setReplyChannel(replyChannel)
-			.build())
+				.build()
+		)
+
+		stepVerifier.verify(Duration.ofSeconds(10))
+	}
+
+	@Autowired
+	private lateinit var flowServiceChannel: MessageChannel
+
+	@Test
+	fun `verify flow function`() {
+		val replyChannel = FluxMessageChannel()
+		val testPayload = "test flow"
+		val stepVerifier =
+			StepVerifier.create(Flux.from(replyChannel).map(Message<*>::getPayload).cast(String::class.java))
+				.expectNext("$testPayload #1", "$testPayload #2", "$testPayload #3")
+				.thenCancel()
+				.verifyLater()
+
+		flowServiceChannel.send(
+			MessageBuilder.withPayload(testPayload)
+				.setReplyChannel(replyChannel)
+				.build()
+		)
 
 		stepVerifier.verify(Duration.ofSeconds(10))
 	}
@@ -195,30 +211,40 @@ class FunctionsTests {
 		fun counterChannel() = DirectChannel()
 
 		@Bean
-		@InboundChannelAdapter(value = "counterChannel", autoStartup = "false",
-				poller = Poller(fixedRate = "10", maxMessagesPerPoll = "1"))
+		@InboundChannelAdapter(
+			value = "counterChannel", autoStartup = "false",
+			poller = Poller(fixedRate = "10", maxMessagesPerPoll = "1")
+		)
 		@EndpointId("kotlinSupplierChannelAdapter")
 		fun kotlinSupplier(): () -> String {
 			return { "baz" }
 		}
 
 		@Bean
 		fun flowFromSupplier() =
-				integrationFlow({ "" }, { poller { it.fixedDelay(10).maxMessagesPerPoll(1) } }) {
-					transform<String> { "blank" }
-					channel { queue("fromSupplierQueue") }
-				}
+			integrationFlow({ "" }, { poller { it.fixedDelay(10).maxMessagesPerPoll(1) } }) {
+				transform<String> { "blank" }
+				channel { queue("fromSupplierQueue") }
+			}
 
 		@Bean
 		fun monoFunctionGateway() =
-				integrationFlow<MonoFunction>({ proxyDefaultMethods(true) }) {
-					handle<String>({ p, _ -> Mono.just(p).map(String::uppercase) }) { async(true) }
-				}
+			integrationFlow<MonoFunction>({ proxyDefaultMethods(true) }) {
+				handle<String>({ p, _ -> Mono.just(p).map(String::uppercase) }) { async(true) }
+			}
 
 
 		@ServiceActivator(inputChannel = "suspendServiceChannel")
 		suspend fun suspendServiceFunction(payload: String) = payload.uppercase()
 
+		@ServiceActivator(inputChannel = "flowServiceChannel")
+		fun flowServiceFunction(payload: String) =
+			flow {
+				for (i in 1..3) {
+					emit("$payload #$i")
+				}
+			}
+
 	}
 
 	interface MonoFunction : Function<String, Mono<Message<*>>>