* Fix KafkaDslKotlinTests according new Kotlin expectations

* Document `suspend fun` for `@MessagingGateway`

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -38,6 +38,7 @@
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.mapping.InboundMessageMapper;
 import org.springframework.integration.mapping.MessageMappingException;
@@ -216,6 +217,14 @@ private Map<String, Object> evaluateHeaders(EvaluationContext methodInvocationEv
 		return evaluatedHeaders;
 	}
 
+	// TODO Remove in the future release. The MethodArgsHolder as a root object covers this use-case.
+	private StandardEvaluationContext createMethodInvocationEvaluationContext(Object[] arguments) {
+		StandardEvaluationContext context = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
+		context.setVariable("args", arguments);
+		context.setVariable("gatewayMethod", this.method);
+		return context;
+	}
+
 	@Nullable
 	private Object evaluatePayloadExpression(String expressionString, Object argumentValue) {
 		Expression expression =
@@ -286,24 +295,23 @@ public class DefaultMethodArgsMessageMapper implements MethodArgsMessageMapper {
 		public Message<?> toMessage(MethodArgsHolder holder, @Nullable Map<String, Object> headersToMap) {
 			Object messageOrPayload = null;
 			boolean foundPayloadAnnotation = false;
-			EvaluationContext methodInvocationEvaluationContext =
-					ExpressionUtils.createStandardEvaluationContext(GatewayMethodInboundMessageMapper.this.beanFactory);
+			Object[] arguments = holder.getArgs();
+			EvaluationContext methodInvocationEvaluationContext = createMethodInvocationEvaluationContext(arguments);
+			Map<String, Object> headersToPopulate =
+					headersToMap != null
+							? new HashMap<>(headersToMap)
+							: new HashMap<>();
 			if (GatewayMethodInboundMessageMapper.this.payloadExpression != null) {
 				messageOrPayload =
 						GatewayMethodInboundMessageMapper.this.payloadExpression.getValue(
 								methodInvocationEvaluationContext, holder);
 			}
-			Map<String, Object> headersToPopulate =
-					headersToMap != null
-							? new HashMap<>(headersToMap)
-							: new HashMap<>();
-			Object[] arguments = holder.getArgs();
 			for (int i = 0; i < GatewayMethodInboundMessageMapper.this.parameterList.size(); i++) {
 				Object argumentValue = arguments[i];
-				if (CoroutinesUtils.isContinuation(argumentValue)) {
+				MethodParameter methodParameter = GatewayMethodInboundMessageMapper.this.parameterList.get(i);
+				if (CoroutinesUtils.isContinuationType(methodParameter.getParameterType())) {
 					continue;
 				}
-				MethodParameter methodParameter = GatewayMethodInboundMessageMapper.this.parameterList.get(i);
 				Annotation annotation =
 						MessagingAnnotationUtils.findMessagePartAnnotation(methodParameter.getParameterAnnotations(),
 								false);

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.util;
 
 import org.springframework.core.KotlinDetector;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
@@ -36,6 +37,7 @@ public final class CoroutinesUtils {
 	/**
 	 * The {@link kotlin.coroutines.Continuation} class object.
 	 */
+	@Nullable
 	public static final Class<?> KOTLIN_CONTINUATION_CLASS;
 
 	static {
@@ -56,16 +58,17 @@ public final class CoroutinesUtils {
 		}
 	}
 
-	public static boolean isContinuationType(Class<?> candidate) {
-		return KOTLIN_CONTINUATION_CLASS != null && KOTLIN_CONTINUATION_CLASS.isAssignableFrom(candidate);
+	public static boolean isContinuation(Object candidate) {
+		return isContinuationType(candidate.getClass());
 	}
 
-	public static boolean isContinuation(Object candidate) {
-		return KOTLIN_CONTINUATION_CLASS != null && KOTLIN_CONTINUATION_CLASS.isAssignableFrom(candidate.getClass());
+	public static boolean isContinuationType(Class<?> candidate) {
+		return KOTLIN_CONTINUATION_CLASS != null && KOTLIN_CONTINUATION_CLASS.isAssignableFrom(candidate);
 	}
 
 	@SuppressWarnings("unchecked")
-	public static <T> T monoAwaitSingleOrNull(Mono<T> source, Object continuation) {
+	public static <T> T monoAwaitSingleOrNull(Mono<? extends T> source, Object continuation) {
+		Assert.notNull(KOTLIN_CONTINUATION_CLASS, "Kotlin Coroutines library is not present in classpath");
 		Assert.isAssignable(KOTLIN_CONTINUATION_CLASS, continuation.getClass());
 		return (T) kotlinx.coroutines.reactor.MonoKt.awaitSingleOrNull(
 				source, (kotlin.coroutines.Continuation<T>) continuation);

spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt

@@ -292,7 +292,7 @@ class KafkaDslKotlinTests {
 
 		private fun kafkaMessageHandler(producerFactory: ProducerFactory<Int, String>, topic: String) =
 			Kafka.outboundChannelAdapter(producerFactory)
-				.messageKey { it.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] }
+				.messageKey<Any> { it.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] }
 				.headerMapper(mapper())
 				.sync(true)
 				.partitionId<Any> { 0 }

src/reference/asciidoc/gateway.adoc

@@ -777,6 +777,8 @@ mono.subscribe(invoice -> handleInvoice(invoice));
 
 The calling thread continues, with `handleInvoice()` being called when the flow completes.
 
+Also see <<./kotlin-functions.adoc#kotlin-coroutines,Kotlin Coroutines>> for more information.
+
 ===== Downstream Flows Returning an Asynchronous Type
 
 As mentioned in the <<gateway-asynctaskexecutor>> section above, if you wish some downstream component to return a message with an async payload (`Future`, `Mono`, and others), you must explicitly set the async executor to `null` (or `""` when using XML configuration).

src/reference/asciidoc/kotlin-functions.adoc

@@ -51,3 +51,35 @@ 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.
+
+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:
+
+====
+[source, kotlin]
+----
+@MessagingGateway(defaultRequestChannel = "suspendRequestChannel")
+interface SuspendFunGateway {
+
+    suspend fun suspendGateway(payload: String): String
+
+}
+----
+====
+
+This method has to be called as a coroutine according to Kotlin language requirements:
+
+====
+[source, kotlin]
+----
+@Autowired
+private lateinit var suspendFunGateway: SuspendFunGateway
+
+fun someServiceMethod() {
+    runBlocking {
+        val reply = suspendFunGateway.suspendGateway("test suspend gateway")
+    }
+}
+----
+====