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

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;
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractTransactionSynchronizingPollingEndpoint.java

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

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;
 

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;

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

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");
 	}
 

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";

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapterWithSpringContextIntegrationTests-context.xml

@@ -64,7 +64,7 @@
 	<bean id="rowMapper" class="org.springframework.integration.jdbc.storedproc.PrimeMapper"/>
 
 	<bean id="storedProcedureEndpoint"
-				class="org.springframework.integration.endpoint.SourcePollingChannelAdapter">
+				class="org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean">
 		<property name="source" ref="source"/>
 		<property name="outputChannel" ref="outputChannel"/>
 		<property name="pollerMetadata" ref="defaultPoller"/>