DSL: Tests for @Autowired handlers

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java

@@ -658,7 +658,7 @@ public final class IntegrationFlowBuilder {
 
 	private static boolean isLambda(Object o) {
 		Class<?> aClass = o.getClass();
-		return aClass.isSynthetic() && !aClass.isAnonymousClass() && aClass.getDeclaredMethods().length == 1;
+		return aClass.isSynthetic() && !aClass.isAnonymousClass() && !aClass.isLocalClass();
 	}
 
 }

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java

@@ -17,7 +17,9 @@
 package org.springframework.integration.dsl;
 
 import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
@@ -30,6 +32,7 @@ import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Artem Bilan
@@ -50,10 +53,24 @@ class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFactoryAwa
 	public LambdaMessageProcessor(Object target, Class<?> payloadType) {
 		Assert.notNull(target);
 		this.target = target;
-		Method[] declaredMethods = target.getClass().getDeclaredMethods();
-		Assert.isTrue(declaredMethods.length == 1, "LambdaMessageProcessor is applicable for inline or lambda" +
+		final AtomicReference<Method> methodValue = new AtomicReference<Method>();
+		ReflectionUtils.doWithMethods(target.getClass(), new ReflectionUtils.MethodCallback() {
+			@Override
+			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+				methodValue.set(method);
+			}
+		}, new ReflectionUtils.MethodFilter() {
+			@Override
+			public boolean matches(Method method) {
+				return !method.isBridge() && method.getDeclaringClass() != Object.class &&
+						Modifier.isPublic(method.getModifiers()) && !Modifier.isStatic(method.getModifiers());
+			}
+		});
+
+		Assert.notNull(methodValue.get(), "LambdaMessageProcessor is applicable for inline or lambda " +
 				"classes with single method - functional interfaces implementations.");
-		this.method = declaredMethods[0];
+
+		this.method = methodValue.get();
 		this.method.setAccessible(true);
 		this.parameterTypes = this.method.getParameterTypes();
 		this.payloadType = TypeDescriptor.valueOf(payloadType);

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java

@@ -98,6 +98,7 @@ import org.springframework.integration.file.DefaultFileNameGenerator;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.FileWritingMessageHandler;
 import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
 import org.springframework.integration.mongodb.store.MongoDbChannelMessageStore;
 import org.springframework.integration.router.MethodInvokingRouter;
@@ -814,7 +815,7 @@ public class IntegrationFlowTests {
 		message = MessageBuilder.withPayload("31").setPriority(3).build();
 		this.priorityChannel.send(message);
 
-		Thread.sleep(1000);
+		Thread.sleep(2000);
 
 		Message<?> receive = this.priorityReplyChannel.receive(1000);
 		assertNotNull(receive);
@@ -921,7 +922,7 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow flow1() {
-			return IntegrationFlows.from(this.integerMessageSource(), c -> c.poller(Pollers.fixedRate(100)))
+			return IntegrationFlows.from(this.integerMessageSource(), c -> c.poller(Pollers.fixedRate(100, 1000)))
 					.fixedSubscriberChannel("integerChannel")
 					.transform("payload.toString()")
 					.channel(MessageChannels.queue("flow1QueueChannel"))
@@ -996,9 +997,8 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) {
-			return IntegrationFlows.from(MessageChannels.priority("priorityChannel",
-					mongoDbChannelMessageStore, "priorityGroup").interceptor())
-					.bridge(s -> s.poller(Pollers.fixedDelay(1000, 2000)))
+			return IntegrationFlows.from(MessageChannels.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup"))
+					.bridge(s -> s.poller(Pollers.fixedDelay(1000, 4000)))
 					.channel(MessageChannels.queue("priorityReplyChannel"))
 					.get();
 		}
@@ -1153,7 +1153,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow methodInvokingFlow() {
 			return IntegrationFlows.from("methodInvokingInput")
-					.transform(this.greetingService::greeting)
+					.handle(Message.class, (p, h) -> this.greetingService.handleRequestMessage(p))
 					.get();
 		}
 
@@ -1365,14 +1365,15 @@ public class IntegrationFlowTests {
 		}
 	}
 
-	@Component("greetingService")
-	public static class GreetingService {
+	@Service
+	public static class GreetingService extends AbstractReplyProducingMessageHandler {
 
 		@Autowired
 		private WorldService worldService;
 
-		public String greeting(String payload) {
-			return "Hello " + this.worldService.world() + " and " + payload;
+		@Override
+		protected Object handleRequestMessage(Message<?> requestMessage) {
+			return "Hello " + this.worldService.world() + " and " + requestMessage.getPayload();
 		}
 	}