diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 5284ec6cb3..e1e80c4620 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -21,7 +21,6 @@ import java.util.List; import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.aop.framework.Advised; import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.AopUtils; @@ -229,6 +228,7 @@ public class ConsumerEndpointFactoryBean pollingConsumer.setErrorHandler(this.pollerMetadata.getErrorHandler()); pollingConsumer.setReceiveTimeout(this.pollerMetadata.getReceiveTimeout()); + pollingConsumer.setTransactionSynchronizationFactory(this.pollerMetadata.getTransactionSynchronizationFactory()); pollingConsumer.setBeanClassLoader(beanClassLoader); pollingConsumer.setBeanFactory(beanFactory); this.endpoint = pollingConsumer; 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 49c9819dcd..2160cd6936 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -22,19 +22,22 @@ import java.util.concurrent.Executor; import java.util.concurrent.ScheduledFuture; import org.aopalliance.aop.Advice; - import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.integration.Message; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessagingException; import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; +import org.springframework.integration.transaction.IntegrationResourceHolder; +import org.springframework.integration.transaction.TransactionSynchronizationFactory; import org.springframework.integration.util.ErrorHandlingTaskExecutor; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.support.PeriodicTrigger; +import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; @@ -43,11 +46,12 @@ import org.springframework.util.ErrorHandler; /** * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implements BeanClassLoaderAware { private volatile Executor taskExecutor = new SyncTaskExecutor(); - + private volatile ErrorHandler errorHandler; private volatile Trigger trigger = new PeriodicTrigger(10); @@ -61,16 +65,17 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private volatile Runnable poller; private volatile boolean initialized; - + private volatile long maxMessagesPerPoll = -1; - + private final Object initializationMonitor = new Object(); + private volatile TransactionSynchronizationFactory transactionSynchronizationFactory; public AbstractPollingEndpoint() { this.setPhase(Integer.MAX_VALUE); } - + /** * @deprecated As of release 2.0.2, use individual setters */ @@ -82,7 +87,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.setTaskExecutor(pollerMetadata.getTaskExecutor()); this.setTrigger(pollerMetadata.getTrigger()); } - + public void setTaskExecutor(Executor taskExecutor) { this.taskExecutor = (taskExecutor != null ? taskExecutor : new SyncTaskExecutor()); } @@ -107,6 +112,11 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.beanClassLoader = classLoader; } + public void setTransactionSynchronizationFactory( + TransactionSynchronizationFactory transactionSynchronizationFactory) { + this.transactionSynchronizationFactory = transactionSynchronizationFactory; + } + @Override protected void onInit() { synchronized (this.initializationMonitor) { @@ -119,7 +129,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.taskExecutor = providedExecutor; } if (this.taskExecutor != null) { - if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) { + if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) { if (this.errorHandler == null) { Assert.notNull(this.getBeanFactory(), "BeanFactory is required"); this.errorHandler = new MessagePublishingErrorHandler( @@ -131,7 +141,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement try { this.poller = this.createPoller(); this.initialized = true; - } + } catch (Exception e) { throw new MessagingException("Failed to create Poller", e); } @@ -140,13 +150,13 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement @SuppressWarnings("unchecked") private Runnable createPoller() throws Exception { - + Callable pollingTask = new Callable() { public Boolean call() throws Exception { return doPoll(); } }; - + List adviceChain = this.adviceChain; if (!CollectionUtils.isEmpty(adviceChain)) { ProxyFactory proxyFactory = new ProxyFactory(pollingTask); @@ -160,7 +170,20 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement return new Poller(pollingTask); } - + /** + * Synchronize with an existing transaction (if any) and receive + * a message using {@link #doReceive()}. + * @return The message (or null). + */ + protected final Message syncIfTxAndReceive() { + IntegrationResourceHolder holder = bindResourceHolderIfNecessary( + this.getResourceKey(), this.getResourceToBind()); + Message message = this.doReceive(); + if (holder != null && message != null) { + holder.setMessage(message); + } + return message; + } // LifecycleSupport implementation @Override // guarded by super#lifecycleLock @@ -182,9 +205,29 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.initialized = false; } - + private IntegrationResourceHolder bindResourceHolderIfNecessary(String key, Object resource) { + IntegrationResourceHolder holder = null; + + if (this.transactionSynchronizationFactory != null) { + if (TransactionSynchronizationManager.isActualTransactionActive()) { + holder = new IntegrationResourceHolder(); + if (key != null) { + holder.addAttribute(key, resource); + } + TransactionSynchronizationManager.bindResource(resource, holder); + TransactionSynchronizationManager.registerSynchronization(this.transactionSynchronizationFactory.create(resource)); + } + } + return holder; + } + protected abstract boolean doPoll(); + protected abstract Message doReceive(); + + protected abstract Object getResourceToBind(); + + protected abstract String getResourceKey(); /** * Default Poller implementation @@ -208,11 +251,11 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement break; } count++; - } + } catch (Exception e) { if (e instanceof RuntimeException) { throw (RuntimeException) e; - } + } else { throw new MessageHandlingException(new ErrorMessage(e)); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index d7d555544e..535fb2183a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java @@ -22,6 +22,7 @@ import org.springframework.context.Lifecycle; import org.springframework.integration.Message; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.transaction.IntegrationResourceHolder; import org.springframework.util.Assert; /** @@ -74,9 +75,7 @@ public class PollingConsumer extends AbstractPollingEndpoint { @Override protected boolean doPoll() { - Message message = (this.receiveTimeout >= 0) - ? this.inputChannel.receive(this.receiveTimeout) - : this.inputChannel.receive(); + Message message = this.syncIfTxAndReceive(); if (this.logger.isDebugEnabled()){ this.logger.debug("Poll resulted in Message: " + message); } @@ -89,4 +88,22 @@ public class PollingConsumer extends AbstractPollingEndpoint { this.handler.handleMessage(message); return true; } + + @Override + protected Message doReceive() { + Message message = (this.receiveTimeout >= 0) + ? this.inputChannel.receive(this.receiveTimeout) + : this.inputChannel.receive(); + return message; + } + + @Override + protected Object getResourceToBind() { + return this.inputChannel; + } + + @Override + protected String getResourceKey() { + return IntegrationResourceHolder.INPUT_CHANNEL; + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 83d3dd55a2..7812c221d1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -25,8 +25,6 @@ import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.history.TrackableComponent; import org.springframework.integration.transaction.IntegrationResourceHolder; -import org.springframework.integration.transaction.TransactionSynchronizationFactory; -import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.util.Assert; /** @@ -47,13 +45,6 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme private final MessagingTemplate messagingTemplate = new MessagingTemplate(); - private volatile TransactionSynchronizationFactory transactionSynchronizationFactory; - - public void setTransactionSynchronizationFactory( - TransactionSynchronizationFactory transactionSynchronizationFactory) { - this.transactionSynchronizationFactory = transactionSynchronizationFactory; - } - /** * Specify the source to be polled for Messages. */ @@ -99,26 +90,12 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme @Override protected boolean doPoll() { - Message message; - IntegrationResourceHolder holder = null; - - if (TransactionSynchronizationManager.isActualTransactionActive()) { - if (transactionSynchronizationFactory != null){ - holder = new IntegrationResourceHolder(); - holder.addAttribute(IntegrationResourceHolder.MESSAGE_SOURCE, source); - TransactionSynchronizationManager.bindResource(source, holder); - TransactionSynchronizationManager.registerSynchronization(transactionSynchronizationFactory.create(source)); - } - } - message = this.source.receive(); + Message message = this.syncIfTxAndReceive(); if (this.logger.isDebugEnabled()){ this.logger.debug("Poll resulted in Message: " + message); } if (message != null) { - if (holder != null) { - holder.setMessage(message); - } if (this.shouldTrack) { message = MessageHistory.write(message, this); } @@ -140,4 +117,20 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme } return false; } + + @Override + protected Message doReceive() { + return this.source.receive(); + } + + @Override + protected Object getResourceToBind() { + return this.source; + } + + @Override + protected String getResourceKey() { + return IntegrationResourceHolder.MESSAGE_SOURCE; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java index a0ec5bafb7..5c358401a6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java @@ -115,7 +115,9 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int "as part of '" + expressionType + "' transaction synchronization"); } try { - spelResultMessage = MessageBuilder.withPayload(value).build(); + spelResultMessage = MessageBuilder.withPayload(value) + .copyHeaders(message.getHeaders()) + .build(); this.sendMessage(messageChannel, spelResultMessage); } catch (Exception e) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolder.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolder.java index 989ce07534..8cdc0b949b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolder.java @@ -32,6 +32,8 @@ public class IntegrationResourceHolder implements ResourceHolder { public static final String MESSAGE_SOURCE = "messageSource"; + public static final String INPUT_CHANNEL = "inputChannel"; + private volatile Message message; private final Map attributes = new HashMap(); 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 new file mode 100644 index 0000000000..7765d1791d --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests-context.xml @@ -0,0 +1,50 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..f077fa1d8e --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests.java @@ -0,0 +1,106 @@ +/* + * Copyright 2002-2012 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.channel; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.Message; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 2.2 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class TransactionSynchronizationQueueChannelTests { + + @Autowired + private PollableChannel queueChannel; + + @Autowired + private PollableChannel good; + + @Autowired + private Service service; + + @Autowired + private PollableChannel queueChannel2; + + @Test + public void testCommit() throws Exception { + service.latch = new CountDownLatch(1); + GenericMessage sentMessage = new GenericMessage("hello"); + queueChannel.send(sentMessage); + assertTrue(service.latch.await(10, TimeUnit.SECONDS)); + Message message = good.receive(1000); + assertNotNull(message); + assertEquals("hello", message.getPayload()); + assertSame(message, sentMessage); + } + + @Test + public void testRollback() throws Exception { + service.latch = new CountDownLatch(1); + queueChannel.send(new GenericMessage("fail")); + assertTrue(service.latch.await(10, TimeUnit.SECONDS)); + Message message = queueChannel.receive(1000); + assertNotNull(message); + assertEquals("retry:fail", message.getPayload()); + assertNull(good.receive(0)); + } + + @Test + public void testIncludeChannelName() throws Exception { + service.latch = new CountDownLatch(1); + Message sentMessage = MessageBuilder.withPayload("hello") + .setHeader("foo", "bar").build(); + queueChannel2.send(sentMessage); + assertTrue(service.latch.await(10, TimeUnit.SECONDS)); + Message message = good.receive(1000); + assertNotNull(message); + assertEquals("hello processed ok from queueChannel2", message.getPayload()); + assertNotNull(message.getHeaders().get("foo")); + assertEquals("bar", message.getHeaders().get("foo")); + } + + public static class Service { + private CountDownLatch latch; + + public void handle(String foo) { + latch.countDown(); + if (foo.startsWith("fail")) { + throw new RuntimeException("planned failure"); + } + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java index 831c9b9a5c..97c1ef5eb5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java @@ -16,6 +16,7 @@ package org.springframework.integration.endpoint; +import org.springframework.integration.Message; import org.springframework.scheduling.support.PeriodicTrigger; /** @@ -32,4 +33,19 @@ public class PollingEndpointStub extends AbstractPollingEndpoint { throw new RuntimeException("intentional test failure"); } + @Override + protected Message doReceive() { + return null; + } + + @Override + protected Object getResourceToBind() { + return null; + } + + @Override + protected String getResourceKey() { + return null; + } + }