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

cleanup and javadoc

Browse Source
This commit is contained in:
David Turanski
2011-08-25 10:15:19 -04:00
parent e611198ef8
commit f1e7c0f122
14 changed files with 267 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/main/java/org/springframework/integration/flow/ChannelNamePortConfiguration.java
View File

@@ -21,17 +21,31 @@ 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<PortMetadata> 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<PortMetadata> outputPortMetadataList) {
this.outputPortMetadataList = outputPortMetadataList;
this.inputPortMetadata = 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);

282
src/main/java/org/springframework/integration/flow/Flow.java
View File

@@ -26,179 +26,227 @@ import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Encapsulates a message flow with inputs and outputs exposed via a message
* port
* Encapsulates a Spring Integration message flow with inputs and outputs
* abstracted by a {@link FlowConfiguration} Creates a Spring Integration flow
* in a child application context. This facilitates reuse of the message flow
* within complex messaging flows. Each flow instance may be configured by
* injecting property values or referenced bean locations to allow for different
* bean definitions used by the flow. In addition, beans defined in the parent
* application context may be referenced or overridden in the flow application
* context.
*
* By convention the flow configuration resource locations are
* classpath:META-INF/spring/integration/flows/[flow-id]/*.xml
*
* The flow-id defaults to the bean name if not set
*
* @author David Turanski
*
*/
public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, ApplicationContextAware {
private static Log logger = LogFactory.getLog(Flow.class);
private static Log logger = LogFactory.getLog(Flow.class);
private volatile ClassPathXmlApplicationContext flowContext;
private volatile ClassPathXmlApplicationContext flowContext;
private ApplicationContext applicationContext;
private ApplicationContext applicationContext;
private volatile FlowConfiguration flowConfiguration;
private volatile FlowConfiguration flowConfiguration;
private volatile String[] configLocations;
private volatile String[] configLocations;
private volatile String[] referencedBeanLocations;
private volatile String[] referencedBeanLocations;
private volatile Properties flowProperties;
private volatile Properties flowProperties;
private volatile String beanName;
private volatile String beanName;
private volatile String flowId;
private volatile String flowId;
private volatile ChannelResolver flowChannelResolver;
private volatile ChannelResolver flowChannelResolver;
private volatile PublishSubscribeChannel flowOutputChannel;
private volatile PublishSubscribeChannel flowOutputChannel;
private volatile boolean help;
private volatile boolean help;
public Flow() {
/**
* Default constructor
*/
public Flow() {
}
}
public Flow(Properties flowProperties, String[] configLocations) {
this.flowProperties = flowProperties;
this.configLocations = configLocations;
}
/**
*
* @param flowProperties properties for this flow instance
* @param configLocations Spring configuration resource locations containing
* bean definitions included in the flow application context
*/
public Flow(Properties flowProperties, String[] configLocations) {
this.flowProperties = flowProperties;
this.configLocations = configLocations;
}
public Flow(String[] configLocations) {
this.configLocations = configLocations;
}
/**
*
* @param configLocations Spring configuration resource locations containing
* bean definitions included in the flow application context
*/
public Flow(String[] configLocations) {
this.configLocations = configLocations;
}
@Override
public void afterPropertiesSet() {
@Override
public void afterPropertiesSet() {
if (this.flowId == null) {
this.flowId = this.beanName;
}
if (this.flowId == null) {
this.flowId = this.beanName;
}
if (this.help) {
System.out.println(FlowUtils.getDocumentation(this.flowId));
}
if (this.help) {
System.out.println(FlowUtils.getDocumentation(this.flowId));
}
if (configLocations == null) {
configLocations = new String[] { String.format("classpath:META-INF/spring/integration/flows/%s/*.xml",
this.flowId) };
}
if (configLocations == null) {
configLocations = new String[] { String.format("classpath:META-INF/spring/integration/flows/%s/*.xml",
this.flowId) };
}
if (referencedBeanLocations != null) {
configLocations = (String[]) ArrayUtils.addAll(configLocations, referencedBeanLocations);
}
if (referencedBeanLocations != null) {
configLocations = (String[]) ArrayUtils.addAll(configLocations, referencedBeanLocations);
}
logger.debug("instantiating flow context from configLocations ["
+ StringUtils.arrayToCommaDelimitedString(configLocations) + "]");
logger.debug("instantiating flow context from configLocations ["
+ StringUtils.arrayToCommaDelimitedString(configLocations) + "]");
Assert.notEmpty(configLocations, "configLocations cannot be empty");
Assert.notEmpty(configLocations, "configLocations cannot be empty");
flowContext = new ClassPathXmlApplicationContext(applicationContext);
flowContext = new ClassPathXmlApplicationContext(applicationContext);
addReferencedProperties();
addReferencedProperties();
this.flowContext.setConfigLocations(configLocations);
this.flowContext.setConfigLocations(configLocations);
this.flowContext.refresh();
this.flowContext.refresh();
this.flowConfiguration = flowContext.getBean(FlowConfiguration.class);
Assert.notNull(flowConfiguration, "flow context does not contain a flow configuration");
this.flowConfiguration = flowContext.getBean(FlowConfiguration.class);
Assert.notNull(flowConfiguration, "flow context does not contain a flow configuration");
validatePortMapping();
validatePortMapping();
this.flowChannelResolver = new BeanFactoryChannelResolver(flowContext);
this.flowChannelResolver = new BeanFactoryChannelResolver(flowContext);
bridgeMessagingPorts();
bridgeMessagingPorts();
}
}
public FlowConfiguration getFlowConfiguration() {
return this.flowConfiguration;
}
public FlowConfiguration getFlowConfiguration() {
return this.flowConfiguration;
}
@Override
public void setBeanName(String name) {
this.beanName = name;
@Override
public void setBeanName(String name) {
this.beanName = name;
}
}
public String getBeanName() {
return this.beanName;
}
public String getBeanName() {
return this.beanName;
}
public void setFlowId(String flowId) {
this.flowId = flowId;
}
/**
* @param flowId The flow identifier used to locate the flow configuration
*/
public void setFlowId(String flowId) {
this.flowId = flowId;
}
/**
*
* @param referencedBeanLocations Additional resource locations containing referenced bean definitions
*/
public void setReferencedBeanLocations(String[] referencedBeanLocations) {
this.referencedBeanLocations = referencedBeanLocations;
}
public void setReferencedBeanLocations(String[] referencedBeanLocations) {
this.referencedBeanLocations = referencedBeanLocations;
}
/**
*
* @param flowProperties properties referenced in the flow definition property placeholders
*/
public void setProperties(Properties flowProperties) {
this.flowProperties = flowProperties;
}
public void setProperties(Properties flowProperties) {
this.flowProperties = flowProperties;
}
public Properties getProperties() {
return this.flowProperties;
}
public Properties getProperties() {
return this.flowProperties;
}
public void setHelp(boolean help) {
this.help = help;
}
/**
*
* @param help if true write the flow documentation to stdout
* The default document location is "classpath:META-INF/spring/integration/flows/[flow-id]/flow.doc"
*/
public void setHelp(boolean help) {
this.help = help;
}
public PublishSubscribeChannel getFlowOutputChannel() {
return flowOutputChannel;
}
/**
* All flow outputs defined in the {@link PortConfiguration} are bridged to a single PublishSubscribeChannel
* @return the publish-subscribe channel
*/
public PublishSubscribeChannel getFlowOutputChannel() {
return flowOutputChannel;
}
public void setFlowOutputChannel(PublishSubscribeChannel flowOutputChannel) {
this.flowOutputChannel = flowOutputChannel;
}
/**
* All flow outputs defined in the {@link PortConfiguration} are bridged to a single PublishSubscribeChannel
* @param the publish-subscribe channel
*/
public void setFlowOutputChannel(PublishSubscribeChannel flowOutputChannel) {
this.flowOutputChannel = flowOutputChannel;
}
@Override
public MessageChannel resolveChannelName(String channelName) {
return flowChannelResolver.resolveChannelName(channelName);
}
@Override
public MessageChannel resolveChannelName(String channelName) {
return flowChannelResolver.resolveChannelName(channelName);
}
private void addReferencedProperties() {
if (flowProperties != null) {
PropertySource<?> propertySource = new PropertiesPropertySource("flowProperties", flowProperties);
private void addReferencedProperties() {
if (flowProperties != null) {
PropertySource<?> propertySource = new PropertiesPropertySource("flowProperties", flowProperties);
MutablePropertySources propertySources = flowContext.getEnvironment().getPropertySources();
propertySources.addLast(propertySource);
}
MutablePropertySources propertySources = flowContext.getEnvironment().getPropertySources();
propertySources.addLast(propertySource);
}
}
}
private void validatePortMapping() {
Assert.notEmpty(this.flowConfiguration.getPortConfigurations(),
"flow configuration contains no port configurations");
}
private void validatePortMapping() {
Assert.notEmpty(this.flowConfiguration.getPortConfigurations(),
"flow configuration contains no port configurations");
}
private void bridgeMessagingPorts() {
private void bridgeMessagingPorts() {
/*
* create a bridge for each target output port to the flow outputChannel
*/
for (PortConfiguration targetPortConfiguration : this.getFlowConfiguration().getPortConfigurations()) {
for (String outputPort : targetPortConfiguration.getOutputPortNames()) {
String targetOutputChannelName = (String) targetPortConfiguration.getOutputChannel(outputPort);
SubscribableChannel inputChannel = (SubscribableChannel) resolveChannelName(targetOutputChannelName);
/*
* create a bridge for each target output port to the flow outputChannel
*/
for (PortConfiguration targetPortConfiguration : this.getFlowConfiguration().getPortConfigurations()) {
for (String outputPort : targetPortConfiguration.getOutputPortNames()) {
String targetOutputChannelName = (String) targetPortConfiguration.getOutputChannel(outputPort);
SubscribableChannel inputChannel = (SubscribableChannel) resolveChannelName(targetOutputChannelName);
((AbstractMessageChannel) inputChannel).addInterceptor(new FlowInterceptor(outputPort));
((AbstractMessageChannel) inputChannel).addInterceptor(new FlowInterceptor(outputPort));
logger.debug("creating output bridge on [" + outputPort + "] inputChannelName = ["
+ targetOutputChannelName + "] outputChannel = [" + this.flowOutputChannel + "]");
FlowUtils.bridgeChannels(inputChannel, this.flowOutputChannel);
}
}
}
logger.debug("creating output bridge on [" + outputPort + "] inputChannelName = ["
+ targetOutputChannelName + "] outputChannel = [" + this.flowOutputChannel + "]");
FlowUtils.bridgeChannels(inputChannel, this.flowOutputChannel);
}
}
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
}
}

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

