Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added an 'adviceChain' property to AbstractPollingEndpoint so that any AOP Advice instance(s) may be applied to the Poller (a Runnable within the AbstractPollingEndpoint). Also, added support for configuring this 'adviceChain' via the @Poller annotation (INT-407).

Browse Source
This commit is contained in:
Mark Fisher
2008-10-09 17:40:17 +00:00
parent b6d22a7644
commit 875b0cc02a
10 changed files with 82 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java
View File

@@ -53,4 +53,6 @@ public @interface Poller {
String transactionManager() default "";
String[] adviceChain() default {};
}

34
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -19,9 +19,13 @@ package org.springframework.integration.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import org.aopalliance.aop.Advice;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.generic.GenericBeanFactoryAccessor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.channel.ChannelRegistry;
@@ -58,16 +62,16 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
private static final String OUTPUT_CHANNEL_ATTRIBUTE = "outputChannel";
private final BeanFactory beanFactory;
private final GenericBeanFactoryAccessor beanFactoryAccessor;
protected final ChannelRegistry channelRegistry;
public AbstractMethodAnnotationPostProcessor(BeanFactory beanFactory) {
public AbstractMethodAnnotationPostProcessor(ListableBeanFactory beanFactory) {
Assert.notNull(beanFactory, "BeanFactory must not be null");
this.beanFactory = beanFactory;
this.channelRegistry = (ChannelRegistry) this.beanFactory.getBean(
MessageBusParser.MESSAGE_BUS_BEAN_NAME);
this.beanFactoryAccessor = new GenericBeanFactoryAccessor(beanFactory);
this.channelRegistry = this.beanFactoryAccessor.getBean(
MessageBusParser.MESSAGE_BUS_BEAN_NAME, ChannelRegistry.class);
}
@@ -116,13 +120,25 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
pollingEndpoint.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
if (StringUtils.hasText(pollerAnnotation.transactionManager())) {
String txManagerRef = pollerAnnotation.transactionManager();
Assert.isTrue(this.beanFactory.containsBean(txManagerRef), "no such bean '" + txManagerRef + "'");
PlatformTransactionManager txManager = (PlatformTransactionManager)
this.beanFactory.getBean(txManagerRef, PlatformTransactionManager.class);
Assert.isTrue(this.beanFactoryAccessor.getBeanFactory().containsBean(txManagerRef),
"failed to resolve transactionManager reference, no such bean '" + txManagerRef + "'");
PlatformTransactionManager txManager =
this.beanFactoryAccessor.getBean(txManagerRef, PlatformTransactionManager.class);
pollingEndpoint.setTransactionManager(txManager);
Transactional txAnnotation = pollerAnnotation.transactionAttributes();
pollingEndpoint.setTransactionDefinition(this.parseTransactionAnnotation(txAnnotation));
}
String[] adviceChainArray = pollerAnnotation.adviceChain();
if (adviceChainArray.length > 0) {
List<Advice> adviceChain = new ArrayList<Advice>();
for (String adviceChainString : adviceChainArray) {
String[] adviceRefs = StringUtils.tokenizeToStringArray(adviceChainString, ",");
for (String adviceRef : adviceRefs) {
adviceChain.add(this.beanFactoryAccessor.getBean(adviceRef, Advice.class));
}
}
pollingEndpoint.setAdviceChain(adviceChain);
}
}
endpoint = pollingEndpoint;
}

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -19,7 +19,7 @@ package org.springframework.integration.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.aggregator.AbstractMessageAggregator;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
@@ -39,7 +39,7 @@ import org.springframework.util.StringUtils;
*/
public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Aggregator> {
public AggregatorAnnotationPostProcessor(BeanFactory beanFactory) {
public AggregatorAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
}

10
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java
View File

@@ -31,7 +31,7 @@ import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.ChannelAdapter;
@@ -60,7 +60,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
private volatile MessageBus messageBus;
private volatile ConfigurableBeanFactory beanFactory;
private volatile ConfigurableListableBeanFactory beanFactory;
private final Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> postProcessors =
@@ -68,9 +68,9 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
public void setBeanFactory(BeanFactory beanFactory) {
Assert.isAssignable(ConfigurableBeanFactory.class, beanFactory.getClass(),
"a ConfigurableBeanFactory is required");
this.beanFactory = (ConfigurableBeanFactory) beanFactory;
Assert.isAssignable(ConfigurableListableBeanFactory.class, beanFactory.getClass(),
"a ConfigurableListableBeanFactory is required");
this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
}
public void afterPropertiesSet() {

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.config.annotation;
import java.lang.reflect.Method;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MessageConsumer;
@@ -34,7 +34,7 @@ import org.springframework.util.StringUtils;
*/
public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Router> {
public RouterAnnotationPostProcessor(BeanFactory beanFactory) {
public RouterAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
}

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.config.annotation;
import java.lang.reflect.Method;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
import org.springframework.integration.message.MessageConsumer;
@@ -31,7 +31,7 @@ import org.springframework.integration.message.MessageMappingMethodInvoker;
*/
public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<ServiceActivator> {
public ServiceActivatorAnnotationPostProcessor(BeanFactory beanFactory) {
public ServiceActivatorAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
}

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.config.annotation;
import java.lang.reflect.Method;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.splitter.MethodInvokingSplitter;
@@ -30,7 +30,7 @@ import org.springframework.integration.splitter.MethodInvokingSplitter;
*/
public class SplitterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Splitter> {
public SplitterAnnotationPostProcessor(BeanFactory beanFactory) {
public SplitterAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
}

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.config.annotation;
import java.lang.reflect.Method;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.transformer.MethodInvokingTransformer;
@@ -31,7 +31,7 @@ import org.springframework.integration.transformer.TransformerEndpoint;
*/
public class TransformerAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Transformer> {
public TransformerAnnotationPostProcessor(BeanFactory beanFactory) {
public TransformerAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
}

40
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
View File

@@ -16,8 +16,14 @@
package org.springframework.integration.endpoint;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledFuture;
import org.aopalliance.aop.Advice;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.task.TaskExecutor;
@@ -32,11 +38,13 @@ import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
* @author Mark Fisher
*/
public abstract class AbstractPollingEndpoint implements MessageEndpoint, TaskSchedulerAware, Lifecycle, InitializingBean {
public abstract class AbstractPollingEndpoint implements MessageEndpoint,
TaskSchedulerAware, Lifecycle, InitializingBean, BeanClassLoaderAware {
public static final int MAX_MESSAGES_UNBOUNDED = -1;
@@ -53,10 +61,16 @@ public abstract class AbstractPollingEndpoint implements MessageEndpoint, TaskSc
private volatile TransactionTemplate transactionTemplate;
private final List<Advice> adviceChain = new CopyOnWriteArrayList<Advice>();
private volatile ClassLoader classLoader = ClassUtils.getDefaultClassLoader();
private volatile TaskScheduler taskScheduler;
private volatile ScheduledFuture<?> runningTask;
private volatile Runnable poller;
private volatile boolean initialized;
private final Object lifecycleMonitor = new Object();
@@ -100,6 +114,16 @@ public abstract class AbstractPollingEndpoint implements MessageEndpoint, TaskSc
this.transactionDefinition = transactionDefinition;
}
public void setBeanClassLoader(ClassLoader classLoader) {
Assert.notNull(classLoader, "ClassLoader must not be null");
this.classLoader = classLoader;
}
public void setAdviceChain(List<Advice> adviceChain) {
this.adviceChain.clear();
this.adviceChain.addAll(adviceChain);
}
private TransactionTemplate getTransactionTemplate() {
if (!this.initialized) {
this.afterPropertiesSet();
@@ -122,10 +146,22 @@ public abstract class AbstractPollingEndpoint implements MessageEndpoint, TaskSc
this.transactionTemplate = new TransactionTemplate(
this.transactionManager, this.transactionDefinition);
}
this.poller = this.createPoller();
this.initialized = true;
}
}
private Runnable createPoller() {
if (this.adviceChain.isEmpty()) {
return new Poller();
}
ProxyFactory proxyFactory = new ProxyFactory(new Poller());
for (Advice advice : this.adviceChain) {
proxyFactory.addAdvice(advice);
}
return (Runnable) proxyFactory.getProxy(this.classLoader);
}
// Lifecycle implementation
@@ -145,7 +181,7 @@ public abstract class AbstractPollingEndpoint implements MessageEndpoint, TaskSc
}
Assert.state(this.taskScheduler != null,
"unable to start polling, no taskScheduler available");
this.runningTask = this.taskScheduler.schedule(new Poller(), this.trigger);
this.runningTask = this.taskScheduler.schedule(this.poller, this.trigger);
}
}

5
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -21,7 +21,6 @@ import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.util.Assert;
/**
@@ -62,8 +61,8 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
Assert.notNull(this.source, "source must not be null");
Assert.notNull(this.outputChannel, "outputChannel must not be null");
super.afterPropertiesSet();
if (this.maxMessagesPerPoll < 0 && source instanceof MethodInvokingSource) {
// the default is 1 since a MethodInvokingSource might return
if (this.maxMessagesPerPoll < 0) {
// the default is 1 since a source might return
// a non-null value every time it is invoked
this.setMaxMessagesPerPoll(1);
}