INT-3351: Add @Poller for Messaging Annotations

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

INT-3351: Polishing according PR comments

* Add `property-placeholder` support for `@Poller`
* Change `ref` to `value`
* Add JavaDoc for `array` workaround
* Add docs

INT-3351: Polishing and fixes

INT-3351 More Tests

INT-3351: PR comments

INT-3351 Final Polishing
This commit is contained in:
Artem Bilan
2014-04-02 21:13:43 +03:00
committed by Gary Russell
parent 8a0d3803d9
commit 8ffc86f3d2
22 changed files with 524 additions and 71 deletions

View File

@@ -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"
}

View File

@@ -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 {};
}

View File

@@ -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 {};
}

View File

@@ -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 <poller/>} 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.
* <p>
* 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 "";
}

View File

@@ -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 {};
}

View File

@@ -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 {};
}

View File

@@ -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 {};
}

View File

@@ -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 {};
}

View File

@@ -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<T extends Annotation
protected final BeanFactory beanFactory;
protected final Environment environment;
protected final DestinationResolver<MessageChannel> 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<T extends Annotation
adviceChain.add((Advice) adviceChainBean);
}
else if (adviceChainBean instanceof Advice[]) {
for (Advice advice : (Advice[]) adviceChainBean) {
adviceChain.add(advice);
}
Collections.addAll(adviceChain, (Advice[]) adviceChainBean);
}
else if (adviceChainBean instanceof Collection) {
@SuppressWarnings("unchecked")
@@ -128,10 +140,6 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
}
}
protected boolean shouldCreateEndpoint(T annotation) {
return (StringUtils.hasText((String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE)));
}
private AbstractEndpoint createEndpoint(MessageHandler handler, T annotation) {
AbstractEndpoint endpoint = null;
String inputChannelName = (String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE);
@@ -149,9 +157,84 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
}
}
Assert.notNull(inputChannel, "failed to resolve inputChannel '" + inputChannelName + "'");
Assert.isInstanceOf(SubscribableChannel.class, inputChannel,
"The input channel for an Annotation-based endpoint must be a SubscribableChannel.");
endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);
if (inputChannel instanceof PollableChannel) {
PollerMetadata pollerMetadata = null;
Poller[] pollers = (Poller[]) AnnotationUtils.getValue(annotation, "poller");
if (!ObjectUtils.isEmpty(pollers)) {
Assert.state(pollers.length == 1, "The 'poller' for an Annotation-based endpoint can have only one '@Poller'.");
Poller poller = pollers[0];
String ref = poller.value();
String triggerRef = poller.trigger();
String executorRef = poller.taskExecutor();
String fixedDelayValue = this.environment.resolvePlaceholders(poller.fixedDelay());
String fixedRateValue = this.environment.resolvePlaceholders(poller.fixedRate());
String maxMessagesPerPollValue = this.environment.resolvePlaceholders(poller.maxMessagesPerPoll());
String cron = this.environment.resolvePlaceholders(poller.cron());
if (StringUtils.hasText(ref)) {
Assert.state(!StringUtils.hasText(triggerRef) && !StringUtils.hasText(executorRef) && !StringUtils.hasText(cron)
&& !StringUtils.hasText(fixedDelayValue) && !StringUtils.hasText(fixedRateValue)
&& !StringUtils.hasText(maxMessagesPerPollValue),
"The '@Poller' 'ref' attribute is mutually exclusive with other attributes.");
pollerMetadata = this.beanFactory.getBean(ref, PollerMetadata.class);
}
else {
pollerMetadata = new PollerMetadata();
if (StringUtils.hasText(maxMessagesPerPollValue)) {
pollerMetadata.setMaxMessagesPerPoll(Long.parseLong(maxMessagesPerPollValue));
}
if (StringUtils.hasText(executorRef)) {
pollerMetadata.setTaskExecutor(this.beanFactory.getBean(executorRef, TaskExecutor.class));
}
Trigger trigger = null;
if (StringUtils.hasText(triggerRef)) {
Assert.state(!StringUtils.hasText(cron) && !StringUtils.hasText(fixedDelayValue) && !StringUtils.hasText(fixedRateValue),
"The '@Poller' 'trigger' attribute is mutually exclusive with other attributes.");
trigger = this.beanFactory.getBean(triggerRef, Trigger.class);
}
else if (StringUtils.hasText(cron)) {
Assert.state(!StringUtils.hasText(fixedDelayValue) && !StringUtils.hasText(fixedRateValue),
"The '@Poller' 'cron' attribute is mutually exclusive with other attributes.");
trigger = new CronTrigger(cron);
}
else if (StringUtils.hasText(fixedDelayValue)) {
Assert.state(!StringUtils.hasText(fixedRateValue),
"The '@Poller' 'fixedDelay' attribute is mutually exclusive with other attributes.");
trigger = new PeriodicTrigger(Long.parseLong(fixedDelayValue));
}
else if (StringUtils.hasText(fixedRateValue)) {
trigger = new PeriodicTrigger(Long.parseLong(fixedRateValue));
((PeriodicTrigger) trigger).setFixedRate(true);
}
//'Trigger' can be null. 'PollingConsumer' does fallback to the 'new PeriodicTrigger(10)'.
pollerMetadata.setTrigger(trigger);
}
}
else {
pollerMetadata = PollerMetadata.getDefaultPollerMetadata(this.beanFactory);
Assert.notNull(pollerMetadata, "No poller has been defined for Annotation-based endpoint, " +
"and no default poller is available within the context.");
}
PollingConsumer pollingConsumer = new PollingConsumer((PollableChannel) inputChannel, handler);
pollingConsumer.setTaskExecutor(pollerMetadata.getTaskExecutor());
pollingConsumer.setTrigger(pollerMetadata.getTrigger());
pollingConsumer.setAdviceChain(pollerMetadata.getAdviceChain());
pollingConsumer.setMaxMessagesPerPoll(pollerMetadata.getMaxMessagesPerPoll());
pollingConsumer.setErrorHandler(pollerMetadata.getErrorHandler());
pollingConsumer.setReceiveTimeout(pollerMetadata.getReceiveTimeout());
pollingConsumer.setTransactionSynchronizationFactory(pollerMetadata.getTransactionSynchronizationFactory());
endpoint = pollingConsumer;
}
else {
Poller[] pollers = (Poller[]) AnnotationUtils.getValue(annotation, "poller");
Assert.state(ObjectUtils.isEmpty(pollers), "A '@Poller' should not be specified for for Annotation-based endpoint, " +
"since '" + inputChannel + "' is a SubscribableChannel (not pollable).");
endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);
}
}
return endpoint;
}

