From 946cc229ebe11e2d45f61365903f5925e97132dc Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 17 Aug 2017 12:01:03 -0400 Subject: [PATCH] INT-4330: Add support for Supplier MessageSource JIRA: https://jira.spring.io/browse/INT-4330 - updated documentation - updated test --- ...ChannelAdapterAnnotationPostProcessor.java | 29 +++++++++++++------ ...ingAnnotationsWithBeanAnnotationTests.java | 16 ++++++++-- src/reference/asciidoc/configuration.adoc | 20 +++++++++++++ 3 files changed, 53 insertions(+), 12 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java index 8c6255b24f..5727c5c182 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java @@ -19,6 +19,7 @@ package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; import java.lang.reflect.Method; import java.util.List; +import java.util.function.Supplier; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; @@ -33,12 +34,14 @@ import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.util.MessagingAnnotationUtils; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; /** * Post-processor for Methods annotated with {@link InboundChannelAdapter @InboundChannelAdapter}. * * @author Artem Bilan * @author Gary Russell + * @author Oleg Zhurakousky * @since 4.0 */ public class InboundChannelAdapterAnnotationPostProcessor extends @@ -80,20 +83,28 @@ public class InboundChannelAdapterAnnotationPostProcessor extends } private MessageSource createMessageSource(Object bean, String beanName, Method method) { + MessageSource messageSource = null; if (AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) { Object target = this.resolveTargetBeanFromMethodWithBeanAnnotation(method); - Assert.isInstanceOf(MessageSource.class, target, "The '" + this.annotationType + "' on @Bean method " + - "level is allowed only for: " + MessageSource.class.getName() + "beans"); - return (MessageSource) target; + Assert.isTrue(target instanceof MessageSource || target instanceof Supplier, "The '" + this.annotationType + "' on @Bean method " + + "level is allowed only for: " + MessageSource.class.getName() + " or " + Supplier.class.getName() + " beans"); + if (target instanceof MessageSource) { + messageSource = (MessageSource) target; + } + else { + method = ReflectionUtils.findMethod(Supplier.class, "get"); + bean = target; + } } - else { - MethodInvokingMessageSource messageSource = new MethodInvokingMessageSource(); - messageSource.setObject(bean); - messageSource.setMethod(method); + if (messageSource == null) { + MethodInvokingMessageSource methodInvokingMessageSource = new MethodInvokingMessageSource(); + methodInvokingMessageSource.setObject(bean); + methodInvokingMessageSource.setMethod(method); String messageSourceBeanName = this.generateHandlerBeanName(beanName, method); - this.beanFactory.registerSingleton(messageSourceBeanName, messageSource); - return (MessageSource) this.beanFactory.initializeBean(messageSource, messageSourceBeanName); + this.beanFactory.registerSingleton(messageSourceBeanName, methodInvokingMessageSource); + messageSource = (MessageSource) this.beanFactory.initializeBean(methodInvokingMessageSource, messageSourceBeanName); } + return messageSource; } @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java index 37ace48592..f0673446db 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java @@ -28,13 +28,14 @@ import static org.junit.Assert.fail; import java.util.ArrayList; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; +import java.util.stream.Stream; import javax.annotation.Resource; import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; - import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -84,6 +85,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Artem Bilan * @author Gary Russell + * @author Oleg Zhurakousky * @since 4.0 */ @ContextConfiguration(classes = MessagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.class) @@ -92,7 +94,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; public class MessagingAnnotationsWithBeanAnnotationTests { @Autowired - private SourcePollingChannelAdapter sourcePollingChannelAdapter; + private SourcePollingChannelAdapter[] sourcePollingChannelAdapters; @Autowired private PollableChannel discardChannel; @@ -125,7 +127,8 @@ public class MessagingAnnotationsWithBeanAnnotationTests { @Test public void testMessagingAnnotationsFlow() { - this.sourcePollingChannelAdapter.start(); + Stream.of(this.sourcePollingChannelAdapters).forEach(a -> a.start()); + //this.sourcePollingChannelAdapter.start(); for (int i = 0; i < 10; i++) { Message receive = this.discardChannel.receive(10000); assertNotNull(receive); @@ -194,6 +197,13 @@ public class MessagingAnnotationsWithBeanAnnotationTests { return () -> new GenericMessage<>(counter.incrementAndGet()); } + @Bean + @InboundChannelAdapter(value = "routerChannel", autoStartup = "false", + poller = @Poller(fixedRate = "10", maxMessagesPerPoll = "1", errorChannel = "counterErrorChannel")) + public Supplier counterMessageSupplier(final AtomicInteger counter) { + return () -> counter.incrementAndGet(); + } + @Bean public PollableChannel counterErrorChannel() { return new QueueChannel(); diff --git a/src/reference/asciidoc/configuration.adoc b/src/reference/asciidoc/configuration.adoc index d0153eb119..0311343a43 100644 --- a/src/reference/asciidoc/configuration.adoc +++ b/src/reference/asciidoc/configuration.adoc @@ -569,6 +569,26 @@ public class MyFlowConfiguration { } ---- +Starting with _version 5.0_, support is also provided for `@Bean` annotated `InboundChannelAdapter`s that return `java.util.function.Supplier` which can produce either a POJO or a `Message`: +[source,java] +---- +@Configuration +@EnableIntegration +public class MyFlowConfiguration { + + @Bean + @InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000")) + public Supplier pojoSupplier() { + return () -> "foo"; + } + + @Bean + @InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000")) + public Supplier> messageSupplier() { + return () -> new GenericMessage<>("foo"); + } +---- + The meta-annotation rules work on `@Bean` methods as well (`@MyServiceActivator` above can be applied to a `@Bean` definition). NOTE: When using these annotations on consumer `@Bean` definitions, if the bean definition returns an appropriate