diff --git a/.project b/.project index a4c1e0d..ba707a1 100644 --- a/.project +++ b/.project @@ -26,8 +26,14 @@ + + org.springframework.ide.eclipse.core.springbuilder + + + + org.springframework.ide.eclipse.core.springnature org.eclipse.jem.workbench.JavaEMFNature org.eclipse.wst.common.modulecore.ModuleCoreNature org.eclipse.jdt.groovy.core.groovyNature diff --git a/.springBeans b/.springBeans index 5ab5ff5..14e9505 100644 --- a/.springBeans +++ b/.springBeans @@ -1,7 +1,7 @@ 1 - + @@ -11,6 +11,8 @@ src/test/resources/FlowClientNamespaceTest-context.xml src/test/resources/ref-bean-config.xml src/test/resources/META-INF/spring/integration/flows/subflow1/subflow1-context.xml + src/test/resources/TransactionalFlowTest-context.xml + src/test/resources/txmanager-config.xml diff --git a/pom.xml b/pom.xml index d34b9f9..1449578 100644 --- a/pom.xml +++ b/pom.xml @@ -49,6 +49,34 @@ ${spring.framework.version} test + + + org.springframework + spring-jms + ${spring.framework.version} + test + + + + org.springframework + spring-tx + ${spring.framework.version} + test + + + + org.springframework.integration + spring-integration-jms + ${spring.integration.version} + test + + + + org.springframework.integration + spring-integration-scripting + ${spring.integration.version} + test + org.springframework.integration diff --git a/src/main/java/org/springframework/integration/flow/FlowConstants.java b/src/main/java/org/springframework/integration/flow/FlowConstants.java new file mode 100644 index 0000000..3235eac --- /dev/null +++ b/src/main/java/org/springframework/integration/flow/FlowConstants.java @@ -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"; + +} diff --git a/src/main/java/org/springframework/integration/flow/config/FlowUtils.java b/src/main/java/org/springframework/integration/flow/config/FlowUtils.java index ad8d6e0..3e1cab8 100644 --- a/src/main/java/org/springframework/integration/flow/config/FlowUtils.java +++ b/src/main/java/org/springframework/integration/flow/config/FlowUtils.java @@ -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() {} /** diff --git a/src/main/java/org/springframework/integration/flow/handler/FlowMessageHandler.java b/src/main/java/org/springframework/integration/flow/handler/FlowMessageHandler.java index e08e785..b570343 100644 --- a/src/main/java/org/springframework/integration/flow/handler/FlowMessageHandler.java +++ b/src/main/java/org/springframework/integration/flow/handler/FlowMessageHandler.java @@ -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 flowConversationIdHeader = Collections.singletonMap(FlowUtils.FLOW_CONVERSATION_ID_HEADER, + Map 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; } } diff --git a/src/main/java/org/springframework/integration/flow/interceptor/FlowInterceptor.java b/src/main/java/org/springframework/integration/flow/interceptor/FlowInterceptor.java index 2657d7e..41fc2e5 100644 --- a/src/main/java/org/springframework/integration/flow/interceptor/FlowInterceptor.java +++ b/src/main/java/org/springframework/integration/flow/interceptor/FlowInterceptor.java @@ -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 headersToCopy = Collections.singletonMap(FlowUtils.FLOW_OUTPUT_PORT_HEADER, (Object) portName); + Map headersToCopy = Collections.singletonMap(FlowConstants.FLOW_OUTPUT_PORT_HEADER, (Object) portName); return MessageBuilder.fromMessage(message).copyHeadersIfAbsent(headersToCopy).build(); } diff --git a/src/test/java/org/springframework/integration/flow/Transaction/StubTransactionManager.java b/src/test/java/org/springframework/integration/flow/Transaction/StubTransactionManager.java new file mode 100644 index 0000000..d463b1e --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/Transaction/StubTransactionManager.java @@ -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; + + } + +} diff --git a/src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceActivator.java b/src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceActivator.java new file mode 100644 index 0000000..aa8fc5a --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceActivator.java @@ -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); + } + +} diff --git a/src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceInterface.java b/src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceInterface.java new file mode 100644 index 0000000..31d09b9 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceInterface.java @@ -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 { + +} diff --git a/src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTest.java b/src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTest.java new file mode 100644 index 0000000..63bf987 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTest.java @@ -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("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("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("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("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; + } + + } +} diff --git a/src/test/resources/META-INF/spring/integration/flows/transactional-flow/flow-context.xml b/src/test/resources/META-INF/spring/integration/flows/transactional-flow/flow-context.xml new file mode 100644 index 0000000..8264bc3 --- /dev/null +++ b/src/test/resources/META-INF/spring/integration/flows/transactional-flow/flow-context.xml @@ -0,0 +1,46 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/test/resources/TransactionalFlowTest-context.xml b/src/test/resources/TransactionalFlowTest-context.xml new file mode 100644 index 0000000..7f4c883 --- /dev/null +++ b/src/test/resources/TransactionalFlowTest-context.xml @@ -0,0 +1,17 @@ + + + + + + + + + + diff --git a/src/test/resources/log4j.xml b/src/test/resources/log4j.xml index 3294ced..75e0f19 100644 --- a/src/test/resources/log4j.xml +++ b/src/test/resources/log4j.xml @@ -16,7 +16,7 @@ - + diff --git a/src/test/resources/txmanager-config.xml b/src/test/resources/txmanager-config.xml new file mode 100644 index 0000000..c4a7b36 --- /dev/null +++ b/src/test/resources/txmanager-config.xml @@ -0,0 +1,7 @@ + + + + +