From 65a2f2462ae09b09e4dfbca4d7c6dd6e5633cfcc Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 15 Sep 2010 10:26:07 -0400 Subject: [PATCH] INT-1371, made spring-tx dependency optional --- .../config/ConsumerEndpointFactoryBean.java | 3 +- ...ourcePollingChannelAdapterFactoryBean.java | 3 +- .../integration/config/xml/PollerParser.java | 32 ++++---- .../endpoint/AbstractPollingEndpoint.java | 79 ++++++------------- .../endpoint/PollerCallbackDecorator.java | 23 ++++++ .../TransactionalCallbackDecorator.java | 65 +++++++++++++++ .../scheduling/PollerMetadata.java | 32 +++----- 7 files changed, 143 insertions(+), 94 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerCallbackDecorator.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/TransactionalCallbackDecorator.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index e86bfd9aea..d7576da57a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java index b08bef7458..fcf1fa38a3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java @@ -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, BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean, SmartLifecycle { @@ -127,8 +128,6 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean adviceChain = new CopyOnWriteArrayList(); @@ -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 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() { - public Boolean doInTransaction(TransactionStatus status) { - return doPoll(); - } - }); - } return doPoll(); } } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerCallbackDecorator.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerCallbackDecorator.java new file mode 100644 index 0000000000..c274797852 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerCallbackDecorator.java @@ -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); +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TransactionalCallbackDecorator.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TransactionalCallbackDecorator.java new file mode 100644 index 0000000000..dc0f66c8e8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TransactionalCallbackDecorator.java @@ -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; + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java index 643ce1f8b1..514af3f62e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java @@ -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 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; - } - }