INT-3408: Fix IRHolder TX-sync Inconsistency

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

* Add `TransactionSynchronizationFactoryBean`
* Introduce `IntegrationResourceHolderSynchronization`
to get deal with `IntegrationResourceHolder` for custom `TransactionSynchronization`
* Encapsulate `IntegrationResourceHolder` in the `DefaultTransactionSynchronizationFactory`
* Get deal with `IntegrationResourceHolder` and bind it to the `TransactionSynchronizationManager`
only if `TransactionSynchronization` from `transactionSynchronizationFactory`
is `synchronization instanceof IntegrationResourceHolderSynchronization`

INT-3408: Encapsulate binding in the `DefTxSyncF`

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

@@ -30,6 +30,7 @@ import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor;
 import org.springframework.integration.transaction.IntegrationResourceHolder;
+import org.springframework.integration.transaction.IntegrationResourceHolderSynchronization;
 import org.springframework.integration.transaction.TransactionSynchronizationFactory;
 import org.springframework.integration.util.ErrorHandlingTaskExecutor;
 import org.springframework.messaging.Message;
@@ -38,6 +39,7 @@ import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.scheduling.Trigger;
 import org.springframework.scheduling.support.PeriodicTrigger;
+import org.springframework.transaction.support.TransactionSynchronization;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -48,6 +50,7 @@ import org.springframework.util.ErrorHandler;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public abstract class AbstractPollingEndpoint extends AbstractEndpoint implements BeanClassLoaderAware {
 
@@ -240,19 +243,22 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 	}
 
 	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);
+				TransactionSynchronization synchronization = this.transactionSynchronizationFactory.create(resource);
+				TransactionSynchronizationManager.registerSynchronization(synchronization);
+				if (synchronization instanceof IntegrationResourceHolderSynchronization) {
+					IntegrationResourceHolder holder =
+							((IntegrationResourceHolderSynchronization) synchronization).getResourceHolder();
+					if (key != null) {
+						holder.addAttribute(key, resource);
+					}
+					return holder;
 				}
-				TransactionSynchronizationManager.bindResource(resource, holder);
-				TransactionSynchronizationManager.registerSynchronization(this.transactionSynchronizationFactory.create(resource));
 			}
 		}
