diff --git a/build.gradle b/build.gradle index c64cb964c2..cf046d9e99 100644 --- a/build.gradle +++ b/build.gradle @@ -133,10 +133,14 @@ subprojects { subproject -> // dependencies that are common across all java projects dependencies { - testCompile "junit:junit:$junitVersion" + testCompile ("junit:junit:$junitVersion") { + exclude group: 'org.hamcrest' + } testCompile "log4j:log4j:$log4jVersion" testCompile "org.hamcrest:hamcrest-all:$hamcrestVersion" - testCompile "org.mockito:mockito-all:$mockitoVersion" + testCompile ("org.mockito:mockito-core:$mockitoVersion") { + exclude group: 'org.hamcrest' + } testCompile "org.springframework:spring-test:$springVersion" if (!(subproject.name ==~ /.*-(core|test)/)) { testCompile project(":spring-integration-test") @@ -530,9 +534,13 @@ project('spring-integration-test') { description = 'Spring Integration Test Support' dependencies { compile project(":spring-integration-core") - compile "junit:junit:$junitVersion" + compile ("junit:junit:$junitVersion") { + exclude group: 'org.hamcrest' + } compile "org.hamcrest:hamcrest-all:$hamcrestVersion" - compile "org.mockito:mockito-core:$mockitoVersion" + compile ("org.mockito:mockito-core:$mockitoVersion") { + exclude group: 'org.hamcrest' + } compile "org.springframework:spring-context:$springVersion" compile "org.springframework:spring-test:$springVersion" } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java index 4521e32696..ec59326970 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java @@ -33,6 +33,7 @@ import org.springframework.integration.aggregator.AbstractCorrelatingMessageHand * * @author Marius Bogoevici * @author Oleg Zhurakousky + * @author Artem Bilan */ @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @@ -64,4 +65,11 @@ public @interface Aggregator { */ boolean sendPartialResultsOnExpiry() default false; + /** + * @return the {@link Poller} options for a polled endpoint + * ({@link org.springframework.integration.scheduling.PollerMetadata}). + * This attribute is an {@code array} just to allow an empty default (no poller). + * Only one {@link Poller} element is allowed. + */ + Poller[] poller() default {}; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Filter.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Filter.java index 5dc286f62b..2f20ea21cc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Filter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Filter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -36,6 +36,7 @@ import java.lang.annotation.Target; * * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan * @since 2.0 */ @Target(ElementType.METHOD) @@ -51,4 +52,11 @@ public @interface Filter { boolean discardWithinAdvice() default true; + /** + * @return the {@link Poller} options for a polled endpoint + * ({@link org.springframework.integration.scheduling.PollerMetadata}). + * This attribute is an {@code array} just to allow an empty default (no poller). + * Only one {@link Poller} element is allowed. + */ + Poller[] poller() default {}; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java new file mode 100644 index 0000000000..9adbb3fc67 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java @@ -0,0 +1,86 @@ +/* + * Copyright 2014 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.annotation; + +import static java.lang.annotation.RetentionPolicy.RUNTIME; + +import java.lang.annotation.Retention; +import java.lang.annotation.Target; + +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.scheduling.support.CronTrigger; +import org.springframework.scheduling.support.PeriodicTrigger; +import org.springframework.scheduling.Trigger; + +/** + * Provides the {@link PollerMetadata} options for the Messaging annotations for + * polled endpoints. + * It is an analogue of the XML {@code } element, but provides only simple attributes. + * If the {@link org.springframework.integration.scheduling.PollerMetadata} requires more options + * (e.g. Transactional and other Advices) or {@code initialDelay, receiveTimeout} etc, + * the {@link org.springframework.integration.scheduling.PollerMetadata} should be configured as + * a generic bean and its bean name can be specified as the {@code value} attribute of this annotation. + * In that case, the other attributes are not allowed. + *

+ * Non-reference attributes support Property Placeholder resolutions. + * + * @author Artem Bilan + * @since 4.0 + */ +@Target({}) +@Retention(RUNTIME) +public @interface Poller { + + /** + * @return The {@link PollerMetadata} bean name. + */ + String value() default ""; + + /** + * @return The {@link Trigger} bean name. + */ + String trigger() default ""; + + /** + * @return The {@link org.springframework.core.task.TaskExecutor} bean name. + */ + String taskExecutor() default ""; + + /** + * @return The maximum number of messages to receive for each poll. + * Can be specified as 'property placeholder', e.g. {@code ${poller.maxMessagesPerPoll}}. + */ + String maxMessagesPerPoll() default ""; + + /** + * @return The fixed delay in milliseconds to create the {@link PeriodicTrigger}. + * Can be specified as 'property placeholder', e.g. {@code ${poller.fixedDelay}}. + */ + String fixedDelay() default ""; + + /** + * @return The fixed rate in milliseconds to create the {@link PeriodicTrigger} with {@code fixedRate}. + * Can be specified as 'property placeholder', e.g. {@code ${poller.fixedRate}}. + */ + String fixedRate() default ""; + + /** + * @return The cron expression to create the {@link CronTrigger}. + * Can be specified as 'property placeholder', e.g. {@code ${poller.cron}}. + */ + String cron() default ""; +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Router.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Router.java index 305d988f29..8821cf3441 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Router.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Router.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2014 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. @@ -41,6 +41,7 @@ import java.lang.annotation.Target; * to resolve each channel name with the Channel Registry. * * @author Mark Fisher + * @author Artem Bilan */ @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @@ -52,4 +53,11 @@ public @interface Router { String defaultOutputChannel() default ""; + /** + * @return the {@link Poller} options for a polled endpoint + * ({@link org.springframework.integration.scheduling.PollerMetadata}). + * This attribute is an {@code array} just to allow an empty default (no poller). + * Only one {@link Poller} element is allowed. + */ + Poller[] poller() default {}; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java index af38b3b7b1..b86ba3c120 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -39,6 +39,7 @@ import java.lang.annotation.Target; * * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan */ @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @@ -52,4 +53,11 @@ public @interface ServiceActivator { String[] adviceChain() default {}; + /** + * @return the {@link Poller} options for a polled endpoint + * ({@link org.springframework.integration.scheduling.PollerMetadata}). + * This attribute is an {@code array} just to allow an empty default (no poller). + * Only one {@link Poller} element is allowed. + */ + Poller[] poller() default {}; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java index 7f56adfb82..d709e74604 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -39,6 +39,7 @@ import java.lang.annotation.Target; * * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan */ @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @@ -51,4 +52,11 @@ public @interface Splitter { String[] adviceChain() default {}; + /** + * @return the {@link Poller} options for a polled endpoint + * ({@link org.springframework.integration.scheduling.PollerMetadata}). + * This attribute is an {@code array} just to allow an empty default (no poller). + * Only one {@link Poller} element is allowed. + */ + Poller[] poller() default {}; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Transformer.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Transformer.java index f536fb5da7..07c6e9b158 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Transformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Transformer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -29,6 +29,7 @@ import java.lang.annotation.Target; * * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan */ @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @@ -42,4 +43,11 @@ public @interface Transformer { String[] adviceChain() default {}; + /** + * @return the {@link Poller} options for a polled endpoint + * ({@link org.springframework.integration.scheduling.PollerMetadata}). + * This attribute is an {@code array} just to allow an empty default (no poller). + * Only one {@link Poller} element is allowed. + */ + Poller[] poller() default {}; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index 048a26b084..f4b45e0df4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -20,6 +20,7 @@ import java.lang.annotation.Annotation; import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.List; import org.aopalliance.aop.Advice; @@ -29,20 +30,30 @@ import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.annotation.Order; +import org.springframework.core.env.Environment; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.annotation.Poller; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.config.IntegrationConfigUtils; import org.springframework.integration.context.Orderable; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.messaging.core.DestinationResolver; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.support.CronTrigger; +import org.springframework.scheduling.support.PeriodicTrigger; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; +import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** @@ -61,12 +72,15 @@ public abstract class AbstractMethodAnnotationPostProcessor channelResolver; - public AbstractMethodAnnotationPostProcessor(ListableBeanFactory beanFactory) { + public AbstractMethodAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { Assert.notNull(beanFactory, "BeanFactory must not be null"); this.beanFactory = beanFactory; + this.environment = environment; this.channelResolver = new BeanFactoryChannelResolver(beanFactory); } @@ -108,9 +122,7 @@ public abstract class AbstractMethodAnnotationPostProcessor { - private final BeanFactory beanFactory; - - public AggregatorAnnotationPostProcessor(ListableBeanFactory beanFactory) { - super(beanFactory); - this.beanFactory = beanFactory; + public AggregatorAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { + super(beanFactory, environment); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/FilterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/FilterAnnotationPostProcessor.java index d2c4f774be..0a3685850f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/FilterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/FilterAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -19,6 +19,7 @@ package org.springframework.integration.config.annotation; import java.lang.reflect.Method; import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.core.env.Environment; import org.springframework.integration.annotation.Filter; import org.springframework.messaging.MessageHandler; import org.springframework.integration.filter.MessageFilter; @@ -35,8 +36,8 @@ import org.springframework.util.StringUtils; */ public class FilterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { - public FilterAnnotationPostProcessor(ListableBeanFactory beanFactory) { - super(beanFactory); + public FilterAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { + super(beanFactory, environment); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java index a35d2a0668..03559384e1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2014 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. @@ -40,9 +40,11 @@ import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationListener; +import org.springframework.context.EnvironmentAware; import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.core.env.Environment; import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.Filter; import org.springframework.integration.annotation.Router; @@ -59,16 +61,19 @@ import org.springframework.util.StringUtils; /** * A {@link BeanPostProcessor} implementation that processes method-level * messaging annotations such as @Transformer, @Splitter, @Router, and @Filter. - * + * * @author Mark Fisher * @author Marius Bogoevici + * @author Artem Bilan */ -public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, InitializingBean, Lifecycle, ApplicationListener { +public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, + InitializingBean, Lifecycle, ApplicationListener, EnvironmentAware { private final Log logger = LogFactory.getLog(this.getClass()); private volatile ConfigurableListableBeanFactory beanFactory; + private Environment environment; private final Map, MethodAnnotationPostProcessor> postProcessors = new HashMap, MethodAnnotationPostProcessor>(); @@ -86,14 +91,19 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean this.beanFactory = (ConfigurableListableBeanFactory) beanFactory; } + @Override + public void setEnvironment(Environment environment) { + this.environment = environment; + } + public void afterPropertiesSet() { Assert.notNull(this.beanFactory, "BeanFactory must not be null"); - postProcessors.put(Filter.class, new FilterAnnotationPostProcessor(this.beanFactory)); - postProcessors.put(Router.class, new RouterAnnotationPostProcessor(this.beanFactory)); - postProcessors.put(Transformer.class, new TransformerAnnotationPostProcessor(this.beanFactory)); - postProcessors.put(ServiceActivator.class, new ServiceActivatorAnnotationPostProcessor(this.beanFactory)); - postProcessors.put(Splitter.class, new SplitterAnnotationPostProcessor(this.beanFactory)); - postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.beanFactory)); + postProcessors.put(Filter.class, new FilterAnnotationPostProcessor(this.beanFactory, this.environment)); + postProcessors.put(Router.class, new RouterAnnotationPostProcessor(this.beanFactory, this.environment)); + postProcessors.put(Transformer.class, new TransformerAnnotationPostProcessor(this.beanFactory, this.environment)); + postProcessors.put(ServiceActivator.class, new ServiceActivatorAnnotationPostProcessor(this.beanFactory, this.environment)); + postProcessors.put(Splitter.class, new SplitterAnnotationPostProcessor(this.beanFactory, this.environment)); + postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.beanFactory, this.environment)); } public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { @@ -185,6 +195,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean return name; } + public void onApplicationEvent(ApplicationEvent event) { for (ApplicationListener listener : listeners) { try { @@ -199,7 +210,6 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean } } - // Lifecycle implementation public boolean isRunning() { @@ -223,5 +233,4 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean } this.running = false; } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java index c5112dbb34..efc880c283 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2014 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. @@ -19,6 +19,7 @@ package org.springframework.integration.config.annotation; import java.lang.reflect.Method; import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.core.env.Environment; import org.springframework.messaging.MessageChannel; import org.springframework.integration.annotation.Router; import org.springframework.messaging.MessageHandler; @@ -33,8 +34,8 @@ import org.springframework.util.StringUtils; */ public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { - public RouterAnnotationPostProcessor(ListableBeanFactory beanFactory) { - super(beanFactory); + public RouterAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { + super(beanFactory, environment); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java index b6acf65ae2..00ece70c8c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2014 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. @@ -19,6 +19,7 @@ package org.springframework.integration.config.annotation; import java.lang.reflect.Method; import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.core.env.Environment; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.messaging.MessageHandler; import org.springframework.integration.handler.ServiceActivatingHandler; @@ -31,8 +32,8 @@ import org.springframework.util.StringUtils; */ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { - public ServiceActivatorAnnotationPostProcessor(ListableBeanFactory beanFactory) { - super(beanFactory); + public ServiceActivatorAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { + super(beanFactory, environment); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java index e506154a61..de72616593 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2014 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. @@ -19,6 +19,7 @@ package org.springframework.integration.config.annotation; import java.lang.reflect.Method; import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.core.env.Environment; import org.springframework.integration.annotation.Splitter; import org.springframework.messaging.MessageHandler; import org.springframework.integration.splitter.MethodInvokingSplitter; @@ -31,8 +32,8 @@ import org.springframework.util.StringUtils; */ public class SplitterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { - public SplitterAnnotationPostProcessor(ListableBeanFactory beanFactory) { - super(beanFactory); + public SplitterAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { + super(beanFactory, environment); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java index d2db77dab2..7de1650f0b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2014 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. @@ -19,6 +19,7 @@ package org.springframework.integration.config.annotation; import java.lang.reflect.Method; import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.core.env.Environment; import org.springframework.integration.annotation.Transformer; import org.springframework.messaging.MessageHandler; import org.springframework.integration.transformer.MessageTransformingHandler; @@ -32,8 +33,8 @@ import org.springframework.util.StringUtils; */ public class TransformerAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { - public TransformerAnnotationPostProcessor(ListableBeanFactory beanFactory) { - super(beanFactory); + public TransformerAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { + super(beanFactory, environment); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java index 5c5a21dde7..228e697dbc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java @@ -39,6 +39,11 @@ public class PollerMetadata { public static final String DEFAULT_POLLER_METADATA_BEAN_NAME = "org.springframework.integration.context.defaultPollerMetadata"; + /** + * A convenient short alias for the global default poller bean name. + */ + public static final String DEFAULT_POLLER = DEFAULT_POLLER_METADATA_BEAN_NAME; + private volatile Trigger trigger; private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MethodInvokingTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MethodInvokingTransformer.java index 53baf6c808..74f9321d8d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MethodInvokingTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MethodInvokingTransformer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -21,6 +21,7 @@ import java.lang.reflect.Method; import org.springframework.integration.annotation.Transformer; import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.handler.MethodInvokingMessageProcessor; +import org.springframework.util.Assert; /** * A Message Transformer implementation that invokes the specified method @@ -35,6 +36,7 @@ public class MethodInvokingTransformer extends AbstractMessageProcessingTransfor public MethodInvokingTransformer(Object object, Method method) { super(new MethodInvokingMessageProcessor(object, method)); + Assert.state(!Void.class.isAssignableFrom(method.getReturnType()), "'transformer' method must not be 'void'."); } public MethodInvokingTransformer(Object object, String methodName) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java index fe400f0af2..8b7cebcc6f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java @@ -21,6 +21,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -33,6 +34,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.config.AbstractFactoryBean; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.AnnotationConfigApplicationContext; @@ -50,6 +52,7 @@ import org.springframework.integration.annotation.IntegrationComponentScan; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.MessagingGateway; import org.springframework.integration.annotation.Payload; +import org.springframework.integration.annotation.Poller; import org.springframework.integration.annotation.Publisher; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.annotation.Transformer; @@ -63,8 +66,10 @@ import org.springframework.integration.config.EnableMessageHistory; import org.springframework.integration.config.EnablePublisher; import org.springframework.integration.config.GlobalChannelInterceptor; import org.springframework.integration.config.IntegrationConverter; +import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.history.MessageHistoryConfigurer; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.MutableMessageBuilder; import org.springframework.integration.test.util.TestUtils; @@ -74,6 +79,9 @@ import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.GenericMessage; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.support.CronTrigger; +import org.springframework.scheduling.support.PeriodicTrigger; import org.springframework.stereotype.Component; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -92,7 +100,34 @@ public class EnableIntegrationTests { private ApplicationContext context; @Autowired - private MessageChannel input; + private PollableChannel input; + + @Autowired + @Qualifier("enableIntegrationTests.AnnotationTestService.handle.serviceActivator") + private PollingConsumer serviceActivatorEndpoint; + + @Autowired + @Qualifier("enableIntegrationTests.AnnotationTestService.handle1.serviceActivator") + private PollingConsumer serviceActivatorEndpoint1; + + @Autowired + @Qualifier("enableIntegrationTests.AnnotationTestService.handle2.serviceActivator") + private PollingConsumer serviceActivatorEndpoint2; + + @Autowired + @Qualifier("enableIntegrationTests.AnnotationTestService.handle3.serviceActivator") + private PollingConsumer serviceActivatorEndpoint3; + + @Autowired + @Qualifier("enableIntegrationTests.AnnotationTestService.handle4.serviceActivator") + private PollingConsumer serviceActivatorEndpoint4; + + @Autowired + @Qualifier("enableIntegrationTests.AnnotationTestService.transform.transformer") + private PollingConsumer transformer; + + @Autowired + private Trigger myTrigger; @Autowired private QueueChannel output; @@ -126,6 +161,38 @@ public class EnableIntegrationTests { @Test public void testAnnotatedServiceActivator() { + assertEquals(10L, TestUtils.getPropertyValue(this.serviceActivatorEndpoint, "maxMessagesPerPoll")); + + Trigger trigger = TestUtils.getPropertyValue(this.serviceActivatorEndpoint, "trigger", Trigger.class); + assertThat(trigger, Matchers.instanceOf(PeriodicTrigger.class)); + assertEquals(100L, TestUtils.getPropertyValue(trigger, "period")); + assertFalse(TestUtils.getPropertyValue(trigger, "fixedRate", Boolean.class)); + + trigger = TestUtils.getPropertyValue(this.serviceActivatorEndpoint1, "trigger", Trigger.class); + assertThat(trigger, Matchers.instanceOf(PeriodicTrigger.class)); + assertEquals(100L, TestUtils.getPropertyValue(trigger, "period")); + assertTrue(TestUtils.getPropertyValue(trigger, "fixedRate", Boolean.class)); + + trigger = TestUtils.getPropertyValue(this.serviceActivatorEndpoint2, "trigger", Trigger.class); + assertThat(trigger, Matchers.instanceOf(CronTrigger.class)); + assertEquals("0 5 7 * * *", TestUtils.getPropertyValue(trigger, "sequenceGenerator.expression")); + + trigger = TestUtils.getPropertyValue(this.serviceActivatorEndpoint3, "trigger", Trigger.class); + assertThat(trigger, Matchers.instanceOf(PeriodicTrigger.class)); + assertEquals(11L, TestUtils.getPropertyValue(trigger, "period")); + assertFalse(TestUtils.getPropertyValue(trigger, "fixedRate", Boolean.class)); + + trigger = TestUtils.getPropertyValue(this.serviceActivatorEndpoint4, "trigger", Trigger.class); + assertThat(trigger, Matchers.instanceOf(PeriodicTrigger.class)); + assertEquals(1000L, TestUtils.getPropertyValue(trigger, "period")); + assertFalse(TestUtils.getPropertyValue(trigger, "fixedRate", Boolean.class)); + assertSame(this.myTrigger, trigger); + + trigger = TestUtils.getPropertyValue(this.transformer, "trigger", Trigger.class); + assertThat(trigger, Matchers.instanceOf(PeriodicTrigger.class)); + assertEquals(10L, TestUtils.getPropertyValue(trigger, "period")); + assertFalse(TestUtils.getPropertyValue(trigger, "fixedRate", Boolean.class)); + this.input.send(MessageBuilder.withPayload("Foo").build()); Message interceptedMessage = this.wireTapChannel.receive(1000); @@ -226,8 +293,33 @@ public class EnableIntegrationTests { public static class ContextConfiguration { @Bean - public MessageChannel input() { - return new DirectChannel(); + public QueueChannel input() { + return new QueueChannel(); + } + + @Bean + public QueueChannel input1() { + return new QueueChannel(); + } + + @Bean + public QueueChannel input2() { + return new QueueChannel(); + } + + @Bean + public QueueChannel input3() { + return new QueueChannel(); + } + + @Bean + public QueueChannel input4() { + return new QueueChannel(); + } + + @Bean + public Trigger myTrigger() { + return new PeriodicTrigger(1000L); } @Bean @@ -331,8 +423,8 @@ public class EnableIntegrationTests { } @Bean - public DirectChannel gatewayChannel() { - return new DirectChannel(); + public PollableChannel gatewayChannel() { + return new QueueChannel(); } @Bean @@ -349,6 +441,20 @@ public class EnableIntegrationTests { return channel; } + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerMetadata defaultPoller() { + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(new PeriodicTrigger(10)); + return pollerMetadata; + } + + @Bean + public PollerMetadata myPoller() { + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(new PeriodicTrigger(11)); + return pollerMetadata; + } + @Bean @IntegrationConverter public SerializingConverter serializingConverter() { @@ -378,13 +484,55 @@ public class EnableIntegrationTests { @MessageEndpoint public static class AnnotationTestService { - @ServiceActivator(inputChannel = "input", outputChannel = "output") + @ServiceActivator(inputChannel = "input", outputChannel = "output", + poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}", fixedDelay = "${poller.interval}")) @Publisher @Payload("#args[0].toLowerCase()") public String handle(String payload) { return payload.toUpperCase(); } + @ServiceActivator(inputChannel = "input1", outputChannel = "output", + poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}", fixedRate = "${poller.interval}")) + @Publisher + @Payload("#args[0].toLowerCase()") + public String handle1(String payload) { + return payload.toUpperCase(); + } + + @ServiceActivator(inputChannel = "input2", outputChannel = "output", + poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}", cron = "0 5 7 * * *")) + @Publisher + @Payload("#args[0].toLowerCase()") + public String handle2(String payload) { + return payload.toUpperCase(); + } + + @ServiceActivator(inputChannel = "input3", outputChannel = "output", poller = @Poller("myPoller")) + @Publisher + @Payload("#args[0].toLowerCase()") + public String handle3(String payload) { + return payload.toUpperCase(); + } + + @ServiceActivator(inputChannel = "input4", outputChannel = "output", + poller = @Poller(trigger = "myTrigger")) + @Publisher + @Payload("#args[0].toLowerCase()") + public String handle4(String payload) { + return payload.toUpperCase(); + } + + /* + * This is an error because input5 is not defined and is therefore a DirectChannel. + */ + /*@ServiceActivator(inputChannel = "input5", outputChannel = "output", poller = @Poller("defaultPollerMetadata")) + @Publisher + @Payload("#args[0].toLowerCase()") + public String handle5(String payload) { + return payload.toUpperCase(); + }*/ + @Transformer(inputChannel = "gatewayChannel") public String transform(Message message) { assertTrue(message.getHeaders().containsKey("foo")); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.properties b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.properties index a843590afd..579f83e22f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.properties +++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.properties @@ -1 +1,3 @@ message.history.tracked.components=input, publishedChannel, *AnnotationTestService* +poller.maxMessagesPerPoll=10 +poller.interval=100 diff --git a/src/reference/docbook/configuration.xml b/src/reference/docbook/configuration.xml index 9cb6db6218..b6bb75543f 100644 --- a/src/reference/docbook/configuration.xml +++ b/src/reference/docbook/configuration.xml @@ -314,19 +314,8 @@ public class FooService { location (and besides the namespace-based XML configuration is not very verbose). If you do prefer to provide channels with the annotations however, you just need to enable a SI Annotations BeanPostProcessor. The following element should be added: ]]> - - When configuring the "inputChannel" and "outputChannel" with annotations, the "inputChannel" - must be a reference to a SubscribableChannel instance. - Otherwise, it would be necessary to also provide the full poller configuration via annotations, and those - settings (e.g. the trigger for scheduling the poller) should be externalized rather than hard-coded within - an annotation. If the input channel that you want to receive Messages from is indeed a - PollableChannel instance, one option to consider is the Messaging Bridge. - Spring Integration's "bridge" element can be used to connect a PollableChannel directly to a - SubscribableChannel. Then, the polling metadata is externally configured, but the annotation option is - still available. For more detail see . - - + The processing of these annotations creates the same beans (EventDrivenConsumers and MessageHandlers) as with similar xml components. The bean names are generated with this pattern: @@ -335,6 +324,67 @@ public class FooService { the MessageHandler bean. The MessageHandlers are also eligible to be tracked by . + + @Poller + + + Before Spring Integration 4.0, the above Messaging Annotations required that the + inputChannel was a reference to a SubscribableChannel. + For PollableChannels there was need to use a <int:bridge/>, + to configure a <int:poller/> to make the composite endpoint - a PollingConsumer. + Starting with version 4.0, + the @Poller annotation has been introduced to allow the configuration of + poller attributes directly on the above Messaging Annotations: + + This annotation provides only simple PollerMetadata options. The + @Poller's attributes maxMessagesPerPoll, fixedDelay, + fixedRate and cron can be configured with property-placeholders. + If it is necessary to provide more polling options (e.g. transaction, advice-chain, error-handler), the + PollerMetadata should be configured as a generic bean with its bean name used for + @Poller's value attribute. In this case, no other attributes are allowed + (they would be specified on the PollerMetadata bean). + Note, if inputChannel is PollableChannel and + no @Poller is configured, + the default PollerMetadata will be used, if it is present in the application context. + To declare the default poller using @Configuration, use: + + With this endpoint using the default poller: + + To use a named poller, use: + + With this endpoint using the default poller: + + Also see . diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index e3b147e62f..1cdfe9c7f1 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -142,6 +142,16 @@ For more information, see and . +
+ @Poller + + Annotation-based messaging configuration can now have a poller attribute. + This means that methods annotated with (@ServiceActivator, + @Aggregator etc.) can now use an inputChannel that + is a reference to a PollableChannel. + For more information, see . + +