@@ -18,6 +18,8 @@ package org.springframework.integration.flow;
import java.util.List;
/**
* A container holding a {@link Flow} configuration. A flow configuration
* may contain multiple {@link PortConfiguration}s
*
* @author David Turanski
*
@@ -28,14 +30,14 @@ public class FlowConfiguration {
/**
*
* @param portConfigurations
* @param portConfigurations
*/
public FlowConfiguration(List<PortConfiguration> portConfigurations) {
this.portConfigurations = portConfigurations;
}
/**
*
* Get the configuration by input port name
* @param inputPortName
* @return
*/
@@ -48,8 +50,8 @@ public class FlowConfiguration {
return null;
}
/**
*
* @return
* Get all port configurations
* @return the port configurations
*/
public List<PortConfiguration> getPortConfigurations() {
return portConfigurations;

2
src/main/java/org/springframework/integration/flow/PortConfiguration.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.flow;
import java.util.Collection;
/**
*
* Strategy interface for Port configurations
* @author David Turanski
*
*/

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

@@ -16,7 +16,8 @@
package org.springframework.integration.flow;
/**
*
* A container for defining a message port configuration used in a {@link Flow}.
* Binds a logical port name to a channel defined internally by the flow
* @author David Turanski
*
*/

15
src/main/java/org/springframework/integration/flow/config/FlowMessageHandlerFactoryBean.java
View File

@@ -27,7 +27,7 @@ import org.springframework.integration.flow.handler.FlowMessageHandler;
import org.springframework.util.Assert;
/**
*
* Creates an instance of {@link FlowMessageHandler}
* @author David Turanski
*
*/
@@ -57,14 +57,27 @@ public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerF
return flowMessageHandler;
}
/**
*
* @param flow the flow handled by the FlowMessageHandler
*/
public void setFlow(Flow flow) {
this.flow = flow;
}
/**
*
* @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;
}
/**
*
* @param timeout send timeout for the handler
*/
public void setTimeout(long timeout) {
this.timeout = timeout;
}

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

