From d5ff74b272b39210af3be39baef28767c29acb59 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@vmware.com>
Date: Wed, 5 May 2021 15:55:42 -0400
Subject: [PATCH] GH-3558: Kotlin DSL: propagate generics info (#3561)

Fixes https://github.com/spring-projects/spring-integration/issues/3558

Kotlin lambdas mostly used to configure endpoints in DSL manner
are not really Java lambdas, but rather anonymous classes implementing
respective Java interfaces.

While in most cases such classes carry generic info for their method impls
properly in Java, it is somehow doesn't work well for `GenericHandler`
implemented by Kotlin lambdas

* Wrap provided `GenericHandler` in the `BaseIntegrationFlowDefinition.handle()`
into a Java lambda and call `handle()` recursively to carry an expected type to
the `LambdaMessageProcessor`
* Fix `LambdaMessageProcessor` to handle `ClassUtils.isKotlinUnit()` result of
an invocation as a `null` reply

**Cherry-pick to `5.4.x` & `5.3.x`**
---
 .../integration/dsl/BaseIntegrationFlowDefinition.java      | 3 +++
 .../integration/handler/LambdaMessageProcessor.java         | 6 +++++-
 .../org/springframework/integration/dsl/KotlinDslTests.kt   | 6 +++---
 3 files changed, 11 insertions(+), 4 deletions(-)

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java
index 8407717f8f..153f1a75b6 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java
@@ -1046,6 +1046,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		if (ClassUtils.isLambda(handler.getClass())) {
 			serviceActivatingHandler = new ServiceActivatingHandler(new LambdaMessageProcessor(handler, payloadType));
 		}
+		else if (payloadType != null) {
+			return handle(payloadType, handler::handle, endpointConfigurer);
+		}
 		else {
 			serviceActivatingHandler = new ServiceActivatingHandler(handler, ClassUtils.HANDLER_HANDLE_METHOD);
 		}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java
index e48118b434..99b2002b14 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java
@@ -94,7 +94,11 @@ public class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFac
 		Object[] args = buildArgs(message);
 
 		try {
-			return this.method.invoke(this.target, args);
+			Object result = this.method.invoke(this.target, args);
+			if (result != null && org.springframework.integration.util.ClassUtils.isKotlinUnit(result.getClass())) {
+				result = null;
+			}
+			return result;
 		}
 		catch (InvocationTargetException e) {
 			if (e.getTargetException() instanceof ClassCastException) {
diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt
index 058a37e02c..df5f8643d8 100644
--- a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt
+++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2021 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.
@@ -212,8 +212,8 @@ class KotlinDslTests {
 	fun `no reply from handle`() {
 		val payloadReference = AtomicReference<String>()
 		val integrationFlow =
-			integrationFlow("handlerInputChanenl") {
-				handle<String> { payload, _ -> payloadReference.set(payload) }
+			integrationFlow("handlerInputChannel") {
+				handle<Message<String>> { message, _ -> payloadReference.set(message.payload) }
 			}
 
 		val registration = this.integrationFlowContext.registration(integrationFlow).register()