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

INT-1466 decoupled Poller from Polling consumers (PollingConsumer and SourcePollingChannelAdapter)

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-09-20 00:39:06 -04:00
parent 1f6bff9ade
commit 35ef39589b
18 changed files with 467 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -13,9 +13,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
@@ -32,6 +32,7 @@ import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.scheduling.PollerFactory;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -42,7 +43,7 @@ import org.springframework.util.StringUtils;
* @author Josh Long
*/
public class ConsumerEndpointFactoryBean
implements FactoryBean<AbstractEndpoint>, BeanFactoryAware, BeanNameAware, InitializingBean, SmartLifecycle {
implements FactoryBean<AbstractEndpoint>, BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean, SmartLifecycle {
private volatile MessageHandler handler;
@@ -51,12 +52,14 @@ public class ConsumerEndpointFactoryBean
private volatile String inputChannelName;
private volatile PollerMetadata pollerMetadata;
private volatile boolean autoStartup = true;
private volatile MessageChannel inputChannel;
private volatile ConfigurableBeanFactory beanFactory;
private volatile ClassLoader beanClassLoader;
private volatile AbstractEndpoint endpoint;
@@ -86,6 +89,10 @@ public class ConsumerEndpointFactoryBean
public void setPollerMetadata(PollerMetadata pollerMetadata) {
this.pollerMetadata = pollerMetadata;
}
public void setBeanClassLoader(ClassLoader classLoader) {
this.beanClassLoader = classLoader;
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
@@ -153,11 +160,12 @@ public class ConsumerEndpointFactoryBean
+ "', and no default poller is available within the context.");
}
pollingConsumer.setTrigger(this.pollerMetadata.getTrigger());
pollingConsumer.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
pollingConsumer.setReceiveTimeout(this.pollerMetadata.getReceiveTimeout());
pollingConsumer.setTaskExecutor(this.pollerMetadata.getTaskExecutor());
pollingConsumer.setPollingDecorator(this.pollerMetadata.getPollingDecorator());
pollingConsumer.setAdviceChain(this.pollerMetadata.getAdviceChain());
PollerFactory pollerFactory = new PollerFactory(pollerMetadata);
pollerFactory.setBeanFactory(this.beanFactory);
pollerFactory.setBeanClassLoader(this.beanClassLoader);
pollingConsumer.setPollerFactory(pollerFactory);
this.endpoint = pollingConsumer;
}
else {
@@ -205,5 +213,4 @@ public class ConsumerEndpointFactoryBean
this.endpoint.stop(callback);
}
}
}

76
spring-integration-core/src/main/java/org/springframework/integration/config/Poller.java Normal file
View File

@@ -0,0 +1,76 @@
/*
* Copyright 2002-2010 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.config;
import java.util.concurrent.Callable;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.message.ErrorMessage;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
public class Poller implements Runnable {
public static final int MAX_MESSAGES_UNBOUNDED = -1;
private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
private Callable<Boolean> pollingTask;
/**
* @param pollingTask
*/
public Poller(Callable<Boolean> pollingTask){
this.pollingTask = pollingTask;
}
/* (non-Javadoc)
* @see java.lang.Runnable#run()
*/
public void run() {
int count = 0;
while (maxMessagesPerPoll <= 0 || count < maxMessagesPerPoll) {
try {
boolean computed = pollingTask.call();
if (!computed){
break;
}
count++;
} catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException)e;
} else {
throw new MessageHandlingException(new ErrorMessage(e));
}
}
}
}
/**
*
* @return
*/
public long getMaxMessagesPerPoll() {
return maxMessagesPerPoll;
}
/**
* Set the maximum number of messages to receive for each poll.
* A non-positive value indicates that polling should repeat as long
* as non-null messages are being received and successfully sent.
*
* <p>The default is unbounded.
*
* @see #MAX_MESSAGES_UNBOUNDED
*/
public void setMaxMessagesPerPoll(long maxMessagesPerPoll) {
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
}

