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 da36e155ee..ff4ef1ebe6 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 @@ -309,19 +309,33 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement 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 (key != null) { - resourceHolder.addAttribute(key, resource); + 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()); + } } - return resourceHolder; + } + + Object resourceHolder = TransactionSynchronizationManager.getResource(resource); + if (resourceHolder instanceof IntegrationResourceHolder) { + IntegrationResourceHolder integrationResourceHolder = (IntegrationResourceHolder) resourceHolder; + if (key != null) { + integrationResourceHolder.addAttribute(key, resource); + } + return integrationResourceHolder; } } + return null; } 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 df971df0ab..2c6b9f24ff 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-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.Log; 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 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Oleg Zhurakousky * @author Artem Bilan + * * @since 2.2 */ public class DefaultTransactionSynchronizationFactory implements TransactionSynchronizationFactory { @@ -46,13 +47,9 @@ public class DefaultTransactionSynchronizationFactory implements TransactionSync @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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/PassThroughTransactionSynchronizationFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/PassThroughTransactionSynchronizationFactory.java index fc61b784e7..6000c54e87 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/PassThroughTransactionSynchronizationFactory.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/PassThroughTransactionSynchronizationFactory.java @@ -17,20 +17,16 @@ 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 { @@ -38,10 +34,7 @@ public class PassThroughTransactionSynchronizationFactory implements Transaction @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); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests-context.xml index 7765d1791d..96fbca9261 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests-context.xml @@ -10,7 +10,7 @@ - + @@ -33,7 +33,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests.java index 843e39c3d8..49c97ef4b5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests.java +++ b/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.Message; 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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java index eff6151ca7..ba9da19024 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java +++ b/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.aopalliance.intercept.MethodInvocation; 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 @@ import org.springframework.transaction.support.DefaultTransactionStatus; * @author Oleg Zhurakousky * @author Gary Russell * @author Andreas Baer + * @author Artem Bilan */ public class PollingTransactionTests { @@ -82,25 +85,23 @@ public class PollingTransactionTests { PollingConsumer advicedPoller = context.getBean("advicedSa", PollingConsumer.class); List 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("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 class PollingTransactionTests { 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 class PollingTransactionTests { } + 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 { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/transactionTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/transactionTests.xml index 74baa454a1..d0ec5a9b68 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/transactionTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/transactionTests.xml @@ -24,25 +24,26 @@ - + - + - + + 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 c6b6538836..6a710495da 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 @@ -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 class ImapIdleChannelAdapter extends MessageProducerSupport implements Be @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 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be @SuppressWarnings("unchecked") org.springframework.messaging.Message message = mailMessage instanceof Message - ? ImapIdleChannelAdapter.this.getMessageBuilderFactory().withPayload(mailMessage).build() - : (org.springframework.messaging.Message) mailMessage; + ? ImapIdleChannelAdapter.this.getMessageBuilderFactory().withPayload(mailMessage).build() + : (org.springframework.messaging.Message) 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 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be @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