-		return holder;
+		return null;
 	}
 
 	/**
@@ -292,6 +298,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 				}
 			});
 		}
+
 	}
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2014 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
@@ -15,7 +15,6 @@ package org.springframework.integration.transaction;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-import org.springframework.transaction.support.ResourceHolderSynchronization;
 import org.springframework.transaction.support.TransactionSynchronization;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
@@ -26,6 +25,7 @@ import org.springframework.util.Assert;
  *
  * @author Gary Russell
  * @author Oleg Zhurakousky
+ * @author Artem Bilan
  * @since 2.2
  */
 public class DefaultTransactionSynchronizationFactory implements TransactionSynchronizationFactory {
@@ -41,22 +41,17 @@ public class DefaultTransactionSynchronizationFactory implements TransactionSync
 
 	public TransactionSynchronization create(Object key) {
 		Assert.notNull(key, "'key' must not be null");
-		Object resourceHolder = TransactionSynchronizationManager.getResource(key);
-		Assert.isInstanceOf(IntegrationResourceHolder.class, resourceHolder);
-		return new DefaultTransactionalResourceSynchronization((IntegrationResourceHolder) resourceHolder, key);
+		DefaultTransactionalResourceSynchronization synchronization = new DefaultTransactionalResourceSynchronization(key);
+		TransactionSynchronizationManager.bindResource(key, synchronization.getResourceHolder());
+		return synchronization;
 	}
 
 	/**
 	 */
-	private class DefaultTransactionalResourceSynchronization
-		extends ResourceHolderSynchronization<IntegrationResourceHolder, Object> {
+	private class DefaultTransactionalResourceSynchronization extends IntegrationResourceHolderSynchronization {
 
-		private final IntegrationResourceHolder resourceHolder;
-
-		public DefaultTransactionalResourceSynchronization(IntegrationResourceHolder resourceHolder,
-				Object resourceKey) {
-			super(resourceHolder, resourceKey);
-			this.resourceHolder = resourceHolder;
+		public DefaultTransactionalResourceSynchronization(Object resourceKey) {
+			super(new IntegrationResourceHolder(), resourceKey);
 		}
 
 		@Override
@@ -95,6 +90,7 @@ public class DefaultTransactionSynchronizationFactory implements TransactionSync
 			}
 			super.afterCompletion(status);
 		}
+
 	}
 
 }

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2014 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.transaction;
+
+import org.springframework.transaction.support.ResourceHolderSynchronization;
+
+/**
+ * The base {@link ResourceHolderSynchronization} for {@link IntegrationResourceHolder}.
+ *
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public abstract class IntegrationResourceHolderSynchronization
+		extends ResourceHolderSynchronization<IntegrationResourceHolder, Object> {
+
+	protected final IntegrationResourceHolder resourceHolder;
+
+	public IntegrationResourceHolderSynchronization(IntegrationResourceHolder resourceHolder,
+			Object resourceKey) {
+		super(resourceHolder, resourceKey);
+		this.resourceHolder = resourceHolder;
+	}
+
+	public IntegrationResourceHolder getResourceHolder() {
+		return resourceHolder;
+	}
+
+}

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

@@ -0,0 +1,198 @@
+/*
+ * Copyright 2014 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.transaction;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * The {@link FactoryBean} implementation (with {@code Builder} style) to be used
+ * from JavaConfig to populate {@link DefaultTransactionSynchronizationFactory} bean.
+ *
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public class TransactionSynchronizationFactoryBean implements FactoryBean<DefaultTransactionSynchronizationFactory>,
+		BeanFactoryAware {
+
+	private final SpelExpressionParser PARSER = new SpelExpressionParser();
+
+	private BeanFactory beanFactory;
+
+	private volatile String beforeCommitExpression;
+
+	private volatile String afterCommitExpression;
+
+	private volatile String afterRollbackExpression;
+
+	private volatile MessageChannel beforeCommitChannel;
+
+	private volatile String beforeCommitChannelName;
+
+	private volatile MessageChannel afterCommitChannel;
+
+	private volatile String afterCommitChannelName;
+
+	private volatile MessageChannel afterRollbackChannel;
+
+	private volatile String afterRollbackChannelName;
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+	}
+
+	public TransactionSynchronizationFactoryBean beforeCommit(String expression) {
+		return beforeCommit(expression, this.beforeCommitChannel);
+	}
+
+	public TransactionSynchronizationFactoryBean beforeCommit(String expression, String messageChannel) {
+		Assert.state(StringUtils.hasText(expression) || StringUtils.hasText(messageChannel),
+				"At least one attribute ('expression' and/or 'messageChannel') must be defined");
+		this.beforeCommitExpression = expression;
+		this.beforeCommitChannelName = messageChannel;
+		this.beforeCommitChannel = null;
+		return this;
+	}
+
+	public TransactionSynchronizationFactoryBean beforeCommit(MessageChannel messageChannel) {
+		return beforeCommit(this.beforeCommitExpression, messageChannel);
+	}
+
+	public TransactionSynchronizationFactoryBean beforeCommit(String expression, MessageChannel messageChannel) {
+		Assert.state(StringUtils.hasText(expression) || messageChannel != null,
+				"At least one attribute ('expression' and/or 'messageChannel') must be defined");
+		this.beforeCommitExpression = expression;
+		this.beforeCommitChannel = messageChannel;
+		this.beforeCommitChannelName = null;
+		return this;
+	}
+
+	public TransactionSynchronizationFactoryBean afterCommit(String expression) {
+		return afterCommit(expression, this.afterCommitChannel);
+	}
+
+	public TransactionSynchronizationFactoryBean afterCommit(String expression, String messageChannel) {
+		Assert.state(StringUtils.hasText(expression) || StringUtils.hasText(messageChannel),
+				"At least one attribute ('expression' and/or 'messageChannel') must be defined");
+		this.afterCommitExpression = expression;
+		this.afterCommitChannelName = messageChannel;
+		this.afterCommitChannel = null;
+		return this;
+	}
+
+	public TransactionSynchronizationFactoryBean afterCommit(MessageChannel messageChannel) {
+		return afterCommit(this.afterCommitExpression, messageChannel);
+	}
+
+	public TransactionSynchronizationFactoryBean afterCommit(String expression, MessageChannel messageChannel) {
+		Assert.state(StringUtils.hasText(expression) || messageChannel != null,
+				"At least one attribute ('expression' and/or 'messageChannel') must be defined");
+		this.afterCommitExpression = expression;
+		this.afterCommitChannel = messageChannel;
+		this.afterCommitChannelName = null;
+		return this;
+	}
+
+	public TransactionSynchronizationFactoryBean afterRollback(String expression) {
+		return afterRollback(expression, this.afterRollbackChannel);
+	}
+
+	public TransactionSynchronizationFactoryBean afterRollback(String expression, String messageChannel) {
+		Assert.state(StringUtils.hasText(expression) || StringUtils.hasText(messageChannel),
+				"At least one attribute ('expression' and/or 'messageChannel') must be defined");
+		this.afterRollbackExpression = expression;
+		this.afterRollbackChannelName = messageChannel;
+		this.afterRollbackChannel = null;
+		return this;
+	}
+
+	public TransactionSynchronizationFactoryBean afterRollback(MessageChannel messageChannel) {
+		return afterRollback(this.afterRollbackExpression, messageChannel);
+	}
+
+	public TransactionSynchronizationFactoryBean afterRollback(String expression, MessageChannel messageChannel) {
+		Assert.state(StringUtils.hasText(expression) || messageChannel != null,
+				"At least one attribute ('expression' and/or 'messageChannel') must be defined");
+		this.afterRollbackExpression = expression;
+		this.afterRollbackChannel = messageChannel;
+		this.afterRollbackChannelName = null;
+		return this;
+	}
+
+	@Override
+	public DefaultTransactionSynchronizationFactory getObject() throws Exception {
+		ExpressionEvaluatingTransactionSynchronizationProcessor processor =
+				new ExpressionEvaluatingTransactionSynchronizationProcessor();
+
+		if (StringUtils.hasText(this.beforeCommitExpression)) {
+			processor.setBeforeCommitExpression(PARSER.parseExpression(this.beforeCommitExpression));
+		}
+		if (StringUtils.hasText(this.afterCommitExpression)) {
+			processor.setAfterCommitExpression(PARSER.parseExpression(this.afterCommitExpression));
+		}
+		if (StringUtils.hasText(this.afterRollbackExpression)) {
+			processor.setAfterRollbackExpression(PARSER.parseExpression(this.afterRollbackExpression));
+		}
+
+		if (StringUtils.hasText(this.beforeCommitChannelName)) {
+			this.beforeCommitChannel = this.beanFactory.getBean(this.beforeCommitChannelName, MessageChannel.class);
+		}
+		if (this.beforeCommitChannel != null) {
+			processor.setBeforeCommitChannel(this.beforeCommitChannel);
+		}
+
+		if (StringUtils.hasText(this.afterCommitChannelName)) {
+			this.afterCommitChannel = this.beanFactory.getBean(this.afterCommitChannelName, MessageChannel.class);
+		}
+		if (this.afterCommitChannel != null) {
+			processor.setAfterCommitChannel(this.afterCommitChannel);
+		}
+
+		if (StringUtils.hasText(this.afterRollbackChannelName)) {
+			this.afterRollbackChannel = this.beanFactory.getBean(this.afterRollbackChannelName, MessageChannel.class);
+		}
+		if (this.afterRollbackChannel != null) {
+			processor.setAfterRollbackChannel(this.afterRollbackChannel);
+		}
+
+		if (beanFactory instanceof AutowireCapableBeanFactory) {
+			((AutowireCapableBeanFactory) beanFactory).initializeBean(processor, null);
+		}
+
+		return new DefaultTransactionSynchronizationFactory(processor);
+	}
+
+	@Override
+	public Class<?> getObjectType() {
+		return DefaultTransactionSynchronizationFactory.class;
+	}
+
+
+	@Override
+	public boolean isSingleton() {
+		return true;
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java

@@ -24,15 +24,19 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.commons.logging.Log;
 import org.hamcrest.Matchers;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.mockito.Mockito;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory;
@@ -40,6 +44,7 @@ import org.springframework.integration.transaction.ExpressionEvaluatingTransacti
 import org.springframework.integration.transaction.IntegrationResourceHolder;
 import org.springframework.integration.transaction.PseudoTransactionManager;
 import org.springframework.integration.transaction.TransactionSynchronizationFactory;
+import org.springframework.integration.transaction.TransactionSynchronizationFactoryBean;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
@@ -83,10 +88,10 @@ public class PseudoTransactionalMessageSourceTests {
 
 			public Message<String> receive() {
 				GenericMessage<String> message = new GenericMessage<String>("foo");
-				((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
-						.addAttribute("baz", "qux");
-				((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
-						.addAttribute("bix", "qox");
+				IntegrationResourceHolder holder =
+						(IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this);
+				holder.addAttribute("baz", "qux");
+				holder.addAttribute("bix", "qox");
 				return message;
 			}
 		});
@@ -120,6 +125,49 @@ public class PseudoTransactionalMessageSourceTests {
 		TransactionSynchronizationManager.setActualTransactionActive(false);
 	}
 
+	@Test
+	public void testTransactionSynchronizationFactoryBean() {
+		ApplicationContext ctx = new AnnotationConfigApplicationContext(TestTxSyncConfiguration.class);
+
+		TransactionSynchronizationFactory syncFactory = ctx.getBean(TransactionSynchronizationFactory.class);
+
+		PollableChannel queueChannel = ctx.getBean("outputChannel", PollableChannel.class);
+
+		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
+
+		adapter.setTransactionSynchronizationFactory(syncFactory);
+
+		QueueChannel outputChannel = new QueueChannel();
+		adapter.setOutputChannel(outputChannel);
+		adapter.setSource(new MessageSource<String>() {
+
+			public Message<String> receive() {
+				GenericMessage<String> message = new GenericMessage<String>("foo");
+				IntegrationResourceHolder holder =
+						(IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this);
+				holder.addAttribute("baz", "qux");
+				holder.addAttribute("bix", "qox");
+				return message;
+			}
+		});
+
+		TransactionSynchronizationManager.initSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(true);
+		doPoll(adapter);
+		TransactionSynchronizationUtils.triggerBeforeCommit(false);
+		TransactionSynchronizationUtils.triggerAfterCommit();
+		Message<?> beforeCommitMessage = queueChannel.receive(1000);
+		assertNotNull(beforeCommitMessage);
+		assertEquals("qox", beforeCommitMessage.getPayload());
+		Message<?> afterCommitMessage = queueChannel.receive(1000);
+		assertNotNull(afterCommitMessage);
+		assertEquals("qux", afterCommitMessage.getPayload());
+		TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED);
+		TransactionSynchronizationManager.clearSynchronization();
+		TransactionSynchronizationManager.setActualTransactionActive(false);
+	}
+
+
 	@Test
 	public void testRollback() {
 		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
@@ -185,10 +233,10 @@ public class PseudoTransactionalMessageSourceTests {
 
 					public Message<String> receive() {
 						GenericMessage<String> message = new GenericMessage<String>("foo");
-						((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
-								.addAttribute("baz", "qux");
-						((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
-								.addAttribute("bix", "qox");
+						IntegrationResourceHolder holder =
+								(IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this);
+						holder.addAttribute("baz", "qux");
+						holder.addAttribute("bix", "qox");
 						return message;
 					}
 				});
@@ -314,7 +362,6 @@ public class PseudoTransactionalMessageSourceTests {
 		TransactionSynchronizationManager.setActualTransactionActive(false);
 	}
 
-	@Ignore
 	@Test
 	public void testInt2777CustomTransactionSynchronizationFactoryWithoutDealWithIntegrationResourceHolder() {
 		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
@@ -349,13 +396,6 @@ public class PseudoTransactionalMessageSourceTests {
 		TransactionSynchronizationManager.setActualTransactionActive(false);
 		assertEquals(1, txSyncCounter.get());
 
-		/*
-		TODO: Failed with 'java.lang.IllegalStateException: Already value
-		 [org.springframework.integration.transaction.IntegrationResourceHolder@46b8c8e6]
-		for key [org.springframework.integration.endpoint.PseudoTransactionalMessageSourceTests$8@78a1d1f4]
-		bound to thread [main]'
-		*/
-		//TODO: Need new JIRA issue to fix it
 		TransactionSynchronizationManager.initSynchronization();
 		TransactionSynchronizationManager.setActualTransactionActive(true);
 		doPoll(adapter);
@@ -383,5 +423,23 @@ public class PseudoTransactionalMessageSourceTests {
 		}
 
 	}
+	@Configuration
+	@EnableIntegration
+	public static class TestTxSyncConfiguration {
+
+		@Bean
+		public MessageChannel outputChannel() {
+			return new QueueChannel();
+		}
+
+		@Bean
+		public TransactionSynchronizationFactoryBean txSync() {
+			return new TransactionSynchronizationFactoryBean()
+					.beforeCommit("#bix")
+					.beforeCommit(outputChannel())
+					.afterCommit("#baz", outputChannel());
+		}
+
+	}
 
 }

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

@@ -37,10 +37,12 @@ import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.mail.event.MailIntegrationEvent;
 import org.springframework.integration.transaction.IntegrationResourceHolder;
+import org.springframework.integration.transaction.IntegrationResourceHolderSynchronization;
 import org.springframework.integration.transaction.TransactionSynchronizationFactory;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.Trigger;
 import org.springframework.scheduling.TriggerContext;
+import org.springframework.transaction.support.TransactionSynchronization;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -56,6 +58,7 @@ import org.springframework.util.CollectionUtils;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class ImapIdleChannelAdapter extends MessageProducerSupport implements BeanClassLoaderAware,
 		ApplicationEventPublisherAware {
@@ -118,7 +121,6 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
 	 * Specify whether the IDLE task should reconnect automatically after
 	 * catching a {@link FolderClosedException} while waiting for messages. The
 	 * default value is <code>true</code>.
-	 *
 	 * @param shouldReconnectAutomatically true to reconnect.
 	 */
 	public void setShouldReconnectAutomatically(boolean shouldReconnectAutomatically) {
@@ -250,18 +252,21 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
 
 				if (TransactionSynchronizationManager.isActualTransactionActive()) {
 					if (transactionSynchronizationFactory != null){
-						IntegrationResourceHolder holder = new IntegrationResourceHolder();
-						holder.setMessage(message);
-						TransactionSynchronizationManager.bindResource(ImapIdleChannelAdapter.this, holder);
-						TransactionSynchronizationManager.
-							registerSynchronization(transactionSynchronizationFactory.create(ImapIdleChannelAdapter.this));
+						TransactionSynchronization synchronization =
+								transactionSynchronizationFactory.create(ImapIdleChannelAdapter.this);
+						TransactionSynchronizationManager.registerSynchronization(synchronization);
+						if (synchronization instanceof IntegrationResourceHolderSynchronization) {
+							IntegrationResourceHolder holder =
+									((IntegrationResourceHolderSynchronization) synchronization).getResourceHolder();
+							holder.setMessage(message);
+						}
 					}
 				}
 				sendMessage(message);
 			}
 		};
 
-		// wrap in the TX proxy if neccessery
+		// wrap in the TX proxy if necessary
 		if (!CollectionUtils.isEmpty(adviceChain)) {
 			ProxyFactory proxyFactory = new ProxyFactory(sendingTask);
 			if (!CollectionUtils.isEmpty(adviceChain)) {