@@ -28,11 +28,26 @@ import org.springframework.integration.handler.BridgeHandler;
import org.springframework.util.ResourceUtils;
/**
* Utility functions used by the flow parsers
*
* @author David Turanski
*
*/
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() {}
/**
* Create a bridge
*
* @param inputChannel
@@ -65,9 +80,15 @@ public class FlowUtils {
return beanName;
}
public static String getDocumentation(String flowName) {
/**
* Read the flow documentation resource into a String if it exists.
* The location is classpath:META-INF/spring/integration/flows/[flowId]/flow.doc
* @param flowId the flow id
* @return the documentation
*/
public static String getDocumentation(String flowId) {
String path = String.format("classpath:META-INF/spring/integration/flows/%s/flow.doc", flowName);
String path = String.format("classpath:META-INF/spring/integration/flows/%s/flow.doc", flowId);
try {
File file = ResourceUtils.getFile(path);

1
src/main/java/org/springframework/integration/flow/config/xml/FlowConfigurationParser.java
View File

@@ -30,6 +30,7 @@ import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* Parse the {@link FlowConfiguration}
*
* @author David Turanski
*

3
src/main/java/org/springframework/integration/flow/config/xml/FlowNamespaceHandler.java
View File

@@ -29,7 +29,6 @@ public class FlowNamespaceHandler extends AbstractIntegrationNamespaceHandler {
registerBeanDefinitionParser("flow", new FlowParser());
registerBeanDefinitionParser("outbound-gateway", new FlowOutboundGatewayParser());
registerBeanDefinitionParser("flow-configuration", new FlowConfigurationParser());
registerBeanDefinitionParser("port-mapping", new FlowPortConfigurationParser());
}
}
}