11
spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
View File

@@ -28,6 +28,7 @@ import org.springframework.integration.MessageChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.scheduling.PollerFactory;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.util.Assert;
@@ -126,14 +127,13 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
+ this.beanName + "', and no default poller is available within the context.");
}
spca.setTrigger(this.pollerMetadata.getTrigger());
spca.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
spca.setTaskExecutor(this.pollerMetadata.getTaskExecutor());
spca.setAdviceChain(this.pollerMetadata.getAdviceChain());
spca.setPollingDecorator(pollerMetadata.getPollingDecorator());
PollerFactory pollerFactory = new PollerFactory(pollerMetadata);
pollerFactory.setBeanFactory(this.beanFactory);
pollerFactory.setBeanClassLoader(this.beanClassLoader);
spca.setPollerFactory(pollerFactory);
spca.setAutoStartup(this.autoStartup);
spca.setBeanName(this.beanName);
spca.setBeanFactory(this.beanFactory);
spca.setBeanClassLoader(this.beanClassLoader);
spca.afterPropertiesSet();
this.adapter = spca;
this.initialized = true;
@@ -173,5 +173,4 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
this.adapter.stop(callback);
}
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PollerParser.java
View File

@@ -95,6 +95,7 @@ public class PollerParser extends AbstractBeanDefinitionParser {
if (adviceChainElement != null) {
configureAdviceChain(adviceChainElement, metadataBuilder, parserContext);
}
Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
if (txElement != null) {
configureTransactionAttributes(txElement, metadataBuilder);
@@ -211,10 +212,9 @@ public class PollerParser extends AbstractBeanDefinitionParser {
transactionalProperties.setProperty("readOnly", txElement.getAttribute("read-only"));
BeanDefinitionBuilder pollingDecoratorBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.config.TransactionalCallbackDecorator");
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.scheduling.PollerTaskTransactionDecorator");
pollingDecoratorBuilder.addPropertyValue("transactionalProperties", transactionalProperties);
targetBuilder.addPropertyValue("pollingDecorator", pollingDecoratorBuilder.getBeanDefinition());
targetBuilder.addPropertyValue("transactionDecorator", pollingDecoratorBuilder.getBeanDefinition());
}
/**

154
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
View File

@@ -13,53 +13,26 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.endpoint;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledFuture;
import org.aopalliance.aop.Advice;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.integration.MessagingException;
import org.springframework.integration.config.Poller;
import org.springframework.integration.scheduling.PollerFactory;
import org.springframework.scheduling.Trigger;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ErrorHandler;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
public abstract class AbstractPollingEndpoint extends AbstractEndpoint implements InitializingBean, BeanClassLoaderAware {
public static final int MAX_MESSAGES_UNBOUNDED = -1;
public abstract class AbstractPollingEndpoint extends AbstractEndpoint implements InitializingBean{
private volatile Trigger trigger;
protected volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
private volatile Executor taskExecutor;
private volatile ErrorHandler errorHandler;
private PollerCallbackDecorator pollingDecorator;
public void setPollingDecorator(PollerCallbackDecorator pollingDecorator) {
this.pollingDecorator = pollingDecorator;
}
private final List<Advice> adviceChain = new CopyOnWriteArrayList<Advice>();
private volatile ClassLoader classLoader = ClassUtils.getDefaultClassLoader();
private PollerFactory pollerFactory;
private volatile ScheduledFuture<?> runningTask;
@@ -68,52 +41,24 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
/**
*
*/
public AbstractPollingEndpoint() {
this.setPhase(Integer.MAX_VALUE);
}
/**
* @param trigger
*/
public void setTrigger(Trigger trigger) {
this.trigger = trigger;
}
/**
* Set the maximum number of messages to receive for each poll.
* A non-positive value indicates that polling should repeat as long
* as non-null messages are being received and successfully sent.
*
* <p>The default is unbounded.
*
* @see #MAX_MESSAGES_UNBOUNDED
* @param pollerFactory
*/
public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
this.maxMessagesPerPoll = maxMessagesPerPoll;
public void setPollerFactory(PollerFactory pollerFactory) {
this.pollerFactory = pollerFactory;
}
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public void setErrorHandler(ErrorHandler errorHandler){
this.errorHandler = errorHandler;
}
public void setBeanClassLoader(ClassLoader classLoader) {
Assert.notNull(classLoader, "ClassLoader must not be null");
this.classLoader = classLoader;
}
//
public void setAdviceChain(List<Advice> adviceChain) {
synchronized (this.adviceChain) {
this.adviceChain.clear();
if (adviceChain != null) {
this.adviceChain.addAll(adviceChain);
}
}
}
@Override
protected void onInit() {
synchronized (this.initializationMonitor) {
@@ -121,40 +66,29 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
return;
}
Assert.notNull(this.trigger, "trigger is required");
if (this.taskExecutor != null && !(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
if (this.errorHandler == null) {
this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(getBeanFactory()));
}
this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, this.errorHandler);
try {
this.poller = this.createPoller();
this.initialized = true;
} catch (Exception e) {
throw new MessagingException("Problems creating a poller", e);
}
this.poller = this.createPoller();
this.initialized = true;
}
}
private Runnable createPoller() {
Runnable poller = new Poller();
if (pollingDecorator != null){
poller = (Runnable) pollingDecorator.decorate(poller);
}
if (poller instanceof Advised){
Advised advised = (Advised) poller;
for (Advice advice : adviceChain) {
advised.addAdvice(advice);
private Runnable createPoller() throws Exception{
Callable<Boolean> pollingTask = new Callable<Boolean>() {
public Boolean call() throws Exception {
return doPoll();
}
};
if (pollerFactory == null){
poller = new Poller(pollingTask);
} else {
if (adviceChain.size() > 0){
ProxyFactory proxyFactory = new ProxyFactory(poller);
for (Advice advice : adviceChain) {
proxyFactory.addAdvice(advice);
}
poller = (Runnable) proxyFactory.getProxy(this.classLoader);
}
poller = pollerFactory.createPoller(pollingTask);
}
return poller;
}
// LifecycleSupport implementation
@Override // guarded by super#lifecycleLock
@@ -175,37 +109,5 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
this.runningTask = null;
}
protected abstract boolean doPoll();
private class Poller implements Runnable {
public void run() {
if (taskExecutor != null) {
taskExecutor.execute(new Runnable() {
public void run() {
poll();
}
});
}
else {
poll();
}
}
private void poll() {
int count = 0;
while (maxMessagesPerPoll <= 0 || count < maxMessagesPerPoll) {
if (!innerPoll()) {
break;
}
count++;
}
}
private boolean innerPoll() {
return doPoll();
}
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -30,6 +30,7 @@ import org.springframework.util.Assert;
* {@link MessageSource} to a {@link MessageChannel}.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
public class SourcePollingChannelAdapter extends AbstractPollingEndpoint implements TrackableComponent {
@@ -81,11 +82,6 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
protected void onInit() {
Assert.notNull(this.source, "source must not be null");
Assert.notNull(this.outputChannel, "outputChannel must not be null");
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);
}
super.onInit();
}

114
spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerFactory.java Normal file
View File

@@ -0,0 +1,114 @@
/*
* Copyright 2002-2010 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.scheduling;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import org.aopalliance.aop.Advice;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.config.Poller;
import org.springframework.integration.util.AsyncInvokerAdvice;
import org.springframework.integration.util.ObjectDecorator;
import org.springframework.util.CollectionUtils;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
public class PollerFactory implements BeanClassLoaderAware, BeanFactoryAware {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile ClassLoader beanClassLoader;
private volatile BeanFactory beanFactory;
private volatile PollerMetadata pollerMetadata;
/**
*
*/
public PollerFactory(){}
/**
*
* @param pollerMetadata
*/
public PollerFactory(PollerMetadata pollerMetadata){
this.pollerMetadata = pollerMetadata;
}
/**
*
* @param pollingTask
* @return
* @throws Exception
*/
@SuppressWarnings("unchecked")
public Runnable createPoller(Callable<Boolean> pollingTask) throws Exception {
if (this.taskDecorationRequired()){
ProxyFactory proxyFactory = new ProxyFactory(pollingTask);
pollingTask = (Callable<Boolean>) proxyFactory.getProxy(this.beanClassLoader);
ObjectDecorator transactionDecorator = this.pollerMetadata.getTransactionDecorator();
// take care of TransactionINterceptor first
if (transactionDecorator != null){
pollingTask = (Callable<Boolean>) transactionDecorator.decorate(pollingTask);
logger.info("Polling task has been decorated with TransactionInterceptor to handle transactions");
}
// ... then add more Advises if provided
List<Advice> advices = this.pollerMetadata.getAdviceChain();
if (advices != null){
for (Advice advice : advices) {
((Advised)pollingTask).addAdvice(advice);
logger.info("Polling task has been decorated with " + advice.getClass().getSimpleName());
}
}
}
Runnable poller = new Poller(pollingTask);
if (pollerMetadata != null){
((Poller)poller).setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
}
// Decorate Poller with AsyncInvokerAdvice
Executor taskExecutor = this.pollerMetadata.getTaskExecutor();
if (taskExecutor != null){
ProxyFactory proxyFactory = new ProxyFactory(poller);
AsyncInvokerAdvice asyncInvokerAdvice = new AsyncInvokerAdvice(taskExecutor);
asyncInvokerAdvice.setBeanFactory(this.beanFactory);
asyncInvokerAdvice.afterPropertiesSet();
proxyFactory.addAdvice(asyncInvokerAdvice);
poller = (Runnable) proxyFactory.getProxy(this.beanClassLoader);
logger.info("Poller has been decorated with AsyncInvokerAdvice for async polling");
}
return poller;
}
public void setBeanClassLoader(ClassLoader beanClassLoader) {
this.beanClassLoader = beanClassLoader;
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
public void setPollerMetadata(PollerMetadata pollerMetadata) {
this.pollerMetadata = pollerMetadata;
}
private boolean taskDecorationRequired(){
return pollerMetadata != null &&
( this.pollerMetadata.getTransactionDecorator() != null ||
!CollectionUtils.isEmpty(this.pollerMetadata.getAdviceChain()) );
}
}

12
spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java
View File

@@ -20,7 +20,7 @@ import java.util.List;
import java.util.concurrent.Executor;
import org.aopalliance.aop.Advice;
import org.springframework.integration.endpoint.PollerCallbackDecorator;
import org.springframework.integration.util.ObjectDecorator;
import org.springframework.scheduling.Trigger;
/**
@@ -39,14 +39,14 @@ public class PollerMetadata {
private volatile Executor taskExecutor;
private PollerCallbackDecorator pollingDecorator;
private volatile ObjectDecorator transactionDecorator;
public PollerCallbackDecorator getPollingDecorator() {
return pollingDecorator;
public ObjectDecorator getTransactionDecorator() {
return transactionDecorator;
}
public void setPollingDecorator(PollerCallbackDecorator pollingDecorator) {
this.pollingDecorator = pollingDecorator;
public void setTransactionDecorator(ObjectDecorator transactionDecorator) {
this.transactionDecorator = transactionDecorator;
}
public void setTrigger(Trigger trigger) {

57
spring-integration-core/src/main/java/org/springframework/integration/config/TransactionalCallbackDecorator.java → spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerTaskTransactionDecorator.java
View File

@@ -13,40 +13,41 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
package org.springframework.integration.scheduling;
import java.util.Properties;
import org.springframework.aop.framework.Advised;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.endpoint.PollerCallbackDecorator;
import org.springframework.integration.config.Poller;
import org.springframework.integration.util.ObjectDecorator;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
import org.springframework.transaction.interceptor.TransactionProxyFactoryBean;
import org.springframework.transaction.interceptor.TransactionAttributeSourceAdvisor;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.util.Assert;
/**
* A simple implementation of {@link ObjectDecorator} which will add
* {@link TransactionInterceptor} advice to any instance of {@link Advised}.
* Currently used to decorate {@link Poller}'s <code>pollingTask</code>.
*
* @author Oleg Zhurakousky
* @since 2.0
*/
class TransactionalCallbackDecorator implements PollerCallbackDecorator, BeanFactoryAware {
public class PollerTaskTransactionDecorator implements ObjectDecorator, BeanFactoryAware {
private BeanFactory beanFactory;
private Properties transactionalProperties;
public Properties getTransactionalProperties() {
return transactionalProperties;
}
public void setTransactionalProperties(Properties transactionalProperties) {
this.transactionalProperties = transactionalProperties;
}
public Object decorate(Object pollingCallback){
TransactionProxyFactoryBean txFactoryBean = new TransactionProxyFactoryBean();
txFactoryBean.setBeanFactory(beanFactory);
/* (non-Javadoc)
* @see org.springframework.integration.util.ObjectDecorator#decorate(java.lang.Object)
*/
public Object decorate(Object advisedPollingTask) {
Assert.isInstanceOf(Advised.class, advisedPollingTask, "'pollingTask' must be an instance of Advised");
PlatformTransactionManager txManager = (PlatformTransactionManager) this.beanFactory.getBean(transactionalProperties.getProperty("transactionManager"));
txFactoryBean.setTransactionManager(txManager);
DefaultTransactionAttribute txDefinition = new DefaultTransactionAttribute();
txDefinition.setPropagationBehaviorName(transactionalProperties.getProperty("PROPAGATION"));
txDefinition.setIsolationLevelName(transactionalProperties.getProperty("ISOLATION"));
@@ -54,13 +55,25 @@ class TransactionalCallbackDecorator implements PollerCallbackDecorator, BeanFac
txDefinition.setReadOnly(transactionalProperties.getProperty("readOnly").equalsIgnoreCase("true"));
MatchAlwaysTransactionAttributeSource attributeSource = new MatchAlwaysTransactionAttributeSource();
attributeSource.setTransactionAttribute(txDefinition);
txFactoryBean.setTransactionAttributeSource(attributeSource);
txFactoryBean.setTarget(pollingCallback);
txFactoryBean.afterPropertiesSet();
return txFactoryBean.getObject();
TransactionInterceptor transactionInterceptor = new TransactionInterceptor();
transactionInterceptor.setTransactionManager(txManager);
transactionInterceptor.setTransactionAttributeSource(attributeSource);
transactionInterceptor.afterPropertiesSet();
((Advised)advisedPollingTask).addAdvisor(new TransactionAttributeSourceAdvisor(transactionInterceptor));
return advisedPollingTask;
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
public Properties getTransactionalProperties() {
return transactionalProperties;
}
public void setTransactionalProperties(Properties transactionalProperties) {
this.transactionalProperties = transactionalProperties;
}
}

88
spring-integration-core/src/main/java/org/springframework/integration/util/AsyncInvokerAdvice.java Normal file
View File

@@ -0,0 +1,88 @@
/*
* Copyright 2002-2010 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.util;
import java.util.concurrent.Executor;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.util.ErrorHandler;
/**
* Simple advise to support async execution of tasks.
* It will simply delegate <code>invocation.proceed()</code> calls to its {@link TaskExecutor}
*
* @author Oleg Zhurakousky
* @since 2.0
*/
public class AsyncInvokerAdvice implements MethodInterceptor, InitializingBean,BeanFactoryAware {
private Executor taskExecutor;
private volatile ErrorHandler errorHandler;
private BeanFactory beanFactory;
/**
* @param taskExecutor
*/
public AsyncInvokerAdvice(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/*
* (non-Javadoc)
* @see org.aopalliance.intercept.MethodInterceptor#invoke(org.aopalliance.intercept.MethodInvocation)
*/
public Object invoke(final MethodInvocation invocation) throws Throwable {
taskExecutor.execute(new Runnable() {
public void run() {
try {
invocation.proceed();
} catch (Throwable e) {
if (e instanceof RuntimeException){
throw (RuntimeException)e;
} else {
throw new MessagingException("Problems during asynchronous invocation of task: " + this, e);
}
}
}
});
return null;
}
/*
* (non-Javadoc)
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
if (this.errorHandler == null) {
this.errorHandler = new MessagePublishingErrorHandler(
new BeanFactoryChannelResolver(this.beanFactory));
}
this.taskExecutor = new ErrorHandlingTaskExecutor(taskExecutor, errorHandler);
}
}
/*
* (non-Javadoc)
* @see org.springframework.beans.factory.BeanFactoryAware#setBeanFactory(org.springframework.beans.factory.BeanFactory)
*/
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
}

13
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerCallbackDecorator.java → spring-integration-core/src/main/java/org/springframework/integration/util/ObjectDecorator.java
View File

@@ -13,11 +13,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.endpoint;
package org.springframework.integration.util;
/**
* Base decorator interface defining common behavior for basic decoration.
*
* @author Oleg Zhurakousky
* @since 2.0
*/
public interface PollerCallbackDecorator {
Object decorate(Object poller);
public interface ObjectDecorator {
/**
* @param object
* @return
*/
Object decorate(Object object);
}

4
spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java
View File

@@ -31,11 +31,11 @@ import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.PollerCallbackDecorator;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.test.util.TestUtils.TestApplicationContext;
import org.springframework.integration.util.ObjectDecorator;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.ClassUtils;
@@ -96,7 +96,7 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
pollerMetadata.setTrigger(new PeriodicTrigger(5000));
pollerMetadata.setMaxMessagesPerPoll(1);
final AtomicInteger count = new AtomicInteger();
pollerMetadata.setPollingDecorator(new PollerCallbackDecorator() {
pollerMetadata.setTransactionDecorator(new ObjectDecorator() {
public Object decorate(Object poller) {
count.incrementAndGet();
return poller;

17
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java
View File

@@ -22,6 +22,7 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.List;
import java.util.concurrent.Callable;
import org.aopalliance.aop.Advice;
import org.aopalliance.intercept.MethodInterceptor;
@@ -35,6 +36,8 @@ import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerFactory;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.TestTransactionManager;
import org.springframework.transaction.IllegalTransactionStateException;
@@ -65,18 +68,22 @@ public class PollingTransactionTests {
}
@Test
@SuppressWarnings("unchecked")
public void transactionWithCommitAndAdvices() throws InterruptedException {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"transactionTests.xml", this.getClass());
PollingConsumer advicedPoller = context.getBean("advicedSa", PollingConsumer.class);
@SuppressWarnings("unchecked")
List<Advice> adviceChain = TestUtils.getPropertyValue(advicedPoller, "adviceChain",List.class);
PollerFactory pollerFactory = TestUtils.getPropertyValue(advicedPoller, "pollerFactory",PollerFactory.class);
PollerMetadata pollerMetedata = TestUtils.getPropertyValue(pollerFactory, "pollerMetadata",PollerMetadata.class);
List<Advice> adviceChain = TestUtils.getPropertyValue(pollerMetedata, "adviceChain",List.class);
assertEquals(2, adviceChain.size());
Runnable poller = TestUtils.getPropertyValue(advicedPoller, "poller", Runnable.class);
assertTrue("Poller is not Advised", poller instanceof Advised);
Advisor[] advisors = ((Advised)poller).getAdvisors();
Callable<?> pollingTask = TestUtils.getPropertyValue(poller, "pollingTask", Callable.class);
assertTrue("Poller is not Advised", pollingTask instanceof Advised);
Advisor[] advisors = ((Advised)pollingTask).getAdvisors();
assertEquals(3, advisors.length);
// System.err.println(Arrays.asList(advisors));
assertTrue("First advisor is not TX", advisors[0] instanceof TransactionAttributeSourceAdvisor);
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
MessageChannel input = (MessageChannel) context.getBean("goodInputWithAdvice");

22
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java
View File

@@ -39,6 +39,8 @@ import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerFactory;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -93,7 +95,9 @@ public class PollingConsumerEndpointTests {
expect(channelMock.receive()).andReturn(message);
expectLastCall();
replay(channelMock);
endpoint.setMaxMessagesPerPoll(1);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.start();
trigger.await();
endpoint.stop();
@@ -105,7 +109,9 @@ public class PollingConsumerEndpointTests {
public void multipleMessages() {
expect(channelMock.receive()).andReturn(message).times(5);
replay(channelMock);
endpoint.setMaxMessagesPerPoll(5);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(5);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.start();
trigger.await();
endpoint.stop();
@@ -118,7 +124,9 @@ public class PollingConsumerEndpointTests {
expect(channelMock.receive()).andReturn(message).times(5);
expect(channelMock.receive()).andReturn(null);
replay(channelMock);
endpoint.setMaxMessagesPerPoll(6);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(6);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.start();
trigger.await();
endpoint.stop();
@@ -151,7 +159,9 @@ public class PollingConsumerEndpointTests {
public void droppedMessage_onePerPoll() throws Throwable {
expect(channelMock.receive()).andReturn(badMessage).times(1);
replay(channelMock);
endpoint.setMaxMessagesPerPoll(10);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(10);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.start();
trigger.await();
endpoint.stop();
@@ -179,7 +189,9 @@ public class PollingConsumerEndpointTests {
expectLastCall();
replay(channelMock);
endpoint.setReceiveTimeout(1);
endpoint.setMaxMessagesPerPoll(1);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.start();
trigger.await();
endpoint.stop();

10
spring-integration-core/src/test/java/org/springframework/integration/endpoint/pollingEndpointErrorHandlingTests.xml
View File

@@ -12,7 +12,15 @@
</si:channel>
<bean id="testEndpoint" class="org.springframework.integration.endpoint.PollingEndpointStub">
<property name="taskExecutor" ref="taskExecutor"/>
<property name="pollerFactory">
<bean class="org.springframework.integration.scheduling.PollerFactory">
<constructor-arg>
<bean class="org.springframework.integration.scheduling.PollerMetadata">
<property name="taskExecutor" ref="taskExecutor"/>
</bean>
</constructor-arg>
</bean>
</property>
</bean>
<task:executor id="taskExecutor" pool-size="1" rejection-policy="CALLER_RUNS"/>

7
spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java
View File

@@ -141,13 +141,6 @@ public class GatewayInvokingMessageHandlerTests {
@Test
public void validateGatewayWithErrorAsyncAndMaper() {
try {
gatewayWithErrorAsync.sendRecieve("echoWithErrorAsyncChannel");
Assert.fail();
} catch (Exception e) {
Assert.assertTrue(e instanceof MessageHandlingException);
}
try {
Object result = gatewayWithErrorAsyncAndMapper.sendRecieve("echoWithErrorAsyncChannel");
Assert.assertEquals("Error happened in message: echoWithErrorAsyncChannel", result);

30
spring-integration-stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java
View File

@@ -32,6 +32,8 @@ import org.junit.Test;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerFactory;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -93,7 +95,9 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void maxMessagesPerTaskSameAsMessageCount() {
endpoint.setMaxMessagesPerPoll(3);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(3);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
@@ -108,7 +112,9 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void maxMessagesPerTaskLessThanMessageCount() {
endpoint.setMaxMessagesPerPoll(2);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
@@ -122,7 +128,9 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void maxMessagesPerTaskExceedsMessageCount() {
endpoint.setMaxMessagesPerPoll(5);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(5);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
@@ -137,7 +145,9 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() {
endpoint.setMaxMessagesPerPoll(2);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
@@ -160,7 +170,9 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() {
endpoint.setMaxMessagesPerPoll(5);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(5);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
@@ -182,7 +194,9 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void testStreamResetBetweenDispatches() {
endpoint.setMaxMessagesPerPoll(2);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
@@ -204,7 +218,9 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void testStreamWriteBetweenDispatches() throws IOException {
endpoint.setMaxMessagesPerPoll(2);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);

31
spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java
View File

@@ -31,6 +31,8 @@ import org.junit.Test;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerFactory;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -81,7 +83,9 @@ public class CharacterStreamWritingMessageHandlerTests {
@Test
public void twoStringsAndNoNewLinesByDefault() {
endpoint.setMaxMessagesPerPoll(1);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
channel.send(new GenericMessage<String>("foo"), 0);
channel.send(new GenericMessage<String>("bar"), 0);
endpoint.start();
@@ -98,8 +102,9 @@ public class CharacterStreamWritingMessageHandlerTests {
@Test
public void twoStringsWithNewLines() {
handler.setShouldAppendNewLine(true);
endpoint.setMaxMessagesPerPoll(1);
channel.send(new GenericMessage<String>("foo"), 0);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); channel.send(new GenericMessage<String>("foo"), 0);
channel.send(new GenericMessage<String>("bar"), 0);
endpoint.start();
trigger.await();
@@ -115,7 +120,9 @@ public class CharacterStreamWritingMessageHandlerTests {
@Test
public void maxMessagesPerTaskSameAsMessageCount() {
endpoint.setMaxMessagesPerPoll(2);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
channel.send(new GenericMessage<String>("foo"), 0);
channel.send(new GenericMessage<String>("bar"), 0);
endpoint.start();
@@ -126,7 +133,9 @@ public class CharacterStreamWritingMessageHandlerTests {
@Test
public void maxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() {
endpoint.setMaxMessagesPerPoll(10);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(10);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setReceiveTimeout(0);
handler.setShouldAppendNewLine(true);
channel.send(new GenericMessage<String>("foo"), 0);
@@ -140,7 +149,9 @@ public class CharacterStreamWritingMessageHandlerTests {
@Test
public void singleNonStringObject() {
endpoint.setMaxMessagesPerPoll(1);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
TestObject testObject = new TestObject("foo");
channel.send(new GenericMessage<TestObject>(testObject));
endpoint.start();
@@ -152,7 +163,9 @@ public class CharacterStreamWritingMessageHandlerTests {
@Test
public void twoNonStringObjectWithOutNewLines() {
endpoint.setReceiveTimeout(0);
endpoint.setMaxMessagesPerPoll(2);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
TestObject testObject1 = new TestObject("foo");
TestObject testObject2 = new TestObject("bar");
channel.send(new GenericMessage<TestObject>(testObject1), 0);
@@ -167,7 +180,9 @@ public class CharacterStreamWritingMessageHandlerTests {
public void twoNonStringObjectWithNewLines() {
handler.setShouldAppendNewLine(true);
endpoint.setReceiveTimeout(0);
endpoint.setMaxMessagesPerPoll(2);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
TestObject testObject1 = new TestObject("foo");
TestObject testObject2 = new TestObject("bar");
channel.send(new GenericMessage<TestObject>(testObject1), 0);