INT-1461: fix source polling channel adapter

spring-integration-core/pom.xml

@@ -28,6 +28,7 @@
 		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-tx</artifactId>
+			<optional>true</optional>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.commons</groupId>

spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java

@@ -129,6 +129,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 			spca.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
 			spca.setTaskExecutor(this.pollerMetadata.getTaskExecutor());
 			spca.setAdviceChain(this.pollerMetadata.getAdviceChain());
+			spca.setPollingDecorator(pollerMetadata.getPollingDecorator());
 			spca.setAutoStartup(this.autoStartup);
 			spca.setBeanName(this.beanName);
 			spca.setBeanFactory(this.beanFactory);

spring-integration-core/src/main/java/org/springframework/integration/config/TransactionalCallbackDecorator.java

@@ -13,13 +13,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.integration.endpoint;
+package org.springframework.integration.config;
 
 import java.util.Properties;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.integration.endpoint.PollerCallbackDecorator;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
 import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;

spring-integration-core/src/main/java/org/springframework/integration/config/xml/PollerParser.java

@@ -211,7 +211,7 @@ public class PollerParser extends AbstractBeanDefinitionParser {
 		transactionalProperties.setProperty("readOnly", txElement.getAttribute("read-only"));
 		
 		BeanDefinitionBuilder pollingDecoratorBuilder = 
-			BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.endpoint.TransactionalCallbackDecorator");
+			BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.config.TransactionalCallbackDecorator");
 		pollingDecoratorBuilder.addPropertyValue("transactionalProperties", transactionalProperties);
 		
 		targetBuilder.addPropertyValue("pollingDecorator", pollingDecoratorBuilder.getBeanDefinition());

spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java

@@ -22,15 +22,16 @@ import static org.junit.Assert.assertTrue;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
 import org.junit.Test;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.endpoint.PollerCallbackDecorator;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.test.util.TestUtils;
@@ -74,6 +75,44 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
 		assertTrue("adviceChain was not applied", adviceApplied.get());
 	}
 
+	@Test
+	public void testTransactionalAdviceChain() throws Exception {
+		SourcePollingChannelAdapterFactoryBean factoryBean = new SourcePollingChannelAdapterFactoryBean();
+		QueueChannel outputChannel = new QueueChannel();
+		TestApplicationContext context = TestUtils.createTestApplicationContext();
+		factoryBean.setBeanFactory(context.getBeanFactory());
+		factoryBean.setBeanClassLoader(ClassUtils.getDefaultClassLoader());
+		factoryBean.setOutputChannel(outputChannel);
+		factoryBean.setSource(new TestSource());
+		PollerMetadata pollerMetadata = new PollerMetadata();
+		List<Advice> adviceChain = new ArrayList<Advice>();
+		final AtomicBoolean adviceApplied = new AtomicBoolean(false);
+		adviceChain.add(new MethodInterceptor() {
+			public Object invoke(MethodInvocation invocation) throws Throwable {
+				adviceApplied.set(true);
+				return invocation.proceed();
+			}
+		});
+		pollerMetadata.setTrigger(new PeriodicTrigger(5000));
+		pollerMetadata.setMaxMessagesPerPoll(1);
+		final AtomicInteger count = new AtomicInteger();
+		pollerMetadata.setPollingDecorator(new PollerCallbackDecorator() {
+			public Object decorate(Object poller) {
+				count.incrementAndGet();
+				return poller;
+			}
+		});
+		pollerMetadata.setAdviceChain(adviceChain);
+		factoryBean.setPollerMetadata(pollerMetadata);
+		factoryBean.setAutoStartup(true);
+		factoryBean.afterPropertiesSet();
+		context.registerEndpoint("testPollingEndpoint", factoryBean.getObject());
+		context.refresh();
+		Message<?> message = outputChannel.receive(30000);
+		assertEquals("test", message.getPayload());
+		assertEquals(1, count.get());
+		assertTrue("adviceChain was not applied", adviceApplied.get());
+	}
 
 	private static class TestSource implements MessageSource<String> {
 

spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java

@@ -19,6 +19,7 @@ package org.springframework.integration.dispatcher;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 
 import java.util.List;
 
@@ -26,17 +27,19 @@ import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
 import org.junit.Test;
-
-import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.aop.Advisor;
+import org.springframework.aop.framework.Advised;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.integration.util.TestTransactionManager;
 import org.springframework.transaction.IllegalTransactionStateException;
 import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.interceptor.TransactionAttributeSourceAdvisor;
 
 /**
  * @author Mark Fisher
@@ -66,8 +69,14 @@ public class PollingTransactionTests {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"transactionTests.xml", this.getClass());
 		PollingConsumer advicedPoller = context.getBean("advicedSa", PollingConsumer.class);
-		List<Advice> adviceChain = (List<Advice>) new DirectFieldAccessor(advicedPoller).getPropertyValue("adviceChain");
+		List<Advice> adviceChain = TestUtils.getPropertyValue(advicedPoller, "adviceChain");
 		assertEquals(2, adviceChain.size());
+		Runnable poller = TestUtils.getPropertyValue(advicedPoller, "poller");
+		assertTrue("Poller is not Advised", poller instanceof Advised);
+		Advisor[] advisors = ((Advised)poller).getAdvisors();
+		assertEquals(3, advisors.length);
+		// System.err.println(Arrays.asList(advisors));
+		assertTrue("First advisor is not TX", advisors[0] instanceof TransactionAttributeSourceAdvisor);
 		TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
 		MessageChannel input = (MessageChannel) context.getBean("goodInputWithAdvice");
 		PollableChannel output = (PollableChannel) context.getBean("output");

spring-integration-core/src/test/java/org/springframework/integration/test/util/TestUtils.java

@@ -22,7 +22,6 @@ import static org.junit.Assert.assertThat;
 import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
 
 import org.hamcrest.Matcher;
-
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.FatalBeanException;
 import org.springframework.beans.factory.BeanFactory;
@@ -54,7 +53,7 @@ import org.springframework.util.ErrorHandler;
  */
 public abstract class TestUtils {
 
-    public static Object getPropertyValue(Object root, String propertyPath) {
+	public static <T> T getPropertyValue(Object root, String propertyPath) {
         Object value = null;
         DirectFieldAccessor accessor = new DirectFieldAccessor(root);
         String[] tokens = propertyPath.split("\\.");
@@ -69,7 +68,9 @@ public abstract class TestUtils {
                         "intermediate property '" + tokens[i] + "' is null");
             }
         }
-        return value;
+        @SuppressWarnings("unchecked")
+        T result = (T) value;
+		return result;
     }
 
     @SuppressWarnings("unchecked")