diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index 8404296..666e9b7 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/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(); } } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java index 42b7c1b..641e57f 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java +++ b/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, 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 methodValue = new AtomicReference(); + 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); diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index b7cb1af..ec6274b 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/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(); } }