diff --git a/pom.xml b/pom.xml index 4a1c6c4..255afc5 100644 --- a/pom.xml +++ b/pom.xml @@ -1,212 +1,187 @@ - - - - 4.0.0 - org.springframework.integration - spring-integration-flow - 2.1.0.M1 - Spring Integration Flow Support - - - The Apache Software License, Version 2.0 - http://www.apache.org/licenses/LICENSE-2.0.txt - repo - - - - - UTF8 - 3.1.0.RC1 - 2.1.0.M2 - true - - - - - - - log4j - log4j - 1.2.15 - test - - - - org.springframework.integration - spring-integration-test - ${spring.integration.version} - test - - - - org.springframework - spring-test - ${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-scripting - ${spring.integration.version} - test - - - - org.springframework.integration - spring-integration-core - ${spring.integration.version} - - - org.springframework.integration - spring-integration-jmx - ${spring.integration.version} - test - - - - org.springframework - spring-context-support - ${spring.framework.version} - compile - - - - org.springframework - spring-context - ${spring.framework.version} - compile - - - - org.springframework - spring-core - ${spring.framework.version} - compile - - - org.springframework - spring-asm - ${spring.framework.version} - compile - - - - org.springframework - spring-expression - ${spring.framework.version} - compile - - - - commons-lang - commons-lang - 2.6 - - - - org.codehaus.groovy - groovy-all - 1.8.0 - - - - junit - junit - 4.8.2 - test - - - - org.springframework.integration - spring-integration-xml - ${spring.integration.version} - test - - - - org.springframework - spring-oxm - ${spring.framework.version} - test - - - - xmlunit - xmlunit - 1.3 - test - - - - - - - - - - - - org.apache.maven.plugins - maven-compiler-plugin - 2.3.2 - - 1.6 - 1.6 - - - - - maven-antrun-plugin - - - test-compile - test-compile - - - - - - - - - - - - - - - - - - - maven_springframework_milestone - http://maven.springframework.org/milestone - - - maven_springframework_snapshot - http://maven.springframework.org/snapshot - - + + + 4.0.0 + org.springframework.integration + spring-integration-flow + 2.1.0.M1 + Spring Integration Flow Support + + 3.1.0.RC1 + 2.1.0.M2 + UTF8 + + + + The Apache Software License, Version 2.0 + http://www.apache.org/licenses/LICENSE-2.0.txt + repo + + + + + + src/main/java + + **/* + + + **/*.java + + + + src/main/resources + + **/* + + + + + + src/test/java + + **/* + + + **/*.java + + + + src/test/resources + + **/* + + + + + + maven-compiler-plugin + + 1.6 + 1.6 + + + + maven-surefire-plugin + + + **/*Tests.java + + + **/*Abstract*.java + + + + + + + + SpringSource External Bundle Repository + http://repository.springsource.com/maven/bundles/external/ + + + SpringSource Milestone Repository + http://maven.springframework.org/milestone/ + + + SpringSource Release Bundle Repository + http://repository.springsource.com/maven/bundles/release/ + + + SpringSource Release Repository + http://maven.springframework.org/release/ + + + + + org.springframework + spring-test + ${spring.version} + test + + + org.springframework + spring-tx + ${spring.version} + test + + + org.springframework.integration + spring-integration-test + ${spring.integration.version} + test + + + cglib + cglib-nodep + 2.2 + test + + + log4j + log4j + 1.2.12 + test + + + org.springframework.integration + spring-integration-core + ${spring.integration.version} + compile + + + org.easymock + easymock + 2.3 + test + + + org.easymock + easymockclassextension + 2.3 + test + + + org.springframework.integration + spring-integration-groovy + ${spring.integration.version} + test + + + org.springframework + spring-context + ${spring.version} + compile + + + org.hamcrest + hamcrest-all + 1.1 + test + + + org.mockito + mockito-all + 1.8.4 + test + + + commons-lang + commons-lang + 2.6 + compile + + + org.springframework.integration + spring-integration-jmx + ${spring.integration.version} + test + + + junit + junit-dep + 4.8.2 + test + + diff --git a/src/main/java/org/springframework/integration/flow/ChannelNamePortConfiguration.java b/src/main/java/org/springframework/integration/flow/ChannelNamePortConfiguration.java index 8bedb57..8548796 100644 --- a/src/main/java/org/springframework/integration/flow/ChannelNamePortConfiguration.java +++ b/src/main/java/org/springframework/integration/flow/ChannelNamePortConfiguration.java @@ -21,81 +21,81 @@ package org.springframework.integration.flow; import java.util.ArrayList; import java.util.Collections; import java.util.List; + /** * Container for {@link Flow} {@link PortConfiguration} metadata * @author David Turanski - * + * */ public class ChannelNamePortConfiguration implements PortConfiguration { - private PortMetadata inputPortMetadata; - private List outputPortMetadataList; - - /** - * Generic configuration - * @param inputPortMetadata metadata defining the input message port - * @param outputPortMetadataList a list of metadata defining multiple output message ports - */ - public ChannelNamePortConfiguration(PortMetadata inputPortMetadata, List outputPortMetadataList) { - this.outputPortMetadataList = outputPortMetadataList; - this.inputPortMetadata = inputPortMetadata; - } + private PortMetadata inputPortMetadata; - /** - * A simple configuration for a flow with one input and one output port - * @param inputChannelName - * @param outputChannelName - */ - public ChannelNamePortConfiguration(String inputChannelName, String outputChannelName) { - this.inputPortMetadata = new PortMetadata("input", inputChannelName); + private List outputPortMetadataList; - if (outputChannelName != null) { - PortMetadata outputPortMetadata = new PortMetadata("output", outputChannelName); - this.outputPortMetadataList = Collections.singletonList(outputPortMetadata); - } else { - // this.outputPortMetadataList = new ArrayList(); - } - } + /** + * Generic configuration + * @param inputPortMetadata metadata defining the input message port + * @param outputPortMetadataList a list of metadata defining multiple output + * message ports + */ + public ChannelNamePortConfiguration(PortMetadata inputPortMetadata, List outputPortMetadataList) { + this.outputPortMetadataList = outputPortMetadataList; + this.inputPortMetadata = inputPortMetadata; + } - @Override - public String getInputPortName() { - return this.inputPortMetadata.getPortName(); - } + /** + * A simple configuration for a flow with one input and one output port + * @param inputChannelName + * @param outputChannelName + */ + public ChannelNamePortConfiguration(String inputChannelName, String outputChannelName) { + this.inputPortMetadata = new PortMetadata("input", inputChannelName); - @Override - public String getInputChannel() { - return this.inputPortMetadata.getChannelName(); - } + if (outputChannelName != null) { + PortMetadata outputPortMetadata = new PortMetadata("output", outputChannelName); + this.outputPortMetadataList = Collections.singletonList(outputPortMetadata); + } + else { + // this.outputPortMetadataList = new ArrayList(); + } + } - @Override - public String getOutputChannel(String portName) { - PortMetadata portMetadata = (PortMetadata) findOutputPort(portName); - if (portMetadata != null) { - return portMetadata.getChannelName(); - } - return null; - } + public String getInputPortName() { + return this.inputPortMetadata.getPortName(); + } - @Override - public List getOutputPortNames() { - List results = new ArrayList(); - if (outputPortMetadataList != null) { - for (PortMetadata portMetadata : outputPortMetadataList) { - results.add(portMetadata.getPortName()); - } - } - return results; - } + public String getInputChannel() { + return this.inputPortMetadata.getChannelName(); + } - private PortMetadata findOutputPort(String portName) { - if (outputPortMetadataList != null) { - for (PortMetadata portMetadata : outputPortMetadataList) { - if (portName.equals(portMetadata.getPortName())) { - return portMetadata; - } - } - } - return null; - } + public String getOutputChannel(String portName) { + PortMetadata portMetadata = findOutputPort(portName); + if (portMetadata != null) { + return portMetadata.getChannelName(); + } + return null; + } + + public List getOutputPortNames() { + List results = new ArrayList(); + if (outputPortMetadataList != null) { + for (PortMetadata portMetadata : outputPortMetadataList) { + results.add(portMetadata.getPortName()); + } + } + return results; + } + + private PortMetadata findOutputPort(String portName) { + if (outputPortMetadataList != null) { + for (PortMetadata portMetadata : outputPortMetadataList) { + if (portName.equals(portMetadata.getPortName())) { + return portMetadata; + } + } + } + return null; + } } diff --git a/src/main/java/org/springframework/integration/flow/Flow.java b/src/main/java/org/springframework/integration/flow/Flow.java index ce81dfb..06cffa2 100644 --- a/src/main/java/org/springframework/integration/flow/Flow.java +++ b/src/main/java/org/springframework/integration/flow/Flow.java @@ -12,16 +12,13 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.PropertyPlaceholderConfigurer; import org.springframework.beans.factory.support.BeanDefinitionValidationException; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.core.env.MutablePropertySources; -import org.springframework.core.env.PropertiesPropertySource; -import org.springframework.core.env.PropertySource; import org.springframework.integration.MessageChannel; import org.springframework.integration.channel.AbstractMessageChannel; -import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.flow.config.FlowUtils; import org.springframework.integration.flow.interceptor.FlowInterceptor; @@ -101,7 +98,6 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A this.configLocations = configLocations; } - @Override public void afterPropertiesSet() { if (this.flowId == null) { @@ -157,7 +153,6 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A return this.flowConfiguration; } - @Override public void setBeanName(String name) { this.beanName = name; @@ -221,20 +216,18 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A * @param the publish-subscribe channel */ public void setFlowOutputChannel(SubscribableChannel flowOutputChannel) { - this.flowOutputChannel = flowOutputChannel; + this.flowOutputChannel = flowOutputChannel; } - @Override public MessageChannel resolveChannelName(String channelName) { return flowChannelResolver.resolveChannelName(channelName); } private void addReferencedProperties() { if (flowProperties != null) { - PropertySource propertySource = new PropertiesPropertySource("flowProperties", flowProperties); - - MutablePropertySources propertySources = flowContext.getEnvironment().getPropertySources(); - propertySources.addLast(propertySource); + PropertyPlaceholderConfigurer ppc = new PropertyPlaceholderConfigurer(); + ppc.setProperties(flowProperties); + flowContext.addBeanFactoryPostProcessor(ppc); } } @@ -284,7 +277,6 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A } } - @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = applicationContext; } diff --git a/src/main/java/org/springframework/integration/flow/FlowConstants.java b/src/main/java/org/springframework/integration/flow/FlowConstants.java index 3235eac..52e697c 100644 --- a/src/main/java/org/springframework/integration/flow/FlowConstants.java +++ b/src/main/java/org/springframework/integration/flow/FlowConstants.java @@ -23,11 +23,6 @@ public interface FlowConstants { */ 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 */ diff --git a/src/main/java/org/springframework/integration/flow/config/FlowMessageHandlerFactoryBean.java b/src/main/java/org/springframework/integration/flow/config/FlowMessageHandlerFactoryBean.java index 373f7e5..10935d3 100644 --- a/src/main/java/org/springframework/integration/flow/config/FlowMessageHandlerFactoryBean.java +++ b/src/main/java/org/springframework/integration/flow/config/FlowMessageHandlerFactoryBean.java @@ -20,7 +20,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.MessageChannel; import org.springframework.integration.config.AbstractSimpleMessageHandlerFactoryBean; -import org.springframework.integration.core.MessageHandler; import org.springframework.integration.flow.Flow; import org.springframework.integration.flow.PortConfiguration; import org.springframework.integration.flow.handler.FlowMessageHandler; @@ -31,33 +30,34 @@ import org.springframework.util.Assert; * @author David Turanski * */ -public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerFactoryBean implements - InitializingBean { +public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerFactoryBean + implements InitializingBean { - @SuppressWarnings("unused") + @SuppressWarnings("unused") private static Log logger = LogFactory.getLog(FlowMessageHandlerFactoryBean.class); - + private volatile Flow flow; private volatile String inputPortName; - + private volatile MessageChannel errorChannel; private volatile long timeout; - //private volatile DirectChannel flowOutputChannel; - - private volatile PortConfiguration flowConfiguration; + // private volatile DirectChannel flowOutputChannel; + + private volatile PortConfiguration flowConfiguration; @Override - protected MessageHandler createHandler() { - + protected FlowMessageHandler createHandler() { + MessageChannel flowInputChannel = flow.resolveChannelName((String) flowConfiguration.getInputChannel()); - FlowMessageHandler flowMessageHandler = new FlowMessageHandler(flowInputChannel, flow.getFlowOutputChannel(), timeout); - + FlowMessageHandler flowMessageHandler = new FlowMessageHandler(flowInputChannel, flow.getFlowOutputChannel(), + timeout); + flowMessageHandler.setErrorChannel(this.errorChannel); - + return flowMessageHandler; } @@ -71,8 +71,9 @@ public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerF /** * - * @param inputPortName the flow input port associated with the handler. If not set - * and the flow defines only one input port, that will be used by default. + * @param inputPortName the flow input port associated with the handler. If + * not set and the flow defines only one input port, that will be used by + * default. */ public void setInputPortName(String inputPortName) { this.inputPortName = inputPortName; @@ -85,7 +86,7 @@ public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerF public void setTimeout(long timeout) { this.timeout = timeout; } - + /** * * @param errorChannel @@ -94,21 +95,19 @@ public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerF this.errorChannel = errorChannel; } - - @Override - public void afterPropertiesSet() throws Exception { - this.flowConfiguration = null; - if (this.inputPortName == null){ - Assert.isTrue(!(this.flow.getFlowConfiguration().getPortConfigurations().size() > 1), - "flow [" + this.flow.getBeanName() +"] exposes multiple port configurations. Must specify an input port"); - - this.flowConfiguration = this.flow.getFlowConfiguration().getPortConfigurations().get(0); - this.inputPortName = this.flowConfiguration.getInputPortName(); - } - else { - this.flowConfiguration = this.flow.getFlowConfiguration().getConfigurationForInputPort( - this.inputPortName); - } - } + public void afterPropertiesSet() throws Exception { + this.flowConfiguration = null; + if (this.inputPortName == null) { + Assert.isTrue(!(this.flow.getFlowConfiguration().getPortConfigurations().size() > 1), + "flow [" + this.flow.getBeanName() + + "] exposes multiple port configurations. Must specify an input port"); + + this.flowConfiguration = this.flow.getFlowConfiguration().getPortConfigurations().get(0); + this.inputPortName = this.flowConfiguration.getInputPortName(); + } + else { + this.flowConfiguration = this.flow.getFlowConfiguration().getConfigurationForInputPort(this.inputPortName); + } + } } diff --git a/src/main/java/org/springframework/integration/flow/config/xml/FlowConfigurationParser.java b/src/main/java/org/springframework/integration/flow/config/xml/FlowConfigurationParser.java index 8ea0ae3..c152630 100644 --- a/src/main/java/org/springframework/integration/flow/config/xml/FlowConfigurationParser.java +++ b/src/main/java/org/springframework/integration/flow/config/xml/FlowConfigurationParser.java @@ -37,66 +37,63 @@ import org.w3c.dom.Element; */ public class FlowConfigurationParser implements BeanDefinitionParser { - @Override public BeanDefinition parse(Element element, ParserContext parserContext) { List portMappings = DomUtils.getChildElementsByTagName(element, "port-mapping"); - + BeanDefinitionBuilder flowConfigurationBuilder = BeanDefinitionBuilder .genericBeanDefinition(FlowConfiguration.class); ManagedList portConfigList = new ManagedList(); for (Element el : portMappings) { - if (!DomUtils.getChildElements(el).isEmpty()){ - if (el.hasAttribute("input-channel") || el.hasAttribute("output-channel")){ - parserContext.getReaderContext().error( - "port-mapping cannot include both channel attributes and child elements", - flowConfigurationBuilder); - } - BeanDefinition portConfiguration = buildFlowProviderPortConfiguration(el, parserContext); - portConfigList.add(portConfiguration); - } - else - { - // A default port configuration - if (!(el.hasAttribute("input-channel"))){ - parserContext.getReaderContext().error( - "port-mapping with no child elements must include an 'input-channel' attribute", - flowConfigurationBuilder); - } - - - BeanDefinitionBuilder portConfigurationBuilder = BeanDefinitionBuilder - .genericBeanDefinition(ChannelNamePortConfiguration.class); - - portConfigurationBuilder.addConstructorArgValue(el.getAttribute("input-channel")); - if (el.hasAttribute("output-channel")){ - portConfigurationBuilder.addConstructorArgValue(el.getAttribute("output-channel")); - } else { - portConfigurationBuilder.addConstructorArgValue(null); - } - portConfigList.add(portConfigurationBuilder.getBeanDefinition()); - } + if (!DomUtils.getChildElements(el).isEmpty()) { + if (el.hasAttribute("input-channel") || el.hasAttribute("output-channel")) { + parserContext.getReaderContext().error( + "port-mapping cannot include both channel attributes and child elements", + flowConfigurationBuilder); + } + BeanDefinition portConfiguration = buildFlowProviderPortConfiguration(el, parserContext); + portConfigList.add(portConfiguration); + } + else { + // A default port configuration + if (!(el.hasAttribute("input-channel"))) { + parserContext.getReaderContext().error( + "port-mapping with no child elements must include an 'input-channel' attribute", + flowConfigurationBuilder); + } + + BeanDefinitionBuilder portConfigurationBuilder = BeanDefinitionBuilder + .genericBeanDefinition(ChannelNamePortConfiguration.class); + + portConfigurationBuilder.addConstructorArgValue(el.getAttribute("input-channel")); + if (el.hasAttribute("output-channel")) { + portConfigurationBuilder.addConstructorArgValue(el.getAttribute("output-channel")); + } + else { + portConfigurationBuilder.addConstructorArgValue(null); + } + portConfigList.add(portConfigurationBuilder.getBeanDefinition()); + } } flowConfigurationBuilder.addConstructorArgValue(portConfigList); - BeanDefinitionReaderUtils.registerWithGeneratedName(flowConfigurationBuilder.getBeanDefinition(), - parserContext.getRegistry()); - + parserContext.getRegistry()); + return null; } private BeanDefinition buildFlowProviderPortConfiguration(Element el, ParserContext parserContext) { Element inputPortEl = DomUtils.getChildElementByTagName(el, "input-port"); - + BeanDefinitionBuilder portConfigurationBuilder = BeanDefinitionBuilder .genericBeanDefinition(ChannelNamePortConfiguration.class); BeanDefinition portMetadata = this.buildPortMetadata(el, inputPortEl); - + portConfigurationBuilder.addConstructorArgValue(portMetadata); List outputPortElements = DomUtils.getChildElementsByTagName(el, "output-port"); @@ -115,11 +112,10 @@ public class FlowConfigurationParser implements BeanDefinitionParser { return portConfigurationBuilder.getBeanDefinition(); } - - + private BeanDefinition buildPortMetadata(Element element, Element portElement) { BeanDefinitionBuilder portMetadataBuilder = BeanDefinitionBuilder.genericBeanDefinition(PortMetadata.class); - portMetadataBuilder.addConstructorArgValue(portElement.getAttribute("name")); + portMetadataBuilder.addConstructorArgValue(portElement.getAttribute("name")); portMetadataBuilder.addConstructorArgValue(portElement.getAttribute("channel")); return portMetadataBuilder.getBeanDefinition(); } diff --git a/src/main/java/org/springframework/integration/flow/config/xml/FlowNamespaceHandler.java b/src/main/java/org/springframework/integration/flow/config/xml/FlowNamespaceHandler.java index 6060b70..ee4379d 100644 --- a/src/main/java/org/springframework/integration/flow/config/xml/FlowNamespaceHandler.java +++ b/src/main/java/org/springframework/integration/flow/config/xml/FlowNamespaceHandler.java @@ -24,11 +24,10 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa */ public class FlowNamespaceHandler extends AbstractIntegrationNamespaceHandler { - @Override public void init() { registerBeanDefinitionParser("flow", new FlowParser()); registerBeanDefinitionParser("outbound-gateway", new FlowOutboundGatewayParser()); registerBeanDefinitionParser("flow-configuration", new FlowConfigurationParser()); - } + } } diff --git a/src/main/java/org/springframework/integration/flow/config/xml/FlowParser.java b/src/main/java/org/springframework/integration/flow/config/xml/FlowParser.java index 88dc3e7..2ca58a7 100644 --- a/src/main/java/org/springframework/integration/flow/config/xml/FlowParser.java +++ b/src/main/java/org/springframework/integration/flow/config/xml/FlowParser.java @@ -33,18 +33,15 @@ import org.w3c.dom.Element; */ public class FlowParser implements BeanDefinitionParser { - @Override public BeanDefinition parse(Element element, ParserContext parserContext) { - - Element props = DomUtils.getChildElementByTagName(element,"props"); - - if(element.hasAttribute("properties") && props !=null) { - parserContext.getReaderContext().error( - "Element cannot have both 'properties' attribute and inner 'props' element",element); - } - - + Element props = DomUtils.getChildElementByTagName(element, "props"); + + if (element.hasAttribute("properties") && props != null) { + parserContext.getReaderContext().error( + "Element cannot have both 'properties' attribute and inner 'props' element", element); + } + BeanDefinitionBuilder flowBuilder = BeanDefinitionBuilder.genericBeanDefinition(Flow.class); String id = element.getAttribute("id"); BeanDefinitionBuilder flowOutputChannelBuilder = BeanDefinitionBuilder @@ -52,20 +49,18 @@ public class FlowParser implements BeanDefinitionParser { String beanName = FlowUtils.registerBeanDefinition(flowOutputChannelBuilder.getBeanDefinition(), parserContext.getRegistry()); flowBuilder.addPropertyReference("flowOutputChannel", beanName); - + IntegrationNamespaceUtils.setValueIfAttributeDefined(flowBuilder, element, "referenced-bean-locations"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(flowBuilder, element, "properties"); IntegrationNamespaceUtils.setValueIfAttributeDefined(flowBuilder, element, "help"); IntegrationNamespaceUtils.setValueIfAttributeDefined(flowBuilder, element, "flow-id"); - - - - if (props != null) { - flowBuilder.addPropertyValue("properties",parserContext.getDelegate().parsePropsElement(props)); + + if (props != null) { + flowBuilder.addPropertyValue("properties", parserContext.getDelegate().parsePropsElement(props)); } - + BeanDefinition beanDefinition = flowBuilder.getBeanDefinition(); - + parserContext.getRegistry().registerBeanDefinition(id, beanDefinition); return beanDefinition; } 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 79c352d..7cb00a1 100644 --- a/src/main/java/org/springframework/integration/flow/handler/FlowMessageHandler.java +++ b/src/main/java/org/springframework/integration/flow/handler/FlowMessageHandler.java @@ -16,7 +16,6 @@ package org.springframework.integration.flow.handler; import java.util.Collections; -import java.util.Map; import java.util.UUID; import org.apache.commons.logging.Log; @@ -85,14 +84,9 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler { @Override protected Object handleRequestMessage(Message requestMessage) { - - String conversationId = this.toString(); - Map flowConversationIdHeader = Collections.singletonMap( - FlowConstants.FLOW_CONVERSATION_ID_HEADER, this.toString()); - - Message message = MessageBuilder.fromMessage(requestMessage).copyHeaders(flowConversationIdHeader) - .build(); - + UUID conversationId = requestMessage.getHeaders().getId(); + Message message = MessageBuilder.fromMessage(requestMessage).pushSequenceDetails(conversationId, 0, 0) + .build(); try { ResponseMessageHandler responseMessageHandler = new ResponseMessageHandler(conversationId); @@ -103,15 +97,16 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler { } catch (MessagingException me) { log.error(me.getMessage(), me); - if (conversationId - .equals(me.getFailedMessage().getHeaders().get(FlowConstants.FLOW_CONVERSATION_ID_HEADER))) { + + if (conversationId.equals(me.getFailedMessage().getHeaders().getCorrelationId())) { if (errorChannel != null) { errorChannel.send(new ErrorMessage(me, Collections.singletonMap( FlowConstants.FLOW_OUTPUT_PORT_HEADER, (Object) FlowConstants.FLOW_HANDLER_EXCEPTION_HEADER_VALUE))); - + } - } else { + } + else { throw me; } } @@ -122,11 +117,11 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler { * Internal MessageHandler for the flow response */ private static class ResponseMessageHandler implements MessageHandler { - private final String conversationId; + private final UUID conversationId; private volatile Message response; - public ResponseMessageHandler(String conversationId) { + public ResponseMessageHandler(UUID conversationId) { this.conversationId = conversationId; } @@ -137,20 +132,27 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler { * org.springframework.integration.core.MessageHandler#handleMessage * (org.springframework.integration.Message) */ - @Override public void handleMessage(Message message) throws MessagingException { - - if (conversationId.equals(message.getHeaders().get(FlowConstants.FLOW_CONVERSATION_ID_HEADER))) { - this.response = message; + if (log.isDebugEnabled()) { + log.debug("handling flow response message with conversation Id " + + message.getHeaders().getCorrelationId() + ". Target conversation Id = " + this.conversationId + + " match = " + conversationId.equals(message.getHeaders().getCorrelationId())); + } + if (conversationId.equals(message.getHeaders().getCorrelationId())) { + this.response = MessageBuilder.fromMessage(message).popSequenceDetails().build(); + if (log.isDebugEnabled()) { + log.debug("set flow response message " + this.response); + } } else { + /* - * Response from flow's ErrorChannel which is mapped to an output port. + * Response from flow's ErrorChannel which is mapped to an + * output port. */ if (message instanceof ErrorMessage) { MessagingException me = (MessagingException) message.getPayload(); - if (conversationId.equals(me.getFailedMessage().getHeaders() - .get(FlowConstants.FLOW_CONVERSATION_ID_HEADER))) { + if (conversationId.equals(me.getFailedMessage().getHeaders().getCorrelationId())) { 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 41fc2e5..e5c14b8 100644 --- a/src/main/java/org/springframework/integration/flow/interceptor/FlowInterceptor.java +++ b/src/main/java/org/springframework/integration/flow/interceptor/FlowInterceptor.java @@ -30,7 +30,7 @@ import org.springframework.integration.support.MessageBuilder; /** * A ChannelInterceptor to set the Flow output port header * @see FlowUtils - * + * * @author David Turanski * */ @@ -39,7 +39,7 @@ public class FlowInterceptor extends ChannelInterceptorAdapter { private final String portName; - /** + /** * @param portName the value of the message header */ public FlowInterceptor(String portName) { @@ -48,11 +48,13 @@ public class FlowInterceptor extends ChannelInterceptorAdapter { @Override public Message preSend(Message message, MessageChannel channel) { - - log.debug("flow interceptor " + this.hashCode() + " received a message from port " + portName + " on channel " - + channel); - Map headersToCopy = Collections.singletonMap(FlowConstants.FLOW_OUTPUT_PORT_HEADER, (Object) portName); - return MessageBuilder.fromMessage(message).copyHeadersIfAbsent(headersToCopy).build(); - + if (log.isDebugEnabled()) { + log.debug(this + " received a message from port " + portName + " on channel " + channel); + } + + Map headersToCopy = Collections.singletonMap(FlowConstants.FLOW_OUTPUT_PORT_HEADER, + (Object) portName); + + return MessageBuilder.fromMessage(message).copyHeaders(headersToCopy).build(); } } diff --git a/src/test/java/org/springframework/integration/flow/FlowUtilsTests.java b/src/test/java/org/springframework/integration/flow/FlowUtilsTests.java new file mode 100644 index 0000000..0e5f559 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/FlowUtilsTests.java @@ -0,0 +1,40 @@ +package org.springframework.integration.flow; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; + +import org.junit.Test; +import org.springframework.integration.Message; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.flow.config.FlowUtils; +import org.springframework.integration.message.GenericMessage; +/** + * + * @author David Turanski + * + */ +public class FlowUtilsTests { + @Test + public void buildBridge(){ + + SubscribableChannel inputChannel = new DirectChannel(); + SubscribableChannel outputChannel = new PublishSubscribeChannel(); + PollableChannel receiveChannel = new QueueChannel(); + + FlowUtils.bridgeChannels(inputChannel, outputChannel); + FlowUtils.bridgeChannels(outputChannel, receiveChannel); + + Message message = new GenericMessage("hello"); + + inputChannel.send(message); + Message result = receiveChannel.receive(100); + assertNotNull(result); + assertSame(message, result); + + } + +} diff --git a/src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceActivator.java b/src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceActivator.java index aa8fc5a..4b718ec 100644 --- a/src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceActivator.java +++ b/src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceActivator.java @@ -30,7 +30,6 @@ public class TransactionalServiceActivator implements RequestReplyExchanger { /* (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/config/xml/BarFactory.java b/src/test/java/org/springframework/integration/flow/config/xml/BarFactory.java index 02ba228..c17b83a 100644 --- a/src/test/java/org/springframework/integration/flow/config/xml/BarFactory.java +++ b/src/test/java/org/springframework/integration/flow/config/xml/BarFactory.java @@ -4,19 +4,16 @@ import org.springframework.beans.factory.FactoryBean; public class BarFactory implements FactoryBean { - @Override public Bar getObject() throws Exception { // TODO Auto-generated method stub return new Bar(); } - @Override - public Class getObjectType() { + public Class getObjectType() { // TODO Auto-generated method stub return Bar.class; } - @Override public boolean isSingleton() { // TODO Auto-generated method stub return true; diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowClientNamespaceTests-context.xml b/src/test/java/org/springframework/integration/flow/config/xml/FlowClientNamespaceTests-context.xml new file mode 100644 index 0000000..59d38ac --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowClientNamespaceTests-context.xml @@ -0,0 +1,42 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + val1 + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowClientNamespaceTests.java b/src/test/java/org/springframework/integration/flow/config/xml/FlowClientNamespaceTests.java new file mode 100644 index 0000000..46a6cb3 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowClientNamespaceTests.java @@ -0,0 +1,91 @@ +/* + * 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.assertNotNull; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.Message; +import org.springframework.integration.flow.Flow; +import org.springframework.integration.test.support.AbstractRequestResponseScenarioTest; +import org.springframework.integration.test.support.MessageValidator; +import org.springframework.integration.test.support.RequestResponseScenario; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * + * @author David Turanski + * + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class FlowClientNamespaceTests extends AbstractRequestResponseScenarioTest { + + @Autowired + @Qualifier("flowWithProps") + Flow flowWithProps; + + @Test + public void testFlowWithInnerProps() { + assertEquals("val1", flowWithProps.getProperties().getProperty("key1")); + } + + /* + * (non-Javadoc) + * + * @see org.springframework.integration.test.support. + * AbstractRequestResponseScenarioTest#defineRequestResponseScenarios() + */ + @Override + protected List defineRequestResponseScenarios() { + List scenarios = new ArrayList(); + RequestResponseScenario scenario1 = new RequestResponseScenario("inputC1", "outputC1") + .setName("gateway-response-scenario").setPayload("hello").setResponseValidator(new MessageValidator() { + + @Override + protected void validateMessage(Message message) { + assertNotNull(message); + + } + }); + + scenarios.add(scenario1); + + RequestResponseScenario scenario2 = new RequestResponseScenario("inputC2", "outputC2") + .setName("gateway-discard-scenario").setPayload("world").setResponseValidator(new MessageValidator() { + + @Override + protected void validateMessage(Message reply) { + assertEquals("gateway-discard", reply.getHeaders().get("flow.output.port")); + assertEquals("yeah!", reply.getHeaders().get("gateway")); + + } + }); + + scenarios.add(scenario2); + + return scenarios; + } + +} diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowConfigNamespaceTests-context.xml b/src/test/java/org/springframework/integration/flow/config/xml/FlowConfigNamespaceTests-context.xml new file mode 100644 index 0000000..4b305af --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowConfigNamespaceTests-context.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowConfigNamespaceTests.java b/src/test/java/org/springframework/integration/flow/config/xml/FlowConfigNamespaceTests.java new file mode 100644 index 0000000..4cea638 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowConfigNamespaceTests.java @@ -0,0 +1,65 @@ +/* + * 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.assertNotNull; + +import java.util.Iterator; +import java.util.Map; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.integration.flow.FlowConfiguration; +import org.springframework.integration.flow.PortConfiguration; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * + * @author David Turanski + * + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class FlowConfigNamespaceTests { + @Autowired + ApplicationContext applicationContext; + + @Test + public void test() { + + Map flowConfigurations = applicationContext.getBeansOfType(FlowConfiguration.class); + + Iterator iterator = flowConfigurations.values().iterator(); + FlowConfiguration flowConfiguration = iterator.next(); + assertNotNull(flowConfiguration.getPortConfigurations()); + assertEquals(2, flowConfiguration.getPortConfigurations().size()); + PortConfiguration pc0 = flowConfiguration.getPortConfigurations().get(0); + assertEquals("input", pc0.getInputPortName()); + assertEquals("subflow-input", pc0.getInputChannel()); + + assertEquals("subflow-output", pc0.getOutputChannel("output")); + assertEquals(1, pc0.getOutputPortNames().size()); + + flowConfiguration = iterator.next(); + assertNotNull(flowConfiguration.getPortConfigurations()); + assertEquals (1, flowConfiguration.getPortConfigurations().size()); + } + +} diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowContextTests-context.xml b/src/test/java/org/springframework/integration/flow/config/xml/FlowContextTests-context.xml new file mode 100644 index 0000000..8827133 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowContextTests-context.xml @@ -0,0 +1,22 @@ + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowContextTests-jmx-context.xml b/src/test/java/org/springframework/integration/flow/config/xml/FlowContextTests-jmx-context.xml new file mode 100644 index 0000000..7692902 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowContextTests-jmx-context.xml @@ -0,0 +1,22 @@ + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowContextTests.java b/src/test/java/org/springframework/integration/flow/config/xml/FlowContextTests.java new file mode 100644 index 0000000..f1581e8 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowContextTests.java @@ -0,0 +1,45 @@ +/* + * 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 org.junit.Test; +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.context.support.ClassPathXmlApplicationContext; + + +/** + * @author David Turanski + * + */ + +public class FlowContextTests { + @Test(expected=BeanCreationException.class) + public void testChannelConflictShouldThrowException() { + try { + new ClassPathXmlApplicationContext( + "/org/springframework/integration/flow/config/xml/FlowContextTests-context.xml"); + } catch (BeanCreationException e) { + System.out.println(e.getCause().getMessage()); + throw e; + } + } + + @Test + public void testFlowWithJMX() { + + new ClassPathXmlApplicationContext( + "/org/springframework/integration/flow/config/xml/FlowContextTests-jmx-context.xml"); + + } + +} diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowWithAutowireTests-context.xml b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithAutowireTests-context.xml new file mode 100644 index 0000000..0bd969b --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithAutowireTests-context.xml @@ -0,0 +1,10 @@ + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowWithAutowireTests.java b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithAutowireTests.java new file mode 100644 index 0000000..895909b --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithAutowireTests.java @@ -0,0 +1,22 @@ +package org.springframework.integration.flow.config.xml; + +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.flow.Flow; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class FlowWithAutowireTests { + @Autowired + Flow flow; + @Test + public void test() { + Foo foo = flow.getFlowContext().getBean(Foo.class); + assertNotNull(foo.bar); + } +} diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowWithErrorTests-context.xml b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithErrorTests-context.xml new file mode 100644 index 0000000..5d2f0f0 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithErrorTests-context.xml @@ -0,0 +1,39 @@ + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowWithErrorTests.java b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithErrorTests.java new file mode 100644 index 0000000..d924dd8 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithErrorTests.java @@ -0,0 +1,106 @@ +/* + * 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.assertNotNull; +import static org.junit.Assert.assertTrue; + +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.PollableChannel; +import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.message.GenericMessage; + +/** + * + * @author David Turanski + * + */ + +public class FlowWithErrorTests { + + @Test + public void testFlowThrowsExceptionWithGatewayErrorChannel() { + ApplicationContext applicationContext = new ClassPathXmlApplicationContext( + "/org/springframework/integration/flow/config/xml/FlowWithErrorTests-context.xml"); + MessageChannel inputChannel = applicationContext.getBean("inputC", MessageChannel.class); + SubscribableChannel errorChannel = applicationContext.getBean("errorChannel", SubscribableChannel.class); + Message msg = new GenericMessage("hello"); + Handler handler = new Handler(); + errorChannel.subscribe(handler); + inputChannel.send(msg); + assertTrue(handler.gotResponse); + + } + + @Test + public void testDirectCallWithErrorChannel() { + ApplicationContext applicationContext = new ClassPathXmlApplicationContext( + "classpath:/META-INF/spring/integration/flows/subflow5/subflow5-context.xml"); + MessageChannel inputChannel = applicationContext.getBean("subflow-input", MessageChannel.class); + SubscribableChannel errorChannel = applicationContext.getBean("errorChannel", SubscribableChannel.class); + + errorChannel.subscribe(new MessageHandler() { + + public void handleMessage(Message message) throws MessagingException { + assertTrue(message.getPayload() instanceof MessagingException); + } + }); + + Message msg = new GenericMessage("hello"); + assertTrue(inputChannel.send(msg)); + } + + @Test + public void testWithErrorChannel() { + ApplicationContext applicationContext = new ClassPathXmlApplicationContext( + "/org/springframework/integration/flow/config/xml/FlowWithErrorTests-context.xml"); + MessageChannel inputChannel = applicationContext.getBean("inputC1", MessageChannel.class); + PollableChannel output = applicationContext.getBean("outputC1", PollableChannel.class); + Message msg = new GenericMessage("hello"); + inputChannel.send(msg); + + Message reply = output.receive(100); + assertNotNull(reply); + assertTrue(reply.getPayload() instanceof MessagingException); + } + + private static class Handler implements MessageHandler { + public boolean gotResponse; + + @SuppressWarnings("unused") + public Message message; + + /* + * (non-Javadoc) + * + * @see + * org.springframework.integration.core.MessageHandler#handleMessage + * (org.springframework.integration.Message) + */ + public void handleMessage(Message message) throws MessagingException { + this.gotResponse = true; + this.message = message; + } + + } + +} diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowWithOptionalResponseTests-context.xml b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithOptionalResponseTests-context.xml new file mode 100644 index 0000000..511e26c --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithOptionalResponseTests-context.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowWithOptionalResponseTests.java b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithOptionalResponseTests.java new file mode 100644 index 0000000..91fad44 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithOptionalResponseTests.java @@ -0,0 +1,60 @@ +package org.springframework.integration.flow.config.xml; + +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +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.message.GenericMessage; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class FlowWithOptionalResponseTests { + @Autowired + @Qualifier("inputC") + MessageChannel input; + + @Autowired + @Qualifier("inputCO") + MessageChannel inputForOptionalResponse; + + @Autowired + @Qualifier("outputC") + SubscribableChannel output; + + @Test + public void testOneWay() { + + input.send(new GenericMessage("hello")); + } + + @Test + public void testOptionResponse() { + TestMessageHandler counter = new TestMessageHandler(); + + output.subscribe(counter); + + for (int i = 0; i < 100; i++) { + inputForOptionalResponse.send(new GenericMessage("hello")); + } + + assertTrue(String.valueOf(counter.count), counter.count > 1 && counter.count < 100); + } + + static class TestMessageHandler implements MessageHandler { + int count = 0; + + public void handleMessage(Message message) throws MessagingException { + count++; + } + } + +} diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowWithReferencesTests-context.xml b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithReferencesTests-context.xml new file mode 100644 index 0000000..134e9b2 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithReferencesTests-context.xml @@ -0,0 +1,29 @@ + + + + + + + + + val1 + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FlowWithReferencesTests.java b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithReferencesTests.java new file mode 100644 index 0000000..72c0c5d --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/FlowWithReferencesTests.java @@ -0,0 +1,58 @@ +/* + * 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 org.junit.runner.RunWith; +import org.springframework.integration.Message; +import org.springframework.integration.test.support.MessageValidator; +import org.springframework.integration.test.support.RequestResponseScenario; +import org.springframework.integration.test.support.SingleRequestResponseScenarioTest; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * + * @author David Turanski + * + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class FlowWithReferencesTests extends SingleRequestResponseScenarioTest { + +/* (non-Javadoc) + * @see org.springframework.integration.test.support.SingleRequestResponseScenarioTest#defineRequestResponseScenario() + */ +@Override +protected RequestResponseScenario defineRequestResponseScenario() { + RequestResponseScenario scenario = + new RequestResponseScenario("inputC","outputC") + .setPayload("hello") + .setResponseValidator(new MessageValidator() { + + @Override + protected void validateMessage(Message reply) { + assertEquals("it works!",reply.getHeaders().get("refbean.value")); + assertEquals("val1",reply.getHeaders().get("property.value.1")); + assertEquals("undefined",reply.getHeaders().get("property.value.2")); + }} + ); + return scenario; +} + + +} diff --git a/src/test/java/org/springframework/integration/flow/config/xml/FooFactory.java b/src/test/java/org/springframework/integration/flow/config/xml/FooFactory.java index 66d31cb..5e8bb60 100644 --- a/src/test/java/org/springframework/integration/flow/config/xml/FooFactory.java +++ b/src/test/java/org/springframework/integration/flow/config/xml/FooFactory.java @@ -6,7 +6,7 @@ public class FooFactory implements FactoryBean{ @Autowired Bar bar; - @Override + public Foo getObject() throws Exception { // TODO Auto-generated method stub Foo foo = new Foo(); @@ -14,13 +14,13 @@ public class FooFactory implements FactoryBean{ return foo; } - @Override + public Class getObjectType() { // TODO Auto-generated method stub return Foo.class; } - @Override + public boolean isSingleton() { // TODO Auto-generated method stub return true; diff --git a/src/test/java/org/springframework/integration/flow/config/xml/NestedFlowTests-context.xml b/src/test/java/org/springframework/integration/flow/config/xml/NestedFlowTests-context.xml new file mode 100644 index 0000000..d938be2 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/NestedFlowTests-context.xml @@ -0,0 +1,21 @@ + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/NestedFlowTests.java b/src/test/java/org/springframework/integration/flow/config/xml/NestedFlowTests.java new file mode 100644 index 0000000..8312f1f --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/NestedFlowTests.java @@ -0,0 +1,62 @@ +package org.springframework.integration.flow.config.xml; + +import static org.junit.Assert.assertEquals; + +import java.util.concurrent.atomic.AtomicInteger; + +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.MessageChannel; +import org.springframework.integration.MessagingException; +import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class NestedFlowTests { + @Autowired + MessageChannel inputChannel; + + @Autowired + SubscribableChannel outputChannel; + + @Test + @DirtiesContext + public void testNestedFlowOneExecution() { + final AtomicInteger count = new AtomicInteger(); + outputChannel.subscribe(new MessageHandler() { + + public void handleMessage(Message message) throws MessagingException { + count.getAndIncrement(); + + }}); + inputChannel.send(new GenericMessage("hello")); + assertEquals(1,count.get()); + } + + @Test + @DirtiesContext + public void testNestedFlowMultipleExecutions() { + final AtomicInteger count = new AtomicInteger(); + + outputChannel.subscribe(new MessageHandler() { + + public void handleMessage(Message message) throws MessagingException { + count.getAndIncrement(); + + }}); + + + for (int i=0; i<2; i++) { + inputChannel.send(new GenericMessage("hello")); + } + assertEquals(2,count.get()); + } + +} diff --git a/src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTests-context.xml b/src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTests-context.xml new file mode 100644 index 0000000..57ba3d3 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTests-context.xml @@ -0,0 +1,18 @@ + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTests.java b/src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTests.java new file mode 100644 index 0000000..c6296c2 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTests.java @@ -0,0 +1,132 @@ +/* + * 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 TransactionalFlowTests { + @Test + public void testFlowDirectCommit() { + ApplicationContext applicationContext = new ClassPathXmlApplicationContext( + "/META-INF/spring/integration/flows/transactional-flow/flow-context.xml", + "/org/springframework/integration/flow/config/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", + "/org/springframework/integration/flow/config/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( + "/org/springframework/integration/flow/config/xml/TransactionalFlowTests-context.xml", + "/org/springframework/integration/flow/config/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 testFlowRollbackWithGatewayErrorChannel() { + ApplicationContext applicationContext = new ClassPathXmlApplicationContext( + "/org/springframework/integration/flow/config/xml/TransactionalFlowTests-context.xml", + "/org/springframework/integration/flow/config/xml/txmanager-config.xml"); + MessageChannel inputChannel = applicationContext.getBean("inputC", MessageChannel.class); + SubscribableChannel errorChannel = applicationContext.getBean("errorChannel", SubscribableChannel.class); + StubTransactionManager transactionManager = applicationContext.getBean(StubTransactionManager.class); + Handler handler = new Handler(); + errorChannel.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) + */ + + public void handleMessage(Message message) throws MessagingException { + this.messageReceived = true; + this.message = message; + } + + } +} diff --git a/src/test/java/org/springframework/integration/flow/config/xml/autowired-referenced-beans.xml b/src/test/java/org/springframework/integration/flow/config/xml/autowired-referenced-beans.xml new file mode 100644 index 0000000..c34e251 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/autowired-referenced-beans.xml @@ -0,0 +1,8 @@ + + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/ref-bean-config.xml b/src/test/java/org/springframework/integration/flow/config/xml/ref-bean-config.xml new file mode 100644 index 0000000..88efa47 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/ref-bean-config.xml @@ -0,0 +1,10 @@ + + + + + + + + diff --git a/src/test/java/org/springframework/integration/flow/config/xml/txmanager-config.xml b/src/test/java/org/springframework/integration/flow/config/xml/txmanager-config.xml new file mode 100644 index 0000000..c4a7b36 --- /dev/null +++ b/src/test/java/org/springframework/integration/flow/config/xml/txmanager-config.xml @@ -0,0 +1,7 @@ + + + + + diff --git a/src/test/resources/META-INF/spring/integration/flows/basic1/flow-context.xml b/src/test/resources/META-INF/spring/integration/flows/basic1/flow-context.xml new file mode 100644 index 0000000..d161bcb --- /dev/null +++ b/src/test/resources/META-INF/spring/integration/flows/basic1/flow-context.xml @@ -0,0 +1,19 @@ + + + + + + + + + + + + + diff --git a/src/test/resources/META-INF/spring/integration/flows/basic2/flow-context.xml b/src/test/resources/META-INF/spring/integration/flows/basic2/flow-context.xml new file mode 100644 index 0000000..a201aef --- /dev/null +++ b/src/test/resources/META-INF/spring/integration/flows/basic2/flow-context.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + diff --git a/src/test/resources/META-INF/spring/integration/flows/nested/flow-context.xml b/src/test/resources/META-INF/spring/integration/flows/nested/flow-context.xml new file mode 100644 index 0000000..1540328 --- /dev/null +++ b/src/test/resources/META-INF/spring/integration/flows/nested/flow-context.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + + + + + + + diff --git a/src/test/resources/log4j.xml b/src/test/resources/log4j.xml index d99732e..377dfd4 100644 --- a/src/test/resources/log4j.xml +++ b/src/test/resources/log4j.xml @@ -16,8 +16,12 @@ - + + + + +