diff --git a/src/main/java/org/springframework/integration/flow/ChannelNamePortConfiguration.java b/src/main/java/org/springframework/integration/flow/ChannelNamePortConfiguration.java index 70ec359..8bedb57 100644 --- a/src/main/java/org/springframework/integration/flow/ChannelNamePortConfiguration.java +++ b/src/main/java/org/springframework/integration/flow/ChannelNamePortConfiguration.java @@ -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 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; } + /** + * 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); diff --git a/src/main/java/org/springframework/integration/flow/Flow.java b/src/main/java/org/springframework/integration/flow/Flow.java index 1ef9c8c..bb1f22e 100644 --- a/src/main/java/org/springframework/integration/flow/Flow.java +++ b/src/main/java/org/springframework/integration/flow/Flow.java @@ -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; - } + } } diff --git a/src/main/java/org/springframework/integration/flow/FlowConfiguration.java b/src/main/java/org/springframework/integration/flow/FlowConfiguration.java index c2772e7..3b50d13 100644 --- a/src/main/java/org/springframework/integration/flow/FlowConfiguration.java +++ b/src/main/java/org/springframework/integration/flow/FlowConfiguration.java @@ -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 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 getPortConfigurations() { return portConfigurations; diff --git a/src/main/java/org/springframework/integration/flow/PortConfiguration.java b/src/main/java/org/springframework/integration/flow/PortConfiguration.java index fcc666e..f236583 100644 --- a/src/main/java/org/springframework/integration/flow/PortConfiguration.java +++ b/src/main/java/org/springframework/integration/flow/PortConfiguration.java @@ -18,7 +18,7 @@ package org.springframework.integration.flow; import java.util.Collection; /** - * + * Strategy interface for Port configurations * @author David Turanski * */ diff --git a/src/main/java/org/springframework/integration/flow/PortMetadata.java b/src/main/java/org/springframework/integration/flow/PortMetadata.java index 6030f82..6e8eb3e 100644 --- a/src/main/java/org/springframework/integration/flow/PortMetadata.java +++ b/src/main/java/org/springframework/integration/flow/PortMetadata.java @@ -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 * */ 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 0ad8df9..edd2a55 100644 --- a/src/main/java/org/springframework/integration/flow/config/FlowMessageHandlerFactoryBean.java +++ b/src/main/java/org/springframework/integration/flow/config/FlowMessageHandlerFactoryBean.java @@ -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; } 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 3c51a9b..ad8d6e0 100644 --- a/src/main/java/org/springframework/integration/flow/config/FlowUtils.java +++ b/src/main/java/org/springframework/integration/flow/config/FlowUtils.java @@ -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); 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 d0ce2c5..8ea0ae3 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 @@ -30,6 +30,7 @@ import org.springframework.util.xml.DomUtils; import org.w3c.dom.Element; /** + * Parse the {@link FlowConfiguration} * * @author David Turanski * 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 073885d..6060b70 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 @@ -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()); - } + } } diff --git a/src/main/java/org/springframework/integration/flow/config/xml/FlowPortConfigurationParser.java b/src/main/java/org/springframework/integration/flow/config/xml/FlowPortConfigurationParser.java deleted file mode 100644 index 898ccf2..0000000 --- a/src/main/java/org/springframework/integration/flow/config/xml/FlowPortConfigurationParser.java +++ /dev/null @@ -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; - } - -} 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 2a4ab0c..e08e785 100644 --- a/src/main/java/org/springframework/integration/flow/handler/FlowMessageHandler.java +++ b/src/main/java/org/springframework/integration/flow/handler/FlowMessageHandler.java @@ -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 flowConversationIdHeader = Collections.singletonMap(FLOW_CONVERSATION_ID_HEADER, + Map 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; } } 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 4fd29f2..2657d7e 100644 --- a/src/main/java/org/springframework/integration/flow/interceptor/FlowInterceptor.java +++ b/src/main/java/org/springframework/integration/flow/interceptor/FlowInterceptor.java @@ -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 headersToCopy = Collections.singletonMap("flow.output.port", (Object) portName); + Map headersToCopy = Collections.singletonMap(FlowUtils.FLOW_OUTPUT_PORT_HEADER, (Object) portName); return MessageBuilder.fromMessage(message).copyHeadersIfAbsent(headersToCopy).build(); } diff --git a/src/main/resources/org/springframework/integration/flow/config/spring-integration-flow-2.1.xsd b/src/main/resources/org/springframework/integration/flow/config/spring-integration-flow-2.1.xsd index b62c7ac..1f7acfc 100644 --- a/src/main/resources/org/springframework/integration/flow/config/spring-integration-flow-2.1.xsd +++ b/src/main/resources/org/springframework/integration/flow/config/spring-integration-flow-2.1.xsd @@ -124,14 +124,6 @@ - - - - - - - - - - - +