From b6ed88eb982ea2fcbad7f5d6a922b2831f4f6f91 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 1730badb93..14fac1d5e2 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 d65c92136a..2ed45478b2 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) {
 			final Throwable cause = e.getCause();
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()