INT-4456: Explicit methods for Kotlin lambdas

JIRA: https://jira.spring.io/browse/INT-4456

Since Kotlin compiles lambdas different way than Java, they are not
synthetic classes at runtime anymore.
Therefore we fallback to the method invocation logic, but since Java 8
interfaces has `default` methods as well the `MessagingMethodInvokerHelper`
can't select the target method for invocation

* Use explicit `Function.apply()` for non-synthetic implementations
* Add `RouterDslTests` Kotlin-based test-case

**Cherry-pick to 5.0.x**

* Add more explicit methods for invokers

Author: artembilan

build.gradle

@@ -1,12 +1,14 @@
 buildscript {
+	ext.kotlinVersion = '1.2.40'
 	repositories {
 		maven { url 'https://repo.spring.io/plugins-release' }
 	}
 	dependencies {
 		classpath 'io.spring.gradle:dependency-management-plugin:1.0.5.RELEASE'
 		classpath 'io.spring.gradle:spring-io-plugin:0.0.8.RELEASE'
 		classpath 'io.spring.gradle:docbook-reference-plugin:0.3.1'
-		classpath 'org.asciidoctor:asciidoctor-gradle-plugin:1.5.0'
+		classpath 'org.asciidoctor:asciidoctor-gradle-plugin:1.5.3'
+		classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlinVersion"
 	}
 }
 
