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

Added transaction tests

Browse Source
This commit is contained in:
David Turanski
2011-08-27 19:22:22 -04:00
parent 90922b460a
commit 16951219a4
15 changed files with 411 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.project
View File

@@ -26,8 +26,14 @@
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.springframework.ide.eclipse.core.springbuilder</name>
<arguments>
</arguments>
</buildCommand>
</buildSpec>
<natures>
<nature>org.springframework.ide.eclipse.core.springnature</nature>
<nature>org.eclipse.jem.workbench.JavaEMFNature</nature>
<nature>org.eclipse.wst.common.modulecore.ModuleCoreNature</nature>
<nature>org.eclipse.jdt.groovy.core.groovyNature</nature>

4
.springBeans
View File

@@ -1,7 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.7.0.201107040453-RELEASE]]></pluginVersion>
<pluginVersion><![CDATA[2.8.0.201108100015-M1]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
@@ -11,6 +11,8 @@
<config>src/test/resources/FlowClientNamespaceTest-context.xml</config>
<config>src/test/resources/ref-bean-config.xml</config>
<config>src/test/resources/META-INF/spring/integration/flows/subflow1/subflow1-context.xml</config>
<config>src/test/resources/TransactionalFlowTest-context.xml</config>
<config>src/test/resources/txmanager-config.xml</config>
</configs>
<configSets>
</configSets>

28
pom.xml
View File

@@ -49,6 +49,34 @@
<version>${spring.framework.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${spring.framework.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-tx</artifactId>
<version>${spring.framework.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jms</artifactId>
<version>${spring.integration.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-scripting</artifactId>
<version>${spring.integration.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>

36
src/main/java/org/springframework/integration/flow/FlowConstants.java Normal file
View File

@@ -0,0 +1,36 @@
/*
* Copyright 2002-2011 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.flow;
/**
* @author David Turanski
*
*/
public interface FlowConstants {
/**
* Message header indicating which port produced the flow output
*/
public static final String FLOW_OUTPUT_PORT_HEADER = "flow.output.port";
/**
* Message header used to correlate port input and output messages
*/
public static final String FLOW_CONVERSATION_ID_HEADER = "flow.conversation.id";
/**
* FLOW_OUTPUT_PORT_HEADER value if FlowHandler catches an exception
*/
public static final String FLOW_HANDLER_EXCEPTION_HEADER_VALUE = "flowhandler.exception";
}

10
src/main/java/org/springframework/integration/flow/config/FlowUtils.java
View File

@@ -35,16 +35,6 @@ import org.springframework.util.ResourceUtils;
*/
public class FlowUtils {
/**
* Message header indicating which port produced the flow output
*/
public static final String FLOW_OUTPUT_PORT_HEADER = "flow.output.port";
/**
* Message header used to correlate port input and output messages
*/
public static final String FLOW_CONVERSATION_ID_HEADER = "flow.conversation.id";
private FlowUtils() {}
/**

12
src/main/java/org/springframework/integration/flow/handler/FlowMessageHandler.java
View File

@@ -26,6 +26,7 @@ import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.flow.FlowConstants;
import org.springframework.integration.flow.config.FlowUtils;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.message.ErrorMessage;
@@ -77,7 +78,7 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
protected Object handleRequestMessage(Message<?> requestMessage) {
UUID conversationId = requestMessage.getHeaders().getId();
Map<String, Object> flowConversationIdHeader = Collections.singletonMap(FlowUtils.FLOW_CONVERSATION_ID_HEADER,
Map<String, Object> flowConversationIdHeader = Collections.singletonMap(FlowConstants.FLOW_CONVERSATION_ID_HEADER,
(Object) conversationId);
Message<?> message = MessageBuilder.fromMessage(requestMessage).copyHeaders(flowConversationIdHeader)
@@ -95,8 +96,9 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
}
catch (MessagingException me) {
log.error(me.getMessage(), me);
if (conversationId.equals(me.getFailedMessage().getHeaders().get(FlowUtils.FLOW_CONVERSATION_ID_HEADER))) {
return new ErrorMessage(me);
if (conversationId.equals(me.getFailedMessage().getHeaders().get(FlowConstants.FLOW_CONVERSATION_ID_HEADER))) {
return new ErrorMessage(me,Collections.singletonMap(FlowConstants.FLOW_OUTPUT_PORT_HEADER,
(Object)FlowConstants.FLOW_HANDLER_EXCEPTION_HEADER_VALUE));
}
}
return null;
@@ -122,12 +124,12 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (conversationId.equals(message.getHeaders().get(FlowUtils.FLOW_CONVERSATION_ID_HEADER))) {
if (conversationId.equals(message.getHeaders().get(FlowConstants.FLOW_CONVERSATION_ID_HEADER))) {
this.response = message;
} else {
if (message instanceof ErrorMessage){
MessagingException me = (MessagingException) message.getPayload();
if (conversationId.equals(me.getFailedMessage().getHeaders().get(FlowUtils.FLOW_CONVERSATION_ID_HEADER))) {
if (conversationId.equals(me.getFailedMessage().getHeaders().get(FlowConstants.FLOW_CONVERSATION_ID_HEADER))) {
this.response = message;
}
}

3
src/main/java/org/springframework/integration/flow/interceptor/FlowInterceptor.java
View File

@@ -23,6 +23,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
import org.springframework.integration.flow.FlowConstants;
import org.springframework.integration.flow.config.FlowUtils;
import org.springframework.integration.support.MessageBuilder;
@@ -50,7 +51,7 @@ public class FlowInterceptor extends ChannelInterceptorAdapter {
log.debug("flow interceptor " + this.hashCode() + " received a message from port " + portName + " on channel "
+ channel);
Map<String, Object> headersToCopy = Collections.singletonMap(FlowUtils.FLOW_OUTPUT_PORT_HEADER, (Object) portName);
Map<String, Object> headersToCopy = Collections.singletonMap(FlowConstants.FLOW_OUTPUT_PORT_HEADER, (Object) portName);
return MessageBuilder.fromMessage(message).copyHeadersIfAbsent(headersToCopy).build();
}

71
src/test/java/org/springframework/integration/flow/Transaction/StubTransactionManager.java Normal file
View File

@@ -0,0 +1,71 @@
/*
* Copyright 2002-2011 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.flow.Transaction;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.support.AbstractPlatformTransactionManager;
import org.springframework.transaction.support.DefaultTransactionStatus;
/**
* @author David Turanski
*
*/
@SuppressWarnings("serial")
public class StubTransactionManager extends AbstractPlatformTransactionManager {
private static Log logger = LogFactory.getLog(StubTransactionManager.class);
public boolean rolledback;
public boolean committed;
private String transaction="stub-transaction";
/* (non-Javadoc)
* @see org.springframework.transaction.support.AbstractPlatformTransactionManager#doBegin(java.lang.Object, org.springframework.transaction.TransactionDefinition)
*/
@Override
protected void doBegin(Object transaction, TransactionDefinition txDef) throws TransactionException {
logger.debug("begining transaction:" + transaction +" def " + txDef);
rolledback = false;
committed = false;
}
/* (non-Javadoc)
* @see org.springframework.transaction.support.AbstractPlatformTransactionManager#doCommit(org.springframework.transaction.support.DefaultTransactionStatus)
*/
@Override
protected void doCommit(DefaultTransactionStatus arg0) throws TransactionException {
logger.debug("committing transaction");
committed = true;
}
/* (non-Javadoc)
* @see org.springframework.transaction.support.AbstractPlatformTransactionManager#doGetTransaction()
*/
@Override
protected Object doGetTransaction() throws TransactionException {
logger.debug("get transaction:" + this.transaction);
return this.transaction;
}
/* (non-Javadoc)
* @see org.springframework.transaction.support.AbstractPlatformTransactionManager#doRollback(org.springframework.transaction.support.DefaultTransactionStatus)
*/
@Override
protected void doRollback(DefaultTransactionStatus arg0) throws TransactionException {
logger.debug("rolling back transaction");
rolledback = true;
}
}

38
src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceActivator.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* Copyright 2002-2011 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.flow.Transaction;
import org.springframework.integration.Message;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.transaction.annotation.Transactional;
/**
* @author David Turanski
*
*/
@Transactional
public class TransactionalServiceActivator implements RequestReplyExchanger {
private RequestReplyExchanger gateway;
public TransactionalServiceActivator(RequestReplyExchanger gateway) {
this.gateway = gateway;
}
/* (non-Javadoc)
* @see org.springframework.integration.gateway.RequestReplyExchanger#exchange(org.springframework.integration.Message)
*/
@Override
public Message<?> exchange(Message<?> request) {
return gateway.exchange(request);
}
}

25
src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceInterface.java Normal file
View File

@@ -0,0 +1,25 @@
/*
* Copyright 2002-2011 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.flow.Transaction;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.transaction.annotation.Transactional;
/**
* @author David Turanski
*
*/
@Transactional
public interface TransactionalServiceInterface extends RequestReplyExchanger {
}

124
src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTest.java Normal file
View File

@@ -0,0 +1,124 @@
/*
* Copyright 2002-2011 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.flow.config.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.flow.FlowConstants;
import org.springframework.integration.flow.Transaction.StubTransactionManager;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
/**
* @author David Turanski
*
*/
public class TransactionalFlowTest {
@Test
public void testFlowDirectCommit() {
ApplicationContext applicationContext =
new ClassPathXmlApplicationContext("/META-INF/spring/integration/flows/transactional-flow/flow-context.xml"
,"/txmanager-config.xml");
MessageChannel inputChannel = applicationContext.getBean("inputChannel", MessageChannel.class);
SubscribableChannel outputChannel = applicationContext.getBean("outputChannel", SubscribableChannel.class);
StubTransactionManager transactionManager = applicationContext.getBean(StubTransactionManager.class);
Handler handler = new Handler();
outputChannel.subscribe(handler);
inputChannel.send(new GenericMessage<String>("hello"));
assertTrue(handler.messageReceived);
assertTrue(transactionManager.committed);
assertFalse(transactionManager.rolledback);
}
@Test
public void testFlowDirectRollback() {
ApplicationContext applicationContext =
new ClassPathXmlApplicationContext("/META-INF/spring/integration/flows/transactional-flow/flow-context.xml",
"/txmanager-config.xml");
MessageChannel inputChannel = applicationContext.getBean("inputChannel", MessageChannel.class);
SubscribableChannel outputChannel = applicationContext.getBean("outputChannel", SubscribableChannel.class);
StubTransactionManager transactionManager = applicationContext.getBean(StubTransactionManager.class);
Handler handler = new Handler();
outputChannel.subscribe(handler);
try {
inputChannel.send(new GenericMessage<String>("rollback"));
fail("should throw exception");
} catch (Exception e) {
assertFalse(handler.messageReceived);
assertTrue(transactionManager.rolledback);
assertFalse(transactionManager.committed);
}
}
@Test
public void testFlowCommit() {
ApplicationContext applicationContext =
new ClassPathXmlApplicationContext("/TransactionalFlowTest-context.xml","/txmanager-config.xml");
MessageChannel inputChannel = applicationContext.getBean("inputC", MessageChannel.class);
SubscribableChannel outputChannel = applicationContext.getBean("outputC", SubscribableChannel.class);
StubTransactionManager transactionManager = applicationContext.getBean(StubTransactionManager.class);
Handler handler = new Handler();
outputChannel.subscribe(handler);
inputChannel.send(new GenericMessage<String>("hello"));
assertTrue(handler.messageReceived);
assertTrue(transactionManager.committed);
assertFalse(transactionManager.rolledback);
}
@Test
public void testFlowRollback() {
ApplicationContext applicationContext =
new ClassPathXmlApplicationContext("/TransactionalFlowTest-context.xml","/txmanager-config.xml");
MessageChannel inputChannel = applicationContext.getBean("inputC", MessageChannel.class);
SubscribableChannel outputChannel = applicationContext.getBean("outputC", SubscribableChannel.class);
StubTransactionManager transactionManager = applicationContext.getBean(StubTransactionManager.class);
Handler handler = new Handler();
outputChannel.subscribe(handler);
inputChannel.send(new GenericMessage<String>("rollback"));
assertTrue(handler.messageReceived);
assertTrue(handler.message instanceof ErrorMessage);
assertEquals(FlowConstants.FLOW_HANDLER_EXCEPTION_HEADER_VALUE,
handler.message.getHeaders().get(FlowConstants.FLOW_OUTPUT_PORT_HEADER));
assertTrue(transactionManager.rolledback);
assertFalse(transactionManager.committed);
}
private static class Handler implements MessageHandler {
public boolean messageReceived;
public Message<?> message;
/* (non-Javadoc)
* @see org.springframework.integration.core.MessageHandler#handleMessage(org.springframework.integration.Message)
*/
@Override
public void handleMessage(Message<?> message) throws MessagingException {
this.messageReceived = true;
this.message = message;
}
}
}

46
src/test/resources/META-INF/spring/integration/flows/transactional-flow/flow-context.xml Normal file
View File

@@ -0,0 +1,46 @@
<?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:tx="http://www.springframework.org/schema/tx"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-flow="http://www.springframework.org/schema/integration/flow"
xmlns:int-script="http://www.springframework.org/schema/integration/script"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.1.xsd
http://www.springframework.org/schema/integration/flow http://www.springframework.org/schema/integration/flow/spring-integration-flow-2.1.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/script http://www.springframework.org/schema/integration/script-2.1.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.1.xsd">
<int-flow:flow-configuration>
<int-flow:port-mapping input-channel="inputChannel" output-channel="outputChannel"/>
</int-flow:flow-configuration>
<tx:annotation-driven/>
<!-- Invoke the gateway via a transactional method -->
<int:service-activator id="serviceActivator"
input-channel="inputChannel"
output-channel="outputChannel">
<bean class="org.springframework.integration.flow.Transaction.TransactionalServiceActivator">
<constructor-arg ref="gateway"/>
</bean>
</int:service-activator>
<int:gateway id="gateway"
default-request-channel="requestChannel"
default-reply-channel="replyChannel"/>
<int:service-activator input-channel="requestChannel" output-channel="replyChannel">
<int-script:script lang="groovy"><![CDATA[
if (payload.equals("rollback")) {
throw new RuntimeException("received rollback command")
}
return payload
]]>
</int-script:script>
</int:service-activator>
<int:channel id="replyChannel"/>
<int:channel id="outputChannel"/>
</beans>

17
src/test/resources/TransactionalFlowTest-context.xml Normal file
View File

@@ -0,0 +1,17 @@
<?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-flow="http://www.springframework.org/schema/integration/flow"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.1.xsd
http://www.springframework.org/schema/integration/flow http://www.springframework.org/schema/integration/flow/spring-integration-flow-2.1.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int-flow:flow id="transactional-flow"/>
<int-flow:outbound-gateway flow="transactional-flow"
input-channel="inputC" output-channel="outputC"/>
<int:channel id="outputC"/>
</beans>

2
src/test/resources/log4j.xml
View File

@@ -16,7 +16,7 @@
</logger>
<logger name="org.springframework.integration">
<level value="info" />
<level value="debug" />
</logger>
<!-- Root Logger -->

7
src/test/resources/txmanager-config.xml Normal file
View File

@@ -0,0 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="transactionManager" class="org.springframework.integration.flow.Transaction.StubTransactionManager"/>
</beans>