From a127a460839af02a4ae5f529b0a5f401858046c7 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 3 Jan 2013 12:08:01 -0500 Subject: [PATCH] INT-2825 Remove Temporary 2.2. Interface AbstractTransactionSynchronizingPollingEndpoint was introduced late in the 2.2 release process to avoid a breaking change to AbstractPollingEndpoint. For 3.0, its methods are now pulled-up into APE. Further change: * remove Deprecated Setter (PollerMetadata) * fix failing test --- .../endpoint/AbstractPollingEndpoint.java | 101 ++++++++++----- ...ansactionSynchronizingPollingEndpoint.java | 117 ------------------ .../integration/endpoint/PollingConsumer.java | 4 +- .../endpoint/SourcePollingChannelAdapter.java | 4 +- .../PollingConsumerEndpointTests.java | 17 +-- .../endpoint/PollingEndpointStub.java | 10 +- ...PseudoTransactionalMessageSourceTests.java | 36 ++++-- ...hSpringContextIntegrationTests-context.xml | 2 +- 8 files changed, 112 insertions(+), 179 deletions(-) delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractTransactionSynchronizingPollingEndpoint.java 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 b9a7341cb1..603fda3c71 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -30,11 +30,14 @@ import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessagingException; 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.ExpressionEvaluatingTransactionSynchronizationProcessor; +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; @@ -67,22 +70,12 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private final Object initializationMonitor = new Object(); + private volatile TransactionSynchronizationFactory transactionSynchronizationFactory; + public AbstractPollingEndpoint() { this.setPhase(Integer.MAX_VALUE); } - /** - * @deprecated As of release 2.0.2, use individual setters - */ - @Deprecated - public void setPollerMetadata(PollerMetadata pollerMetadata){ - Assert.notNull(pollerMetadata, "'pollerMetadata' must not be null."); - this.setAdviceChain(pollerMetadata.getAdviceChain()); - this.setMaxMessagesPerPoll(pollerMetadata.getMaxMessagesPerPoll()); - this.setTaskExecutor(pollerMetadata.getTaskExecutor()); - this.setTrigger(pollerMetadata.getTrigger()); - } - public void setTaskExecutor(Executor taskExecutor) { this.taskExecutor = (taskExecutor != null ? taskExecutor : new SyncTaskExecutor()); } @@ -107,6 +100,10 @@ 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) { @@ -181,21 +178,28 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.initialized = false; } - /** - * @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() { + private boolean doPoll() { + IntegrationResourceHolder holder = this.bindResourceHolderIfNecessary( + this.getResourceKey(), this.getResourceToBind()); Message message = this.receiveMessage(); - if (message != null) { - this.handleMessage(message); - return true; + boolean result; + if (message == null) { + if (this.logger.isDebugEnabled()){ + this.logger.debug("Received no Message during the poll, returning 'false'"); + } + result = false; } - return 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; } /** @@ -203,16 +207,49 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement * if no message is immediately available. * @return The message or null. */ - protected Message receiveMessage() { - throw new UnsupportedOperationException("Subclass must implement receiveMessage()"); - } + protected abstract Message receiveMessage(); /** * Handle a message. * @param message The message. */ - protected void handleMessage(Message message) { - throw new UnsupportedOperationException("Subclass must implement handleMessage()"); + protected abstract void handleMessage(Message message); + + /** + * 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; + } + + 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; } /** 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 deleted file mode 100644 index 26da36a7cf..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractTransactionSynchronizingPollingEndpoint.java +++ /dev/null @@ -1,117 +0,0 @@ -/* - * 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 048e735642..5e0e4328ed 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -31,7 +31,7 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky * @author Gary Russell */ -public class PollingConsumer extends AbstractTransactionSynchronizingPollingEndpoint { +public class PollingConsumer extends AbstractPollingEndpoint { private final PollableChannel inputChannel; 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 2d1bef02e6..fd550f7fd1 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -35,7 +35,7 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky * @author Gary Russell */ -public class SourcePollingChannelAdapter extends AbstractTransactionSynchronizingPollingEndpoint +public class SourcePollingChannelAdapter extends AbstractPollingEndpoint implements TrackableComponent { private volatile MessageSource source; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java index 86d76a93e1..cf6175a5e1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java @@ -34,14 +34,12 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.After; import org.junit.Before; import org.junit.Test; - import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.Message; 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.PollerMetadata; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.TriggerContext; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -121,18 +119,15 @@ public class PollingConsumerEndpointTests { assertEquals(5, consumer.counter.get()); verify(channelMock); } - - @SuppressWarnings("deprecation") + @Test - public void multipleMessagesWithPollerMetadata() { + public void multipleMessagesWithMaxMessagesAndTrigger() { expect(channelMock.receive()).andReturn(message).times(5); replay(channelMock); - - PollerMetadata pollerMetadata = new PollerMetadata(); - pollerMetadata.setMaxMessagesPerPoll(5); - pollerMetadata.setTrigger(trigger); - endpoint.setPollerMetadata(pollerMetadata); - + + endpoint.setMaxMessagesPerPoll(5); + endpoint.setTrigger(trigger); + endpoint.start(); trigger.await(); endpoint.stop(); 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 831c9b9a5c..ed800db767 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2013 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. @@ -16,10 +16,12 @@ package org.springframework.integration.endpoint; +import org.springframework.integration.Message; import org.springframework.scheduling.support.PeriodicTrigger; /** * @author Jonas Partner + * @author Gary Russell */ public class PollingEndpointStub extends AbstractPollingEndpoint { @@ -28,7 +30,11 @@ public class PollingEndpointStub extends AbstractPollingEndpoint { } @Override - protected boolean doPoll() { + protected void handleMessage(Message message) { + } + + @Override + protected Message receiveMessage() { throw new RuntimeException("intentional test failure"); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java index e44cbaad98..4c4dbc7e48 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -17,12 +17,13 @@ package org.springframework.integration.endpoint; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; +import java.lang.reflect.Method; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Ignore; import org.junit.Test; - import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; @@ -31,8 +32,8 @@ import org.springframework.integration.core.PollableChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory; import org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor; -import org.springframework.integration.transaction.PseudoTransactionManager; import org.springframework.integration.transaction.IntegrationResourceHolder; +import org.springframework.integration.transaction.PseudoTransactionManager; import org.springframework.integration.transaction.TransactionSynchronizationFactory; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; @@ -81,7 +82,7 @@ public class PseudoTransactionalMessageSourceTests { TransactionSynchronizationManager.initSynchronization(); TransactionSynchronizationManager.setActualTransactionActive(true); - adapter.doPoll(); + doPoll(adapter); TransactionSynchronizationUtils.triggerBeforeCommit(false); TransactionSynchronizationUtils.triggerAfterCommit(); Message beforeCommitMessage = queueChannel.receive(1000); @@ -122,7 +123,7 @@ public class PseudoTransactionalMessageSourceTests { TransactionSynchronizationManager.initSynchronization(); TransactionSynchronizationManager.setActualTransactionActive(true); - adapter.doPoll(); + doPoll(adapter); TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK); Message rollbackMessage = queueChannel.receive(1000); assertNotNull(rollbackMessage); @@ -163,7 +164,7 @@ public class PseudoTransactionalMessageSourceTests { } }); - adapter.doPoll(); + doPoll(adapter); return null; } }); @@ -206,7 +207,7 @@ public class PseudoTransactionalMessageSourceTests { } }); - adapter.doPoll(); + doPoll(adapter); throw new RuntimeException("Force rollback"); } }); @@ -249,7 +250,7 @@ public class PseudoTransactionalMessageSourceTests { } }); - adapter.doPoll(); + doPoll(adapter); status.setRollbackOnly(); return null; } @@ -271,12 +272,12 @@ public class PseudoTransactionalMessageSourceTests { }); TransactionSynchronizationManager.setActualTransactionActive(true); - adapter.doPoll(); + doPoll(adapter); TransactionSynchronizationManager.setActualTransactionActive(false); // Before INT-2777 this test was failed here TransactionSynchronizationManager.setActualTransactionActive(true); - adapter.doPoll(); + doPoll(adapter); TransactionSynchronizationManager.setActualTransactionActive(false); } @@ -309,7 +310,7 @@ public class PseudoTransactionalMessageSourceTests { TransactionSynchronizationManager.initSynchronization(); TransactionSynchronizationManager.setActualTransactionActive(true); - adapter.doPoll(); + doPoll(adapter); TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED); TransactionSynchronizationManager.clearSynchronization(); TransactionSynchronizationManager.setActualTransactionActive(false); @@ -321,13 +322,24 @@ public class PseudoTransactionalMessageSourceTests { //TODO: Need new JIRA issue to fix it TransactionSynchronizationManager.initSynchronization(); TransactionSynchronizationManager.setActualTransactionActive(true); - adapter.doPoll(); + doPoll(adapter); TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED); TransactionSynchronizationManager.clearSynchronization(); TransactionSynchronizationManager.setActualTransactionActive(false); assertEquals(2, txSyncCounter.get()); } + protected void doPoll(SourcePollingChannelAdapter adapter) { + try { + Method method = AbstractPollingEndpoint.class.getDeclaredMethod("doPoll"); + method.setAccessible(true); + method.invoke(adapter); + } + catch (Exception e) { + fail("Failed to invoke doPoll(): " + e.toString()); + } + } + public class Bar { public String getValue() { return "bar"; diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapterWithSpringContextIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapterWithSpringContextIntegrationTests-context.xml index 24f67a0959..65a42fa758 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapterWithSpringContextIntegrationTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapterWithSpringContextIntegrationTests-context.xml @@ -64,7 +64,7 @@ + class="org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean">