From 39ef91ff843ad3aad4dc4a564344c6e6c55d08ff Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 16 May 2014 17:30:25 +0300 Subject: [PATCH] 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` --- .../endpoint/AbstractPollingEndpoint.java | 21 +- ...aultTransactionSynchronizationFactory.java | 22 +- ...egrationResourceHolderSynchronization.java | 42 ++++ ...TransactionSynchronizationFactoryBean.java | 198 ++++++++++++++++++ ...PseudoTransactionalMessageSourceTests.java | 92 ++++++-- .../mail/ImapIdleChannelAdapter.java | 19 +- 6 files changed, 350 insertions(+), 44 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolderSynchronization.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationFactoryBean.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 626cf131e1..ea383b8de5 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 @@ -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 } }); } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java index c121e15135..fe10f43163 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/DefaultTransactionSynchronizationFactory.java +++ b/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 { + 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); } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolderSynchronization.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolderSynchronization.java new file mode 100644 index 0000000000..fcfa4fb964 --- /dev/null +++ b/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 { + + protected final IntegrationResourceHolder resourceHolder; + + public IntegrationResourceHolderSynchronization(IntegrationResourceHolder resourceHolder, + Object resourceKey) { + super(resourceHolder, resourceKey); + this.resourceHolder = resourceHolder; + } + + public IntegrationResourceHolder getResourceHolder() { + return resourceHolder; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationFactoryBean.java new file mode 100644 index 0000000000..4d7451ef7b --- /dev/null +++ b/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, + 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; + } + +} 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 21f447859a..10d89539c0 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 @@ -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 receive() { GenericMessage message = new GenericMessage("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() { + + public Message receive() { + GenericMessage message = new GenericMessage("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 receive() { GenericMessage message = new GenericMessage("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()); + } + + } } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java index 011fa29a66..a9e367f230 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java +++ b/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 true. - * * @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)) {