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

INT-1371, made spring-tx dependency optional

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-09-15 10:26:07 -04:00
parent b3bba292db
commit 65a2f2462a
7 changed files with 143 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -156,8 +156,7 @@ public class ConsumerEndpointFactoryBean
pollingConsumer.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
pollingConsumer.setReceiveTimeout(this.pollerMetadata.getReceiveTimeout());
pollingConsumer.setTaskExecutor(this.pollerMetadata.getTaskExecutor());
pollingConsumer.setTransactionManager(this.pollerMetadata.getTransactionManager());
pollingConsumer.setTransactionDefinition(this.pollerMetadata.getTransactionDefinition());
pollingConsumer.setPollingDecorator(this.pollerMetadata.getPollingDecorator());
pollingConsumer.setAdviceChain(this.pollerMetadata.getAdviceChain());
this.endpoint = pollingConsumer;
}

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

@@ -35,6 +35,7 @@ import org.springframework.util.Assert;
* FactoryBean for creating a SourcePollingChannelAdapter instance.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<SourcePollingChannelAdapter>,
BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean, SmartLifecycle {
@@ -127,8 +128,6 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
spca.setTrigger(this.pollerMetadata.getTrigger());
spca.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
spca.setTaskExecutor(this.pollerMetadata.getTaskExecutor());
spca.setTransactionManager(this.pollerMetadata.getTransactionManager());
spca.setTransactionDefinition(this.pollerMetadata.getTransactionDefinition());
spca.setAdviceChain(this.pollerMetadata.getAdviceChain());
spca.setAutoStartup(this.autoStartup);
spca.setBeanName(this.beanName);

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

@@ -20,10 +20,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
@@ -33,18 +29,22 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.support.ManagedProperties;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
/**
* Parser for the &lt;poller&gt; element.
*
* @author Mark Fisher
* @author Marius Bogoevici
* @author Oleg Zhurakousky
*/
public class PollerParser extends AbstractBeanDefinitionParser {
@@ -202,15 +202,19 @@ public class PollerParser extends AbstractBeanDefinitionParser {
* and "transactionDefinition" properties for the target builder.
*/
private void configureTransactionAttributes(Element txElement, BeanDefinitionBuilder targetBuilder) {
targetBuilder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
DefaultTransactionDefinition txDefinition = new DefaultTransactionDefinition();
txDefinition.setPropagationBehaviorName(
DefaultTransactionDefinition.PREFIX_PROPAGATION + txElement.getAttribute("propagation"));
txDefinition.setIsolationLevelName(
DefaultTransactionDefinition.PREFIX_ISOLATION + txElement.getAttribute("isolation"));
txDefinition.setTimeout(Integer.valueOf(txElement.getAttribute("timeout")));
txDefinition.setReadOnly(txElement.getAttribute("read-only").equalsIgnoreCase("true"));
targetBuilder.addPropertyValue("transactionDefinition", txDefinition);
ManagedProperties transactionalProperties = new ManagedProperties();
transactionalProperties.setProperty("transactionManager", txElement.getAttribute("transaction-manager"));
transactionalProperties.setProperty("PROPAGATION", "PROPAGATION_" + txElement.getAttribute("propagation"));
transactionalProperties.setProperty("ISOLATION", "ISOLATION_" + txElement.getAttribute("isolation"));
transactionalProperties.setProperty("timeout", txElement.getAttribute("timeout"));
transactionalProperties.setProperty("readOnly", txElement.getAttribute("read-only"));
BeanDefinitionBuilder pollingDecoratorBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.endpoint.TransactionalCallbackDecorator");
pollingDecoratorBuilder.addPropertyValue("transactionalProperties", transactionalProperties);
targetBuilder.addPropertyValue("pollingDecorator", pollingDecoratorBuilder.getBeanDefinition());
}
/**

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

@@ -22,6 +22,7 @@ import java.util.concurrent.Executor;
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;
@@ -29,18 +30,13 @@ import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.scheduling.Trigger;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
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;
import org.springframework.util.ErrorHandler;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
public abstract class AbstractPollingEndpoint extends AbstractEndpoint implements InitializingBean, BeanClassLoaderAware {
@@ -54,12 +50,12 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
private volatile Executor taskExecutor;
private volatile ErrorHandler errorHandler;
private PollerCallbackDecorator pollingDecorator;
private volatile PlatformTransactionManager transactionManager;
private volatile TransactionDefinition transactionDefinition;
private volatile TransactionTemplate transactionTemplate;
public void setPollingDecorator(PollerCallbackDecorator pollingDecorator) {
this.pollingDecorator = pollingDecorator;
}
private final List<Advice> adviceChain = new CopyOnWriteArrayList<Advice>();
@@ -104,24 +100,11 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
this.errorHandler = errorHandler;
}
/**
* Specify a transaction manager to use for all polling operations.
* If none is provided, then the operations will occur without any
* transactional behavior (i.e. there is no default transaction manager).
*/
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
public void setTransactionDefinition(TransactionDefinition transactionDefinition) {
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) {
synchronized (this.adviceChain) {
this.adviceChain.clear();
@@ -131,13 +114,6 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
}
}
private TransactionTemplate getTransactionTemplate() {
if (!this.initialized) {
this.onInit();
}
return this.transactionTemplate;
}
@Override
protected void onInit() {
synchronized (this.initializationMonitor) {
@@ -145,13 +121,6 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
return;
}
Assert.notNull(this.trigger, "trigger is required");
if (this.transactionManager != null) {
if (this.transactionDefinition == null) {
this.transactionDefinition = new DefaultTransactionDefinition();
}
this.transactionTemplate = new TransactionTemplate(
this.transactionManager, this.transactionDefinition);
}
if (this.taskExecutor != null && !(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
if (this.errorHandler == null) {
this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(getBeanFactory()));
@@ -164,14 +133,25 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
}
private Runnable createPoller() {
if (this.adviceChain.isEmpty()) {
return new Poller();
Runnable poller = new Poller();
if (pollingDecorator != null){
poller = (Runnable) pollingDecorator.decorate(poller);
}
ProxyFactory proxyFactory = new ProxyFactory(new Poller());
for (Advice advice : this.adviceChain) {
proxyFactory.addAdvice(advice);
if (poller instanceof Advised){
Advised advised = (Advised) poller;
for (Advice advice : adviceChain) {
advised.addAdvice(advice);
}
} else {
if (adviceChain.size() > 0){
ProxyFactory proxyFactory = new ProxyFactory(poller);
for (Advice advice : adviceChain) {
proxyFactory.addAdvice(advice);
}
poller = (Runnable) proxyFactory.getProxy(this.classLoader);
}
}
return (Runnable) proxyFactory.getProxy(this.classLoader);
return poller;
}
@@ -225,16 +205,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
}
private boolean innerPoll() {
TransactionTemplate txTemplate = getTransactionTemplate();
if (txTemplate != null) {
return txTemplate.execute(new TransactionCallback<Boolean>() {
public Boolean doInTransaction(TransactionStatus status) {
return doPoll();
}
});
}
return doPoll();
}
}
}

23
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerCallbackDecorator.java Normal file
View File

@@ -0,0 +1,23 @@
/*
* 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.endpoint;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
public interface PollerCallbackDecorator {
Object decorate(Object poller);
}

65
spring-integration-core/src/main/java/org/springframework/integration/endpoint/TransactionalCallbackDecorator.java Normal file
View File

@@ -0,0 +1,65 @@
/*
* 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.endpoint;
import java.util.Properties;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
import org.springframework.transaction.interceptor.TransactionProxyFactoryBean;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
class TransactionalCallbackDecorator implements PollerCallbackDecorator, 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);
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"));
txDefinition.setTimeout(Integer.valueOf(transactionalProperties.getProperty("timeout")));
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();
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
}

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

@@ -20,12 +20,12 @@ import java.util.List;
import java.util.concurrent.Executor;
import org.aopalliance.aop.Advice;
import org.springframework.integration.endpoint.PollerCallbackDecorator;
import org.springframework.scheduling.Trigger;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
public class PollerMetadata {
@@ -38,11 +38,16 @@ public class PollerMetadata {
private List<Advice> adviceChain;
private volatile Executor taskExecutor;
private PollerCallbackDecorator pollingDecorator;
private volatile PlatformTransactionManager transactionManager;
private volatile TransactionDefinition transactionDefinition;
public PollerCallbackDecorator getPollingDecorator() {
return pollingDecorator;
}
public void setPollingDecorator(PollerCallbackDecorator pollingDecorator) {
this.pollingDecorator = pollingDecorator;
}
public void setTrigger(Trigger trigger) {
this.trigger = trigger;
@@ -83,21 +88,4 @@ public class PollerMetadata {
public Executor getTaskExecutor() {
return this.taskExecutor;
}
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
public PlatformTransactionManager getTransactionManager() {
return this.transactionManager;
}
public void setTransactionDefinition(TransactionDefinition transactionDefinition) {
this.transactionDefinition = transactionDefinition;
}
public TransactionDefinition getTransactionDefinition() {
return this.transactionDefinition;
}
}