View File

@@ -20,9 +20,9 @@ import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.concurrent.atomic.AtomicReference;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.env.Environment;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
@@ -45,11 +45,8 @@ import org.springframework.util.StringUtils;
*/
public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Aggregator> {
private final BeanFactory beanFactory;
public AggregatorAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
this.beanFactory = beanFactory;
public AggregatorAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) {
super(beanFactory, environment);
}

View File

@@ -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<Filter> {
public FilterAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
public FilterAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) {
super(beanFactory, environment);
}

View File

@@ -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<ApplicationEvent> {
public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware,
InitializingBean, Lifecycle, ApplicationListener<ApplicationEvent>, EnvironmentAware {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile ConfigurableListableBeanFactory beanFactory;
private Environment environment;
private final Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> postProcessors =
new HashMap<Class<? extends Annotation>, 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<ApplicationEvent> 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;
}
}

View File

@@ -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<Router> {
public RouterAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
public RouterAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) {
super(beanFactory, environment);
}

View File

@@ -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<ServiceActivator> {
public ServiceActivatorAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
public ServiceActivatorAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) {
super(beanFactory, environment);
}

View File

@@ -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<Splitter> {
public SplitterAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
public SplitterAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) {
super(beanFactory, environment);
}

View File

@@ -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<Transformer> {
public TransformerAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
public TransformerAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) {
super(beanFactory, environment);
}

View File

@@ -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;

View File