36
src/main/java/org/springframework/integration/flow/config/xml/FlowPortConfigurationParser.java
View File

@@ -1,36 +0,0 @@
/*
* 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.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.w3c.dom.Element;
/**
*
* @author David Turanski
*
*/
public class FlowPortConfigurationParser implements BeanDefinitionParser {
@Override
public BeanDefinition parse(Element arg0, ParserContext arg1) {
// TODO Auto-generated method stub
return null;
}
}

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

@@ -26,21 +26,32 @@ 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.config.FlowUtils;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.support.MessageBuilder;
/**
* A MessageHandler for Handling Flow input and output. Sends messages on its
* input channel to the flow input channel and replies with the flow output (if there
* is one) to its output channel.
*
* Internally creates a subscriber to a PublishSubscribeChannel automatically created for
* the flow. Since all FlowMessageHandler instances subscribe to this channel, a unique
* flow conversationId is used to correlate flow input and output messages
*
* The output message contains a FLOW_OUTPUT_PORT_HEADER identifying which flow output port produced the message
* @see FlowUtils
*
* Error handling is done in a standard way. If the flow includes an error channel which is bound to an output port, the
* handler will send the ErrorMessage to the outputChannel. If the flow throws an exception, the handler will create an
* ErrorMessage and send it to the outputChannel
*
* @author David Turanski
*
*/
public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
/**
*
*/
private static final String FLOW_CONVERSATION_ID_HEADER = "flow.conversation.id";
private static Log log = LogFactory.getLog(FlowMessageHandler.class);
@@ -49,7 +60,12 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
private final SubscribableChannel flowOutputChannel;
private final long timeout;
/**
*
* @param flowInputChannel the Flow input channel
* @param flowOutputChannel a PublishSubscribeChannel internally created and bridged to all flow output channels
* @param timeout the send timeout duration
*/
public FlowMessageHandler(MessageChannel flowInputChannel, SubscribableChannel flowOutputChannel, long timeout) {
this.flowInputChannel = flowInputChannel;
this.flowOutputChannel = flowOutputChannel;
@@ -61,7 +77,7 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
protected Object handleRequestMessage(Message<?> requestMessage) {
UUID conversationId = requestMessage.getHeaders().getId();
Map<String, Object> flowConversationIdHeader = Collections.singletonMap(FLOW_CONVERSATION_ID_HEADER,
Map<String, Object> flowConversationIdHeader = Collections.singletonMap(FlowUtils.FLOW_CONVERSATION_ID_HEADER,
(Object) conversationId);
Message<?> message = MessageBuilder.fromMessage(requestMessage).copyHeaders(flowConversationIdHeader)
@@ -79,13 +95,16 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
}
catch (MessagingException me) {
log.error(me.getMessage(), me);
if (conversationId.equals(me.getFailedMessage().getHeaders().get(FLOW_CONVERSATION_ID_HEADER))) {
if (conversationId.equals(me.getFailedMessage().getHeaders().get(FlowUtils.FLOW_CONVERSATION_ID_HEADER))) {
return new ErrorMessage(me);
}
}
return null;
}
/*
* Internal MessageHandler for the flow response
*/
private static class ResponseMessageHandler implements MessageHandler {
private final UUID conversationId;
private volatile Message<?> response;
@@ -103,12 +122,12 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (conversationId.equals(message.getHeaders().get(FLOW_CONVERSATION_ID_HEADER))) {
if (conversationId.equals(message.getHeaders().get(FlowUtils.FLOW_CONVERSATION_ID_HEADER))) {
this.response = message;
} else {
if (message instanceof ErrorMessage){
MessagingException me = (MessagingException) message.getPayload();
if (conversationId.equals(me.getFailedMessage().getHeaders().get(FLOW_CONVERSATION_ID_HEADER))) {
if (conversationId.equals(me.getFailedMessage().getHeaders().get(FlowUtils.FLOW_CONVERSATION_ID_HEADER))) {
this.response = message;
}
}

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

@@ -23,10 +23,13 @@ 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.config.FlowUtils;
import org.springframework.integration.support.MessageBuilder;
/**
*
* A ChannelInterceptor to set the Flow output port header
* @see FlowUtils
*
* @author David Turanski
*
*/
@@ -35,9 +38,11 @@ public class FlowInterceptor extends ChannelInterceptorAdapter {
private final String portName;
/**
* @param portName the value of the message header
*/
public FlowInterceptor(String portName) {
this.portName = portName;
}
@Override
@@ -45,7 +50,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("flow.output.port", (Object) portName);
Map<String, Object> headersToCopy = Collections.singletonMap(FlowUtils.FLOW_OUTPUT_PORT_HEADER, (Object) portName);
return MessageBuilder.fromMessage(message).copyHeadersIfAbsent(headersToCopy).build();
}

8
src/main/resources/org/springframework/integration/flow/config/spring-integration-flow-2.1.xsd
View File

@@ -124,14 +124,6 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="port-mapping" type="PortMappingType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines an integration flow port mapping configuraiotn
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:complexType name="PortMappingType">
<xsd:sequence>
<xsd:element name="input-port" minOccurs="0"

6
src/test/resources/FlowConfigNamespaceTest-context.xml
View File

@@ -17,11 +17,7 @@
</int-flow:port-mapping>
</int-flow:flow-configuration>
<int-flow:port-mapping>
<int-flow:input-port name="input2" channel="subflow-input2" />
<int-flow:output-port name="output2" channel="subflow-output2" />
</int-flow:port-mapping>
<int-flow:flow-configuration>
<int-flow:port-mapping input-channel="inputChannel"
output-channel="outputChannel" />