Added checks for channel bean definition conflicts in flow context

src/main/java/org/springframework/integration/flow/Flow.java

@@ -1,5 +1,7 @@
 package org.springframework.integration.flow;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Properties;
 
 import org.apache.commons.lang.ArrayUtils;
@@ -8,6 +10,7 @@ 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.support.BeanDefinitionValidationException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -35,7 +38,7 @@ import org.springframework.util.StringUtils;
  * application context may be referenced or overridden in the flow application
  * context.
  * 
- * By convention the flow configuration resource locations are 
+ * 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
@@ -116,15 +119,20 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
 			configLocations = (String[]) ArrayUtils.addAll(configLocations, referencedBeanLocations);
 		}
 
-		logger.debug("instantiating flow context from configLocations ["
-				+ StringUtils.arrayToCommaDelimitedString(configLocations) + "]");
-
 		Assert.notEmpty(configLocations, "configLocations cannot be empty");
 
+		/*
+		 * create a child application context
+		 */
 		flowContext = new ClassPathXmlApplicationContext(applicationContext);
 
 		addReferencedProperties();
 
+		if (logger.isDebugEnabled()) {
+			logger.debug("instantiating flow context from configLocations ["
+					+ StringUtils.arrayToCommaDelimitedString(configLocations) + "]");
+		}
+
 		this.flowContext.setConfigLocations(configLocations);
 
 		this.flowContext.refresh();
@@ -160,9 +168,11 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
 	public void setFlowId(String flowId) {
 		this.flowId = flowId;
 	}
+
 	/**
 	 * 
-	 * @param referencedBeanLocations Additional resource locations containing referenced bean definitions
+	 * @param referencedBeanLocations Additional resource locations containing
+	 * referenced bean definitions
 	 */
 	public void setReferencedBeanLocations(String[] referencedBeanLocations) {
 		this.referencedBeanLocations = referencedBeanLocations;
@@ -170,7 +180,8 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
 
 	/**
 	 * 
-	 * @param flowProperties properties referenced in the flow definition property placeholders
+	 * @param flowProperties properties referenced in the flow definition
+	 * property placeholders
 	 */
 	public void setProperties(Properties flowProperties) {
 		this.flowProperties = flowProperties;
@@ -182,15 +193,17 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
 
 	/**
 	 * 
-	 * @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"
+	 * @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;
 	}
 
 	/**
-	 * All flow outputs defined in the {@link PortConfiguration} are bridged to a single PublishSubscribeChannel
+	 * All flow outputs defined in the {@link PortConfiguration} are bridged to
+	 * a single PublishSubscribeChannel
 	 * @return the publish-subscribe channel
 	 */
 	public PublishSubscribeChannel getFlowOutputChannel() {
@@ -198,7 +211,8 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
 	}
 
 	/**
-	 * All flow outputs defined in the {@link PortConfiguration} are bridged to a single PublishSubscribeChannel
+	 * All flow outputs defined in the {@link PortConfiguration} are bridged to
+	 * a single PublishSubscribeChannel
 	 * @param the publish-subscribe channel
 	 */
 	public void setFlowOutputChannel(PublishSubscribeChannel flowOutputChannel) {
@@ -223,10 +237,50 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
 	private void validatePortMapping() {
 		Assert.notEmpty(this.flowConfiguration.getPortConfigurations(),
 				"flow configuration contains no port configurations");
+
+		List<String> errors = new ArrayList<String>();
+		for (PortConfiguration portConfiguration : this.flowConfiguration.getPortConfigurations()) {
+			String inputChannelName = (String) portConfiguration.getInputChannel();
+			validateFlowChannelDefinition(inputChannelName, errors, false);
+
+			for (String outputPortName : portConfiguration.getOutputPortNames()) {
+				String outputChannelName = (String) portConfiguration.getOutputChannel(outputPortName);
+				validateFlowChannelDefinition(outputChannelName, errors, true);
+			}
+		}
+		if (errors.size() > 0 ) {
+			 
+			throw new BeanDefinitionValidationException("\n"+StringUtils.arrayToDelimitedString(errors.toArray(),"\n"));
+		}
+	}
+
+	/*
+	 * If flow context does not contain the bean definition then the definition
+	 * comes from the parent context. The flow should should still work with a
+	 * 'global' PublishSubscribeChannel output channel
+	 */
+	private void validateFlowChannelDefinition(String channelName, List<String> errors, boolean allowPubSub) {
+
+		MessageChannel channel = this.flowContext.getBean(channelName, MessageChannel.class);
+
+		if (!this.flowContext.containsBeanDefinition(channelName)) {
+			if (channel instanceof PublishSubscribeChannel && allowPubSub) {
+				 if (logger.isDebugEnabled()) {
+					 logger.warn("Flow '" +  this.flowId +"'" +
+						" is sharing the publish-subscribe channel '" + channelName +"'" + 
+						" with the parent context.");
+				 }
+			} else {
+				errors.add("The flow channel '"
+						+ channelName
+						+ "' in flow '"
+						+ this.flowId
+						+ "' conflicts with a bean definition in the parent context. It must be explicitly declared in the flow'");
+			}
+		}
 	}
 
 	private void bridgeMessagingPorts() {
-
 		/*
 		 * create a bridge for each target output port to the flow outputChannel
 		 */
@@ -247,6 +301,5 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
 		this.applicationContext = applicationContext;
-
 	}
 }

src/test/java/org/springframework/integration/flow/config/xml/FlowContextTest.java

@@ -0,0 +1,35 @@
+/*
+ * 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 FlowContextTest {
+	@Test(expected=BeanCreationException.class)
+	public void testChannelConflictShouldThrowException() {
+	  try {
+		new ClassPathXmlApplicationContext("/FlowContextTest-context.xml");
+	  } catch (BeanCreationException e) {
+		  System.out.println(e.getCause().getMessage());
+		  throw e;
+	  }
+	}
+}

src/test/resources/FlowContextTest-context.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-flow="http://www.springframework.org/schema/integration/flow"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/flow http://www.springframework.org/schema/integration/flow/spring-integration-flow.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<!-- Instantiate the flow -->
+	<int-flow:flow id="subflow" flow-id="flow-with-channel-conflict"/>
+       
+    <!-- input port not required if only one -->
+	<int-flow:outbound-gateway flow="subflow" 
+		input-channel="inputC"
+		output-channel="outputC"/>
+	
+	<int:channel id="outputC"/>
+	 
+
+</beans>

src/test/resources/META-INF/spring/integration/flows/flow-with-channel-conflict/invalid-flow-context.xml

@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:util="http://www.springframework.org/schema/util"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-flow="http://www.springframework.org/schema/integration/flow"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
+		http://www.springframework.org/schema/integration/flow http://www.springframework.org/schema/integration/flow/spring-integration-flow.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
+		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.0.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+	
+	<int-flow:flow-configuration>
+		<int-flow:port-mapping input-channel="inputC" output-channel="outputC"/>	 
+	</int-flow:flow-configuration>
+
+<!-- implicit channel declaration the same name as parent context. Should throw exception -->
+	<int:bridge input-channel="inputC" output-channel="outputC"/> 
+    
+  
+	 
+</beans>