diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index 2160cd6936..b9a7341cb1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -32,12 +32,9 @@ import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; -import org.springframework.integration.transaction.IntegrationResourceHolder; -import org.springframework.integration.transaction.TransactionSynchronizationFactory; import org.springframework.integration.util.ErrorHandlingTaskExecutor; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.support.PeriodicTrigger; -import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; @@ -70,8 +67,6 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private final Object initializationMonitor = new Object(); - private volatile TransactionSynchronizationFactory transactionSynchronizationFactory; - public AbstractPollingEndpoint() { this.setPhase(Integer.MAX_VALUE); } @@ -112,11 +107,6 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.beanClassLoader = classLoader; } - public void setTransactionSynchronizationFactory( - TransactionSynchronizationFactory transactionSynchronizationFactory) { - this.transactionSynchronizationFactory = transactionSynchronizationFactory; - } - @Override protected void onInit() { synchronized (this.initializationMonitor) { @@ -170,20 +160,6 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement return new Poller(pollingTask); } - /** - * Synchronize with an existing transaction (if any) and receive - * a message using {@link #doReceive()}. - * @return The message (or null). - */ - protected final Message syncIfTxAndReceive() { - IntegrationResourceHolder holder = bindResourceHolderIfNecessary( - this.getResourceKey(), this.getResourceToBind()); - Message message = this.doReceive(); - if (holder != null && message != null) { - holder.setMessage(message); - } - return message; - } // LifecycleSupport implementation @Override // guarded by super#lifecycleLock @@ -205,29 +181,39 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.initialized = false; } - private IntegrationResourceHolder bindResourceHolderIfNecessary(String key, Object resource) { - IntegrationResourceHolder holder = null; - - if (this.transactionSynchronizationFactory != null) { - if (TransactionSynchronizationManager.isActualTransactionActive()) { - holder = new IntegrationResourceHolder(); - if (key != null) { - holder.addAttribute(key, resource); - } - TransactionSynchronizationManager.bindResource(resource, holder); - TransactionSynchronizationManager.registerSynchronization(this.transactionSynchronizationFactory.create(resource)); - } + /** + * @deprecated Starting with Spring Integration 3.0, subclasses will not be able to + * override this method. Use {@link #receiveMessage()} and {@link #handleMessage(Message)} instead, + * to separate the concerns of retrieving and processing a message. + * Consider refactoring now rather than waiting for 3.0. + * @return true if a message was processed. + */ + @Deprecated + protected boolean doPoll() { + Message message = this.receiveMessage(); + if (message != null) { + this.handleMessage(message); + return true; } - return holder; + return false; } - protected abstract boolean doPoll(); + /** + * Obtain the next message (if one is available). MAY return null + * if no message is immediately available. + * @return The message or null. + */ + protected Message receiveMessage() { + throw new UnsupportedOperationException("Subclass must implement receiveMessage()"); + } - protected abstract Message doReceive(); - - protected abstract Object getResourceToBind(); - - protected abstract String getResourceKey(); + /** + * Handle a message. + * @param message The message. + */ + protected void handleMessage(Message message) { + throw new UnsupportedOperationException("Subclass must implement handleMessage()"); + } /** * Default Poller implementation diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractTransactionSynchronizingPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractTransactionSynchronizingPollingEndpoint.java new file mode 100644 index 0000000000..26da36a7cf --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractTransactionSynchronizingPollingEndpoint.java @@ -0,0 +1,117 @@ +/* + * Copyright 2002-2012 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 org.springframework.integration.Message; +import org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor; +import org.springframework.integration.transaction.IntegrationResourceHolder; +import org.springframework.integration.transaction.TransactionSynchronizationFactory; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +/** + * Subclasses support pollers with transaction synchronization. + *

+ * This class will be removed in version 3.0.0 when its methods will be pulled up + * into {@link AbstractPollingEndpoint}. + * @author Gary Russell + * @since 2.2 + * + */ +abstract class AbstractTransactionSynchronizingPollingEndpoint extends AbstractPollingEndpoint { + + private volatile TransactionSynchronizationFactory transactionSynchronizationFactory; + + public void setTransactionSynchronizationFactory( + TransactionSynchronizationFactory transactionSynchronizationFactory) { + this.transactionSynchronizationFactory = transactionSynchronizationFactory; + } + + /** + * Return a resource (MessageSource etc) to bind when using transaction + * synchronization. + * @return The resource, or null if transaction synchronization is not required. + */ + protected Object getResourceToBind() { + return null; + } + + /** + * Return the key under which the resource will be made available as an + * attribute on the {@link IntegrationResourceHolder}. The default + * {@link ExpressionEvaluatingTransactionSynchronizationProcessor} + * makes this attribute available as a variable in SpEL expressions. + * @return The key, or null (default) if the resource shouldn't be + * made available as a attribute. + */ + protected String getResourceKey() { + return null; + } + + @Override + protected final boolean doPoll() { + IntegrationResourceHolder holder = bindResourceHolderIfNecessary( + this.getResourceKey(), this.getResourceToBind()); + Message message = this.receiveMessage(); + boolean result; + if (message == null) { + if (this.logger.isDebugEnabled()){ + this.logger.debug("Received no Message during the poll, returning 'false'"); + } + result = false; + } + else { + if (this.logger.isDebugEnabled()){ + this.logger.debug("Poll resulted in Message: " + message); + } + if (holder != null) { + holder.setMessage(message); + } + this.handleMessage(message); + result = true; + } + return result; + } + + private IntegrationResourceHolder bindResourceHolderIfNecessary(String key, Object resource) { + IntegrationResourceHolder holder = null; + + if (this.transactionSynchronizationFactory != null && resource != null) { + if (TransactionSynchronizationManager.isActualTransactionActive()) { + holder = new IntegrationResourceHolder(); + if (key != null) { + holder.addAttribute(key, resource); + } + TransactionSynchronizationManager.bindResource(resource, holder); + TransactionSynchronizationManager.registerSynchronization(this.transactionSynchronizationFactory.create(resource)); + } + } + return holder; + } + + /** + * Obtain the next message (if one is available). MAY return null + * if no message is immediately available. + * @return The message or null. + */ + protected abstract Message receiveMessage(); + + /** + * Handle a message. + * @param message The message. + */ + protected abstract void handleMessage(Message message); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index 535fb2183a..048e735642 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java @@ -16,8 +16,6 @@ package org.springframework.integration.endpoint; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.springframework.context.Lifecycle; import org.springframework.integration.Message; import org.springframework.integration.core.MessageHandler; @@ -33,9 +31,7 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky * @author Gary Russell */ -public class PollingConsumer extends AbstractPollingEndpoint { - - private final Log logger = LogFactory.getLog(this.getClass()); +public class PollingConsumer extends AbstractTransactionSynchronizingPollingEndpoint { private final PollableChannel inputChannel; @@ -74,23 +70,12 @@ public class PollingConsumer extends AbstractPollingEndpoint { @Override - protected boolean doPoll() { - Message message = this.syncIfTxAndReceive(); - if (this.logger.isDebugEnabled()){ - this.logger.debug("Poll resulted in Message: " + message); - } - if (message == null) { - if (this.logger.isDebugEnabled()){ - this.logger.debug("Received no Message during the poll, returning 'false'"); - } - return false; - } + protected void handleMessage(Message message) { this.handler.handleMessage(message); - return true; } @Override - protected Message doReceive() { + protected Message receiveMessage() { Message message = (this.receiveTimeout >= 0) ? this.inputChannel.receive(this.receiveTimeout) : this.inputChannel.receive(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 7812c221d1..2d1bef02e6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -35,7 +35,8 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky * @author Gary Russell */ -public class SourcePollingChannelAdapter extends AbstractPollingEndpoint implements TrackableComponent { +public class SourcePollingChannelAdapter extends AbstractTransactionSynchronizingPollingEndpoint + implements TrackableComponent { private volatile MessageSource source; @@ -88,38 +89,25 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme } @Override - protected boolean doPoll() { - - Message message = this.syncIfTxAndReceive(); - - if (this.logger.isDebugEnabled()){ - this.logger.debug("Poll resulted in Message: " + message); + protected void handleMessage(Message message) { + if (this.shouldTrack) { + message = MessageHistory.write(message, this); } - if (message != null) { - if (this.shouldTrack) { - message = MessageHistory.write(message, this); - } - try { - this.messagingTemplate.send(this.outputChannel, message); - } - catch (Exception e) { - if (e instanceof MessagingException) { - throw (MessagingException) e; - } - else { - throw new MessagingException(message, e); - } - } - return true; + try { + this.messagingTemplate.send(this.outputChannel, message); } - if (this.logger.isDebugEnabled()){ - this.logger.debug("Received no Message during the poll, returning 'false'"); + catch (Exception e) { + if (e instanceof MessagingException) { + throw (MessagingException) e; + } + else { + throw new MessagingException(message, e); + } } - return false; } @Override - protected Message doReceive() { + protected Message receiveMessage() { return this.source.receive(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java index 97c1ef5eb5..831c9b9a5c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java @@ -16,7 +16,6 @@ package org.springframework.integration.endpoint; -import org.springframework.integration.Message; import org.springframework.scheduling.support.PeriodicTrigger; /** @@ -33,19 +32,4 @@ public class PollingEndpointStub extends AbstractPollingEndpoint { throw new RuntimeException("intentional test failure"); } - @Override - protected Message doReceive() { - return null; - } - - @Override - protected Object getResourceToBind() { - return null; - } - - @Override - protected String getResourceKey() { - return null; - } - }