INT-4261: Do not Register Resource to TX Twice

JIRA: https://jira.spring.io/browse/INT-4261

When we have some `Advice` withing TX Advice which may perform `doPoll()`
several times, we unconditionally call
`transactionSynchronizationFactory.create(resource)`.
With the out-of-the-box implementations
`DefaultTransactionSynchronizationFactory` and
`PassThroughTransactionSynchronizationFactory`
we preform `TransactionSynchronizationManager.bindResource()`.
If resource is already there, an `IllegalStateException` is thrown

* Check that resource isn't bound already to the TX and don't create a new
`TransactionalResourceSynchronization` - just return `null`
* Check in the target users for the `null` before registering synchronization

Move resource registration to TX outside of out-of-the-box factories

* Fix condition in the `AbstractPollingEndpoint` for the resource
* Increase responsiveness of TX test to decreasing `fixed-delay`
and using `receive-timeout="-1"`

Author: artembilan

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

@@ -309,19 +309,33 @@ protected String getResourceKey() {
 	private IntegrationResourceHolder bindResourceHolderIfNecessary(String key, Object resource) {
 		if (this.transactionSynchronizationFactory != null && resource != null &&
 				TransactionSynchronizationManager.isActualTransactionActive()) {
+
 			TransactionSynchronization synchronization = this.transactionSynchronizationFactory.create(resource);
-			TransactionSynchronizationManager.registerSynchronization(synchronization);
-			if (synchronization instanceof IntegrationResourceHolderSynchronization) {
-				IntegrationResourceHolderSynchronization integrationSynchronization =
-						((IntegrationResourceHolderSynchronization) synchronization);
-				integrationSynchronization.setShouldUnbindAtCompletion(false);
-				IntegrationResourceHolder resourceHolder = integrationSynchronization.getResourceHolder();
+			if (synchronization != null) {
+				TransactionSynchronizationManager.registerSynchronization(synchronization);
+
+				if (synchronization instanceof IntegrationResourceHolderSynchronization) {
+					IntegrationResourceHolderSynchronization integrationSynchronization =
+							((IntegrationResourceHolderSynchronization) synchronization);
+					integrationSynchronization.setShouldUnbindAtCompletion(false);
+
+					if (!TransactionSynchronizationManager.hasResource(resource)) {
+						TransactionSynchronizationManager.bindResource(resource,
+								integrationSynchronization.getResourceHolder());
+					}
+				}
+			}
+
+			Object resourceHolder = TransactionSynchronizationManager.getResource(resource);
+			if (resourceHolder instanceof IntegrationResourceHolder) {
+				IntegrationResourceHolder integrationResourceHolder = (IntegrationResourceHolder) resourceHolder;
 				if (key != null) {
-					resourceHolder.addAttribute(key, resource);
+					integrationResourceHolder.addAttribute(key, resource);
 				}
-				return resourceHolder;
+				return integrationResourceHolder;
 			}
 		}
+
 		return null;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -20,8 +20,8 @@
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.transaction.support.TransactionSynchronization;
-import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
+
 /**
  * Default implementation of {@link TransactionSynchronizationFactory} which takes an instance of
  * {@link TransactionSynchronizationProcessor} allowing you to create a {@link TransactionSynchronization}
@@ -30,6 +30,7 @@
  * @author Gary Russell
  * @author Oleg Zhurakousky
  * @author Artem Bilan
+ *
  * @since 2.2
  */
 public class DefaultTransactionSynchronizationFactory implements TransactionSynchronizationFactory {
@@ -46,13 +47,9 @@ public DefaultTransactionSynchronizationFactory(TransactionSynchronizationProces
 	@Override
 	public TransactionSynchronization create(Object key) {
 		Assert.notNull(key, "'key' must not be null");
-		DefaultTransactionalResourceSynchronization synchronization = new DefaultTransactionalResourceSynchronization(key);
-		TransactionSynchronizationManager.bindResource(key, synchronization.getResourceHolder());
-		return synchronization;
+		return new DefaultTransactionalResourceSynchronization(key);
 	}
 
-	/**
-	 */
 	private final class DefaultTransactionalResourceSynchronization extends IntegrationResourceHolderSynchronization {
 
 		DefaultTransactionalResourceSynchronization(Object resourceKey) {

spring-integration-core/src/main/java/org/springframework/integration/transaction/PassThroughTransactionSynchronizationFactory.java

@@ -17,31 +17,24 @@
 package org.springframework.integration.transaction;
 
 import org.springframework.transaction.support.TransactionSynchronization;
-import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
 
 /**
  * A simple {@link TransactionSynchronizationFactory} implementation which produces
- * an {@link IntegrationResourceHolderSynchronization} and registers
- * an {@link IntegrationResourceHolder} under the provided {@code key} with
- * the current transaction scope.
+ * an {@link IntegrationResourceHolderSynchronization} with an {@link IntegrationResourceHolder}.
  *
  * @author Andreas Baer
+ * @author Artem Bilan
  *
  * @since 5.0
- *
- * @see TransactionSynchronizationManager#bindResource(Object, Object)
  */
 public class PassThroughTransactionSynchronizationFactory implements TransactionSynchronizationFactory {
 
 
 	@Override
 	public TransactionSynchronization create(Object key) {
 		Assert.notNull(key, "'key' must not be null");
-		IntegrationResourceHolderSynchronization synchronization =
-				new IntegrationResourceHolderSynchronization(new IntegrationResourceHolder(), key);
-		TransactionSynchronizationManager.bindResource(key, synchronization.getResourceHolder());
-		return synchronization;
+		return new IntegrationResourceHolderSynchronization(new IntegrationResourceHolder(), key);
 	}
 
 }

spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests-context.xml

@@ -10,7 +10,7 @@
 	</int:channel>
 
 	<int:service-activator input-channel="queueChannel" ref="service" method="handle">
-		<int:poller fixed-delay="5000">
+		<int:poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
 			<int:transactional synchronization-factory="txSyncFactory"/>
 		</int:poller>
 	</int:service-activator>
@@ -33,7 +33,7 @@
 	</int:channel>
 
 	<int:service-activator input-channel="queueChannel2" ref="service" method="handle">
-		<int:poller fixed-delay="5000">
+		<int:poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
 			<int:transactional synchronization-factory="txSyncFactory2"/>
 		</int:poller>
 	</int:service-activator>

spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -32,16 +32,20 @@
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.2
  *
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 public class TransactionSynchronizationQueueChannelTests {
 
 	@Autowired

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

@@ -16,9 +16,11 @@
 
 package org.springframework.integration.dispatcher;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import java.util.List;
@@ -30,8 +32,8 @@
 import org.junit.Test;
 
 import org.springframework.aop.Advisor;
+import org.springframework.aop.ProxyMethodInvocation;
 import org.springframework.aop.framework.Advised;
-import org.springframework.aop.support.DefaultPointcutAdvisor;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.test.util.TestUtils;
@@ -53,6 +55,7 @@
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Andreas Baer
+ * @author Artem Bilan
  */
 public class PollingTransactionTests {
 
@@ -82,25 +85,23 @@ public void transactionWithCommitAndAdvices() throws InterruptedException {
 		PollingConsumer advicedPoller = context.getBean("advicedSa", PollingConsumer.class);
 
 		List<Advice> adviceChain = TestUtils.getPropertyValue(advicedPoller, "adviceChain", List.class);
-		assertEquals(3, adviceChain.size());
+		assertEquals(4, adviceChain.size());
 		Runnable poller = TestUtils.getPropertyValue(advicedPoller, "poller", Runnable.class);
 		Callable<?> pollingTask = TestUtils.getPropertyValue(poller, "pollingTask", Callable.class);
 		assertTrue("Poller is not Advised", pollingTask instanceof Advised);
 		Advisor[] advisors = ((Advised) pollingTask).getAdvisors();
-		assertEquals(3, advisors.length);
+		assertEquals(4, advisors.length);
 
-		assertTrue("First advisor is not TX", ((DefaultPointcutAdvisor) advisors[0]).getAdvice() instanceof
-				TransactionInterceptor);
+		assertThat("First advisor is not TX", advisors[0].getAdvice(), instanceOf(TransactionInterceptor.class));
 		TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
 		MessageChannel input = (MessageChannel) context.getBean("goodInputWithAdvice");
 		PollableChannel output = (PollableChannel) context.getBean("output");
 		assertEquals(0, txManager.getCommitCount());
 		assertEquals(0, txManager.getRollbackCount());
-		input.send(new GenericMessage<String>("test"));
+		input.send(new GenericMessage<>("test"));
 		txManager.waitForCompletion(10000);
 		Message<?> message = output.receive(0);
 		assertNotNull(message);
-		assertEquals(1, txManager.getCommitCount());
 		assertEquals(0, txManager.getRollbackCount());
 		context.close();
 	}
@@ -220,7 +221,7 @@ public void commitFailureAndHandlerFailureTest() throws Throwable {
 		PollableChannel output = (PollableChannel) context.getBean("output");
 		PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel");
 		assertEquals(0, txManager.getCommitCount());
-		inputTxFail.send(new GenericMessage<>("commitFalilureTest"));
+		inputTxFail.send(new GenericMessage<>("commitFailureTest"));
 		Message<?> errorMessage = errorChannel.receive(10000);
 		assertNotNull(errorMessage);
 		Object payload = errorMessage.getPayload();
@@ -254,6 +255,17 @@ public Object invoke(MethodInvocation invocation) throws Throwable {
 
 	}
 
+	public static class SimpleRepeatAdvice implements MethodInterceptor {
+
+		@Override
+		public Object invoke(MethodInvocation invocation) throws Throwable {
+			((ProxyMethodInvocation) invocation).invocableClone().proceed();
+
+			return invocation.proceed();
+		}
+
+	}
+
 	@SuppressWarnings("serial")
 	public static class FailingCommitTransactionManager extends TestTransactionManager {
 

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

@@ -24,25 +24,26 @@
 
 	<service-activator input-channel="badInput" ref="testBean"
 		method="bad" output-channel="output">
-		<poller max-messages-per-poll="1" fixed-rate="10000">
+		<poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
 			<transactional transaction-manager="txManager" />
 		</poller>
 	</service-activator>
 
 	<service-activator input-channel="goodInput" ref="testBean"
 		method="good" output-channel="output">
-		<poller max-messages-per-poll="1" fixed-rate="10000">
+		<poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
 			<transactional transaction-manager="txManager" />
 		</poller>
 	</service-activator>
 
 	<service-activator id="advicedSa" input-channel="goodInputWithAdvice" ref="testBean"
 		method="good" output-channel="output">
-		<poller max-messages-per-poll="1" fixed-rate="10000">
+		<poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
 			<advice-chain>
 				<ref bean="txAdvise"/>
 				<ref bean="adviceA" />
 				<beans:bean class="org.springframework.integration.dispatcher.PollingTransactionTests.SampleAdvice"/>
+				<beans:bean class="org.springframework.integration.dispatcher.PollingTransactionTests.SimpleRepeatAdvice"/>
 			</advice-chain>
 		</poller>
 	</service-activator>

spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -154,7 +154,7 @@ public void setApplicationEventPublisher(ApplicationEventPublisher applicationEv
 
 	@Override // guarded by super#lifecycleLock
 	protected void doStart() {
-		final TaskScheduler scheduler =  this.getTaskScheduler();
+		final TaskScheduler scheduler = this.getTaskScheduler();
 		Assert.notNull(scheduler, "'taskScheduler' must not be null");
 		if (this.sendingTaskExecutor == null) {
 			this.sendingTaskExecutor = Executors.newFixedThreadPool(1);
@@ -187,19 +187,29 @@ private Runnable createMessageSendingTask(final Object mailMessage) {
 			@SuppressWarnings("unchecked")
 			org.springframework.messaging.Message<?> message =
 					mailMessage instanceof Message
-						? ImapIdleChannelAdapter.this.getMessageBuilderFactory().withPayload(mailMessage).build()
-						:  (org.springframework.messaging.Message<Object>) mailMessage;
+							? ImapIdleChannelAdapter.this.getMessageBuilderFactory().withPayload(mailMessage).build()
+							: (org.springframework.messaging.Message<Object>) mailMessage;
 
 			if (TransactionSynchronizationManager.isActualTransactionActive()) {
 				if (ImapIdleChannelAdapter.this.transactionSynchronizationFactory != null) {
 					TransactionSynchronization synchronization =
 							ImapIdleChannelAdapter.this.transactionSynchronizationFactory
 									.create(ImapIdleChannelAdapter.this);
-					TransactionSynchronizationManager.registerSynchronization(synchronization);
-					if (synchronization instanceof IntegrationResourceHolderSynchronization) {
-						IntegrationResourceHolder holder =
-								((IntegrationResourceHolderSynchronization) synchronization).getResourceHolder();
-						holder.setMessage(message);
+					if (synchronization != null) {
+						TransactionSynchronizationManager.registerSynchronization(synchronization);
+
+						if (synchronization instanceof IntegrationResourceHolderSynchronization
+								&& !TransactionSynchronizationManager.hasResource(ImapIdleChannelAdapter.this)) {
+
+							TransactionSynchronizationManager.bindResource(ImapIdleChannelAdapter.this,
+									((IntegrationResourceHolderSynchronization) synchronization).getResourceHolder());
+						}
+
+						Object resourceHolder =
+								TransactionSynchronizationManager.getResource(ImapIdleChannelAdapter.this);
+						if (resourceHolder instanceof IntegrationResourceHolder) {
+							((IntegrationResourceHolder) resourceHolder).setMessage(message);
+						}
 					}
 				}
 			}
@@ -262,7 +272,7 @@ private class IdleTask implements Runnable {
 
 		@Override
 		public void run() {
-			final TaskScheduler scheduler =  getTaskScheduler();
+			final TaskScheduler scheduler = getTaskScheduler();
 			Assert.notNull(scheduler, "'taskScheduler' must not be null");
 			/*
 			 * The following shouldn't be necessary because doStart() will have ensured we have