Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #675 from garyrussell/INT-2815

Browse Source 
# By Gary Russell
* garyrussell-INT-2815:
  INT-2815 Support Tx Synch In Pollable Consumers
This commit is contained in:
Gunnar Hillert
2012-11-14 01:45:52 -05:00
parent 5f2cc5a587 bcce3277e5
commit 5b0b0d731f
9 changed files with 273 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -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;

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

@@ -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<Boolean> pollingTask = new Callable<Boolean>() {
public Boolean call() throws Exception {
return doPoll();
}
};
List<Advice> 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));
}

23
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java
View File

@@ -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;
}
}

41
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -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;
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java
View File

@@ -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) {

2
spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolder.java
View File

@@ -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<String, Object> attributes = new HashMap<String, Object>();

50
spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests-context.xml Normal file
View File

@@ -0,0 +1,50 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:channel id="queueChannel">
<int:queue />
</int:channel>
<int:service-activator input-channel="queueChannel" ref="service" method="handle">
<int:poller fixed-delay="5000">
<int:transactional synchronization-factory="txSyncFactory"/>
</int:poller>
</int:service-activator>
<bean id="service" class="org.springframework.integration.channel.TransactionSynchronizationQueueChannelTests$Service"/>
<int:transaction-synchronization-factory id="txSyncFactory">
<int:after-commit channel="good" />
<int:after-rollback expression="'retry:' + payload" channel="queueChannel" />
</int:transaction-synchronization-factory>
<int:channel id="good">
<int:queue />
</int:channel>
<bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager" />
<int:channel id="queueChannel2">
<int:queue />
</int:channel>
<int:service-activator input-channel="queueChannel2" ref="service" method="handle">
<int:poller fixed-delay="5000">
<int:transactional synchronization-factory="txSyncFactory2"/>
</int:poller>
</int:service-activator>
<int:transaction-synchronization-factory id="txSyncFactory2">
<int:after-commit expression="payload + ' processed ok from ' + #inputChannel.componentName"
channel="good" />
</int:transaction-synchronization-factory>
<int:channel id="good">
<int:queue />
</int:channel>
</beans>

106
spring-integration-core/src/test/java/org/springframework/integration/channel/TransactionSynchronizationQueueChannelTests.java Normal file
View File

@@ -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<String> sentMessage = new GenericMessage<String>("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<String>("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<String> 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");
}
}
}
}

16
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java
View File

@@ -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;
}
}