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

INT-2815 Refactor AbstractPollingEndpoint

Browse Source 
Recent changes to APE were sub-optimal because the subclasses
had to call back into the parent abstract class.

Deprecate doPoll() on APE.

Introduce a temporary intermediate abstract class that supports
transaction synchronization.

This class will be deprecated in 3.0.0 when its methods will
be pulled up into APE.

INT-2815 Polishing

Make ATSPE package visibility.

Add helpers to enable early migration to 3.0 structure.

INT-2815 Polishing

Fix javadoc to emphasize that doPoll() will no longer
be overridable in 3.0.

Rename doReceive() to receiveMessage() to match more closely
with handleMessage().

INT-2815 Polishing

Fix message in UnsupportedOperationException.
This commit is contained in:
Gary Russell
2012-11-19 15:24:25 -05:00
committed by Mark Fisher
parent 4822d9b209
commit 6f4800a904
5 changed files with 164 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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

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

@@ -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.
* <p/>
* 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);
}

21
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java
View File

@@ -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();

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

@@ -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();
}

16
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java
View File

@@ -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;
}
}