@@ -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>(object, method));
Assert.state(!Void.class.isAssignableFrom(method.getReturnType()), "'transformer' method must not be 'void'.");
}
public MethodInvokingTransformer(Object object, String methodName) {

View File

@@ -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<String> message) {
assertTrue(message.getHeaders().containsKey("foo"));

View File

@@ -1 +1,3 @@
message.history.tracked.components=input, publishedChannel, *AnnotationTestService*
poller.maxMessagesPerPoll=10
poller.interval=100

View File

@@ -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: <programlisting language="xml"><![CDATA[<int:annotation-config/>]]></programlisting>
<note>
When configuring the "inputChannel" and "outputChannel" with annotations, the "inputChannel"
<emphasis>must</emphasis> be a reference to a <interfacename>SubscribableChannel</interfacename> 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
<interfacename>PollableChannel</interfacename> 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 <xref linkend="bridge"/>.
</note>
</para>
<para>
<para>
The processing of these annotations creates the same beans (<classname>EventDrivenConsumer</classname>s and
<interfacename>MessageHandler</interfacename>s) as with similar xml components. The bean names are generated
with this pattern:
@@ -335,6 +324,67 @@ public class FooService {
the <interfacename>MessageHandler</interfacename> bean. The
<interfacename>MessageHandler</interfacename>s are also eligible to be tracked by <xref linkend="message-history"/>.
</para>
<para>
<emphasis role="bold">@Poller</emphasis>
</para>
<para>
Before <emphasis>Spring Integration 4.0</emphasis>, the above Messaging Annotations required that the
<code>inputChannel</code> was a reference to a <interfacename>SubscribableChannel</interfacename>.
For <interfacename>PollableChannel</interfacename>s there was need to use a <code>&lt;int:bridge/&gt;</code>,
to configure a <code>&lt;int:poller/&gt;</code> to make the composite endpoint - a <classname>PollingConsumer</classname>.
Starting with <emphasis>version 4.0</emphasis>,
the <interfacename>@Poller</interfacename> annotation has been introduced to allow the configuration of
<code>poller</code> attributes directly on the above Messaging Annotations:
<programlisting language="java"><![CDATA[public class AnnotationService {
@Transformer(inputChannel = "input", outputChannel = "output",
poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}", fixedDelay = "${poller.fixedDelay}"))
public String handle(String payload) {
...
}
}]]></programlisting>
This annotation provides only simple <classname>PollerMetadata</classname> options. The
<interfacename>@Poller</interfacename>'s attributes <code>maxMessagesPerPoll</code>, <code>fixedDelay</code>,
<code>fixedRate</code> and <code>cron</code> can be configured with <emphasis>property-placeholder</emphasis>s.
If it is necessary to provide more polling options (e.g. transaction, advice-chain, error-handler), the
<classname>PollerMetadata</classname> should be configured as a generic bean with its bean name used for
<interfacename>@Poller</interfacename>'s <code>value</code> attribute. In this case, no other attributes are allowed
(they would be specified on the <classname>PollerMetadata</classname> bean).
Note, if <code>inputChannel</code> is <interfacename>PollableChannel</interfacename> and
no <interfacename>@Poller</interfacename> is configured,
the default <classname>PollerMetadata</classname> will be used, if it is present in the application context.
To declare the default poller using <code>@Configuration</code>, use:
<programlisting language="java"><![CDATA[@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata defaultPoller() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(10));
return pollerMetadata;
}]]></programlisting>
With this endpoint using the default poller:
<programlisting language="java"><![CDATA[public class AnnotationService {
@Transformer(inputChannel = "aPollableChannel", outputChannel = "output")
public String handle(String payload) {
...
}
}]]></programlisting>
To use a named poller, use:
<programlisting language="java"><![CDATA[@Bean
public PollerMetadata myPoller() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(1000));
return pollerMetadata;
}]]></programlisting>
With this endpoint using the default poller:
<programlisting language="java"><![CDATA[public class AnnotationService {
@Transformer(inputChannel = "aPollableChannel", outputChannel = "output"
poller = @Poller("myPoller")
public String handle(String payload) {
...
}
}]]></programlisting>
</para>
<para>
Also see <xref linkend="advising-with-annotations"/>.
</para>

View File

@@ -142,6 +142,16 @@
For more information, see <xref linkend="redis-lock-registry"/> and <xref linkend="aggregator"/>.
</para>
</section>
<section id="4.0-poller-annotation">
<title>@Poller</title>
<para>
Annotation-based messaging configuration can now have a <code>poller</code> attribute.
This means that methods annotated with (<interfacename>@ServiceActivator</interfacename>,
<interfacename>@Aggregator</interfacename> etc.) can now use an <code>inputChannel</code> that
is a reference to a <interfacename>PollableChannel</interfacename>.
For more information, see <xref linkend="annotations"/>.
</para>
</section>
</section>
<section id="4.0-general">