@@ -60,6 +62,7 @@ subprojects { subproject ->
 	apply plugin: 'idea'
 	apply plugin: 'jacoco'
 	apply plugin: 'checkstyle'
+	apply plugin: 'kotlin'
 
 	sourceSets {
 		test {
@@ -86,11 +89,18 @@ subprojects { subproject ->
 		targetCompatibility = 1.8
 	}
 
+	compileTestKotlin {
+		kotlinOptions {
+			jvmTarget = '1.8'
+		}
+	}
+
 	ext {
 		activeMqVersion = '5.15.2'
 		apacheSshdVersion = '1.6.0'
 		aspectjVersion = '1.8.13'
 		assertjVersion = '3.9.1'
+		assertkVersion = '0.10'
 		boonVersion = '0.34'
 		commonsDbcp2Version = '2.1.1'
 		commonsIoVersion = '2.4'
@@ -176,6 +186,14 @@ subprojects { subproject ->
 
 		testRuntime "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion"
 		testRuntime "org.apache.logging.log4j:log4j-jcl:$log4jVersion"
+
+		testCompile("com.willowtreeapps.assertk:assertk:$assertkVersion") {
+				exclude group: 'org.jetbrains.kotlin', module: 'kotlin-reflect'
+		}
+
+		testCompile "org.jetbrains.kotlin:kotlin-stdlib:$kotlinVersion"
+		testRuntime "org.jetbrains.kotlin:kotlin-reflect:$kotlinVersion"
+
 	}
 
 	// enable all compiler warnings; individual projects may customize further

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java

@@ -88,6 +88,7 @@
 import org.springframework.integration.transformer.MessageTransformingHandler;
 import org.springframework.integration.transformer.MethodInvokingTransformer;
 import org.springframework.integration.transformer.Transformer;
+import org.springframework.integration.util.ClassUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -635,7 +636,7 @@ public <P, T> B transform(Class<P> payloadType, GenericTransformer<P, T> generic
 		Transformer transformer = genericTransformer instanceof Transformer ? (Transformer) genericTransformer :
 				(isLambda(genericTransformer)
 						? new MethodInvokingTransformer(new LambdaMessageProcessor(genericTransformer, payloadType))
-						: new MethodInvokingTransformer(genericTransformer));
+						: new MethodInvokingTransformer(genericTransformer, ClassUtils.TRANSFORMER_TRANSFORM_METHOD));
 		return addComponent(transformer)
 				.handle(new MessageTransformingHandler(transformer), endpointConfigurer);
 	}
@@ -828,7 +829,7 @@ public <P> B filter(Class<P> payloadType, GenericSelector<P> genericSelector,
 		MessageSelector selector = genericSelector instanceof MessageSelector ? (MessageSelector) genericSelector :
 				(isLambda(genericSelector)
 						? new MethodInvokingSelector(new LambdaMessageProcessor(genericSelector, payloadType))
-						: new MethodInvokingSelector(genericSelector));
+						: new MethodInvokingSelector(genericSelector, ClassUtils.SELECTOR_ACCEPT_METHOD));
 		return this.register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer);
 	}
 
@@ -1024,12 +1025,12 @@ public <P> B handle(Class<P> payloadType, GenericHandler<P> handler) {
 	 */
 	public <P> B handle(Class<P> payloadType, GenericHandler<P> handler,
 			Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
-		ServiceActivatingHandler serviceActivatingHandler = null;
+		ServiceActivatingHandler serviceActivatingHandler;
 		if (isLambda(handler)) {
 			serviceActivatingHandler = new ServiceActivatingHandler(new LambdaMessageProcessor(handler, payloadType));
 		}
 		else {
-			serviceActivatingHandler = new ServiceActivatingHandler(handler, "handle");
+			serviceActivatingHandler = new ServiceActivatingHandler(handler, ClassUtils.HANDLER_HANDLE_METHOD);
 		}
 		return this.handle(serviceActivatingHandler, endpointConfigurer);
 	}
@@ -1535,7 +1536,7 @@ public <P> B split(Class<P> payloadType, Function<P, ?> splitter,
 			Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
 		MethodInvokingSplitter split = isLambda(splitter)
 				? new MethodInvokingSplitter(new LambdaMessageProcessor(splitter, payloadType))
-				: new MethodInvokingSplitter(splitter);
+				: new MethodInvokingSplitter(splitter, ClassUtils.FUNCTION_APPLY_METHOD);
 		return this.split(split, endpointConfigurer);
 	}
 
@@ -1935,7 +1936,7 @@ public <P, T> B route(Class<P> payloadType, Function<P, T> router,
 			Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
 		MethodInvokingRouter methodInvokingRouter = isLambda(router)
 				? new MethodInvokingRouter(new LambdaMessageProcessor(router, payloadType))
-				: new MethodInvokingRouter(router);
+				: new MethodInvokingRouter(router, ClassUtils.FUNCTION_APPLY_METHOD);
 		return route(new RouterSpec<>(methodInvokingRouter), routerConfigurer);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/dsl/RecipientListRouterSpec.java

@@ -21,8 +21,8 @@
 import org.springframework.integration.core.MessageSelector;
 import org.springframework.integration.filter.ExpressionEvaluatingSelector;
 import org.springframework.integration.filter.MethodInvokingSelector;
-import org.springframework.integration.handler.LambdaMessageProcessor;
 import org.springframework.integration.router.RecipientListRouter;
+import org.springframework.integration.util.ClassUtils;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.StringUtils;
 
@@ -96,23 +96,20 @@ public RecipientListRouterSpec recipientMessageSelector(String channelName, Mess
 	 * @return the router spec.
 	 */
 	public <P> RecipientListRouterSpec recipient(String channelName, GenericSelector<P> selector) {
+		MessageSelector messageSelector = wrapToMessageSelectorIfNecessary(selector);
+		this.handler.addRecipient(channelName, messageSelector);
+		return _this();
+	}
+
+	private <P> MessageSelector wrapToMessageSelectorIfNecessary(GenericSelector<P> selector) {
 		MessageSelector messageSelector;
 		if (selector instanceof MessageSelector) {
 			messageSelector = (MessageSelector) selector;
 		}
 		else {
-			messageSelector =
-					isLambda(selector)
-							? new MethodInvokingSelector(new LambdaMessageProcessor(selector, null))
-							: new MethodInvokingSelector(selector);
+			messageSelector = new MethodInvokingSelector(selector, ClassUtils.SELECTOR_ACCEPT_METHOD);
 		}
-		this.handler.addRecipient(channelName, messageSelector);
-		return _this();
-	}
-
-	private static boolean isLambda(Object o) {
-		Class<?> aClass = o.getClass();
-		return aClass.isSynthetic() && !aClass.isAnonymousClass() && !aClass.isLocalClass();
+		return messageSelector;
 	}
 
 	/**
@@ -170,16 +167,7 @@ public RecipientListRouterSpec recipientMessageSelector(MessageChannel channel,
 	 * @return the router spec.
 	 */
 	public <P> RecipientListRouterSpec recipient(MessageChannel channel, GenericSelector<P> selector) {
-		MessageSelector messageSelector;
-		if (selector instanceof MessageSelector) {
-			messageSelector = (MessageSelector) selector;
-		}
-		else {
-			messageSelector =
-					isLambda(selector)
-							? new MethodInvokingSelector(new LambdaMessageProcessor(selector, null))
-							: new MethodInvokingSelector(selector);
-		}
+		MessageSelector messageSelector = wrapToMessageSelectorIfNecessary(selector);
 		this.handler.addRecipient(channel, messageSelector);
 		return _this();
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -16,16 +16,63 @@
 
 package org.springframework.integration.util;
 
+import java.lang.reflect.Method;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
+
+import org.springframework.integration.core.GenericSelector;
+import org.springframework.integration.transformer.GenericTransformer;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Mark Fisher
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public abstract class ClassUtils {
 
+	/**
+	 * The {@link Function#apply(Object)} method object.
+	 */
+	public static final Method FUNCTION_APPLY_METHOD =
+			ReflectionUtils.findMethod(Function.class, "apply", (Class<?>[]) null);
+
+	/**
+	 * The {@link GenericSelector#accept(Object)} method object.
+	 */
+	public static final Method SELECTOR_ACCEPT_METHOD =
+			ReflectionUtils.findMethod(GenericSelector.class, "accept", (Class<?>[]) null);
+
+	/**
+	 * The {@link GenericSelector#accept(Object)} method object.
+	 */
+	public static final Method TRANSFORMER_TRANSFORM_METHOD =
+			ReflectionUtils.findMethod(GenericTransformer.class, "transform", (Class<?>[]) null);
+
+	/**
+	 * The {@code org.springframework.integration.handler.GenericHandler#handle(Object, Map)} method object.
+	 */
+	public static final Method HANDLER_HANDLE_METHOD;
+
+	static {
+		Class<?> genericHandlerClass = null;
+		try {
+			genericHandlerClass =
+					org.springframework.util.ClassUtils.forName(
+							"org.springframework.integration.handler.GenericHandler",
+							org.springframework.util.ClassUtils.getDefaultClassLoader());
+		}
+		catch (ClassNotFoundException e) {
+			ReflectionUtils.rethrowRuntimeException(e);
+		}
+
+		HANDLER_HANDLE_METHOD = ReflectionUtils.findMethod(genericHandlerClass, "handle", (Class<?>[]) null);
+	}
+
+
 	/**
 	 * Map with primitive wrapper type as key and corresponding primitive
 	 * type as value, for example: Integer.class -> int.class.

spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.dsl.routers
+
+import assertk.assert
+import assertk.assertions.isEqualTo
+import assertk.assertions.isInstanceOf
+import assertk.assertions.isNotNull
+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.config.EnableIntegration
+import org.springframework.integration.dsl.IntegrationFlow
+import org.springframework.messaging.MessageChannel
+import org.springframework.messaging.PollableChannel
+import org.springframework.messaging.support.GenericMessage
+import org.springframework.test.annotation.DirtiesContext
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.0.5
+ */
+@SpringJUnitConfig
+@DirtiesContext
+class RouterDslTests {
+
+    @Autowired
+    @Qualifier("routerTwoSubFlows.input")
+    private lateinit var routerTwoSubFlowsInput: MessageChannel
+
+    @Autowired
+    @Qualifier("routerTwoSubFlowsOutput")
+    private lateinit var routerTwoSubFlowsOutput: PollableChannel
+
+    @Test
+    fun `route to two subflows using lambda`() {
+
+        this.routerTwoSubFlowsInput.send(GenericMessage<Any>(listOf(1, 2, 3, 4, 5, 6)))
+        val receive = this.routerTwoSubFlowsOutput.receive(10000)
+
+        val payload = receive?.payload
+
+        assert(payload).isNotNull {
+            it.isInstanceOf(List::class.java)
+        }
+
+        assert(payload).isEqualTo(listOf(3, 4, 9, 8, 15, 12))
+    }
+
+
+    @Configuration
+    @EnableIntegration
+    open class Config {
+
+        @Bean
+        open fun routerTwoSubFlows() =
+                IntegrationFlow { f ->
+                    f.split()
+                            .route<Int, Boolean>({ p -> p % 2 == 0 },
+                                    { m ->
+                                        m.subFlowMapping(true, { sf -> sf.handle<Int> { p, _ -> p * 2 } })
+                                                .subFlowMapping(false, { sf -> sf.handle<Int> { p, _ -> p * 3 } })
+                                    })
+                            .aggregate()
+                            .channel { c -> c.queue("routerTwoSubFlowsOutput") }
+                }
+
+    }
+
+}