ChannelFactory is no longer tied to the MessageBus. Instead it will be discovered based on the bean name ("channelFactory"). The DefaultChannelFactoryBean no longer creates proxies. This also fixes INT-322.

org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultChannelFactoryBean.java

@@ -16,37 +16,31 @@
 
 package org.springframework.integration.bus;
 
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.lang.reflect.Proxy;
 import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicReference;
 
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.FactoryBean;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.integration.channel.ChannelInterceptor;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.channel.factory.ChannelFactory;
 import org.springframework.integration.channel.factory.QueueChannelFactory;
-import org.springframework.util.Assert;
 
 /**
- * Creates a channel by delegating to the current message bus' configured
- * ChannelFactory. Tries to retrieve the {@link ChannelFactory} from the
- * single {@link MessageBus} defined in the {@link ApplicationContext}.
+ * Creates a channel by delegating to the "channelFactory" bean defined
+ * within the {@link ApplicationContext} or else the default implementation
+ * (QueueChannelFactory).
+ * <p>
  * As a {@link FactoryBean}, this class is solely intended to be used within
  * an ApplicationContext.
  *
  * @author Marius Bogoevici
  * @author Mark Fisher
  */
-public class DefaultChannelFactoryBean implements ApplicationContextAware, FactoryBean, BeanNameAware, InitializingBean {
+public class DefaultChannelFactoryBean implements ApplicationContextAware, FactoryBean, BeanNameAware {
+
+	public static final String CHANNEL_FACTORY_BEAN_NAME = "channelFactory";
 
 	private volatile String beanName;
 
@@ -54,12 +48,10 @@ public class DefaultChannelFactoryBean implements ApplicationContextAware, Facto
 
 	private volatile ApplicationContext applicationContext;
 
-	private volatile boolean initialized;
+	private volatile MessageChannel channel;
 
 	private final Object initializationMonitor = new Object();
 
-	private volatile Object proxyBean;
-
 
 	public void setApplicationContext(ApplicationContext applicationContext) {
 		this.applicationContext = applicationContext;
@@ -73,76 +65,31 @@ public class DefaultChannelFactoryBean implements ApplicationContextAware, Facto
 		this.interceptors = interceptors;
 	}
 
-	public void afterPropertiesSet() throws Exception {
+	public Object getObject() throws Exception {
 		synchronized (this.initializationMonitor) {
-			if (!initialized) {
-				this.proxyBean = Proxy.newProxyInstance(
-						getClass().getClassLoader(),
-						new Class[] { PollableChannel.class },
-						new DefaultChannelInvocationHandler());
-				this.initialized = true;
+			if (this.channel == null) {
+				ChannelFactory channelFactory = null;
+				if (this.applicationContext.containsBean(CHANNEL_FACTORY_BEAN_NAME)) {
+					channelFactory = (ChannelFactory) this.applicationContext.getBean(CHANNEL_FACTORY_BEAN_NAME);
+				}
+				else {
+					channelFactory = new QueueChannelFactory();
+				}
+				this.channel = channelFactory.getChannel(this.beanName, this.interceptors);
 			}
 		}
-	}
-
-	public Object getObject() throws Exception {
-		if (!this.initialized) {
-			afterPropertiesSet();
-		}
-		return proxyBean;
+		return this.channel;
 	}
 
 	public Class<?> getObjectType() {
-		return PollableChannel.class;
+		if (this.channel == null) {
+			return MessageChannel.class;
+		}
+		return this.channel.getClass();
 	}
 
 	public boolean isSingleton() {
 		return true;
 	}
 
-
-	private class DefaultChannelInvocationHandler implements InvocationHandler {
-
-		private volatile AtomicReference<MessageChannel> targetChannelReference = new AtomicReference<MessageChannel>();
-
-
-		private MessageChannel getTargetChannel() {
-			if (targetChannelReference.get() == null) {
-				targetChannelReference.compareAndSet(null, createMessageChannel());
-			}
-			return targetChannelReference.get();
-		}
-
-		private MessageChannel createMessageChannel() {
-			ChannelFactory channelFactory;
-			Map map = DefaultChannelFactoryBean.this.applicationContext.getBeansOfType(MessageBus.class);
-			Assert.state(map.size() <= 1, "There is more than one MessageBus in the ApplicationContext");
-			if (map.isEmpty()) {
-				channelFactory = new QueueChannelFactory();
-			}
-			else {
-				channelFactory = ((MessageBus) map.values().iterator().next()).getChannelFactory();
-			}
-			return channelFactory.getChannel(beanName, interceptors);
-		}
-
-		public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
-			if (method.getName().equals("equals")) {
-				return proxy == args[0];
-			}
-			else if (method.getName().equals("hashCode")) {
-				return System.identityHashCode(proxy);
-			}
-			else {
-				try {
-					return method.invoke(getTargetChannel(), args);
-				}
-				catch (InvocationTargetException e) {
-					throw e.getCause();
-				}
-			}
-		}
-
-	}
-
 }

org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java

@@ -44,8 +44,6 @@ import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.DefaultChannelRegistry;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
-import org.springframework.integration.channel.factory.ChannelFactory;
-import org.springframework.integration.channel.factory.QueueChannelFactory;
 import org.springframework.integration.dispatcher.PollingDispatcher;
 import org.springframework.integration.endpoint.DefaultEndpointRegistry;
 import org.springframework.integration.endpoint.EndpointRegistry;
@@ -76,8 +74,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
-	private volatile ChannelFactory channelFactory = new QueueChannelFactory();
-
 	private final ChannelRegistry channelRegistry = new DefaultChannelRegistry();
 
 	private final EndpointRegistry endpointRegistry = new DefaultEndpointRegistry();
@@ -108,16 +104,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 
 	private final Object lifecycleMonitor = new Object();
 
-	/**
-	 * Set the {@link ChannelFactory} to use for auto-creating channels.
-	 */
-	public void setChannelFactory(ChannelFactory channelFactory) {
-		this.channelFactory = channelFactory;
-	}
-
-	public ChannelFactory getChannelFactory() {
-		return channelFactory;
-	}
 
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
 		Assert.notNull(applicationContext, "'applicationContext' must not be null");

org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java

@@ -20,7 +20,6 @@ import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.channel.factory.ChannelFactory;
 import org.springframework.integration.endpoint.EndpointRegistry;
 
 /**
@@ -32,6 +31,4 @@ public interface MessageBus extends ChannelRegistry, EndpointRegistry, Lifecycle
 
 	MessageChannel getErrorChannel();
 
-	ChannelFactory getChannelFactory();
-
 }

org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd

@@ -33,7 +33,6 @@
 			</xsd:sequence>
 			<xsd:attribute name="auto-startup" type="xsd:boolean"/>
 			<xsd:attribute name="task-scheduler" type="xsd:string"/>
-			<xsd:attribute name="channel-factory" type="xsd:string"/>
 			<xsd:attribute name="configure-async-event-multicaster" type="xsd:boolean"/>
 		</xsd:complexType>
 	</xsd:element>

org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java

@@ -16,10 +16,6 @@
 
 package org.springframework.integration.channel.config;
 
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.Proxy;
-import java.util.concurrent.atomic.AtomicReference;
-
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -71,7 +67,7 @@ public class ChannelParserTests {
 		MessageChannel channel = (MessageChannel) context.getBean("queueChannelByDefault");
 		//called to initialize the channel instance
 		channel.getName();
-		assertEquals(QueueChannel.class, extractProxifiedChannel(channel).getClass());
+		assertEquals(QueueChannel.class, channel.getClass());
 	}
 
 	@Test
@@ -224,14 +220,4 @@ public class ChannelParserTests {
 		assertTrue(threwException);
 	}
 
-
-	@SuppressWarnings("unchecked")
-    public static MessageChannel extractProxifiedChannel (Object channelProxy) {
-	    InvocationHandler handler = Proxy.getInvocationHandler(channelProxy);
-	    DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler);
-	    AtomicReference<MessageChannel> reference =
-			    (AtomicReference<MessageChannel>) handlerAccessor.getPropertyValue("targetChannelReference");
-	    return reference.get();
-    }
-
 }

org.springframework.integration/src/test/java/org/springframework/integration/channel/config/DefaultChannelParserTests.java

@@ -16,7 +16,7 @@
 
 package org.springframework.integration.channel.config;
 
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertEquals;
 
 import org.junit.Test;
 
@@ -34,7 +34,7 @@ public class DefaultChannelParserTests {
     public void testDefaultChannel() {
         ApplicationContext context = new ClassPathXmlApplicationContext("defaultChannelParserTests.xml", this.getClass());
         MessageChannel channel = (MessageChannel) context.getBean("defaultChannel");
-        assertTrue(StubChannel.class.isAssignableFrom(ChannelParserTests.extractProxifiedChannel(channel).getClass()));
+        assertEquals(StubChannel.class, channel.getClass());
     }
 
 }

org.springframework.integration/src/test/java/org/springframework/integration/channel/config/defaultChannelParserTests.xml

@@ -7,9 +7,7 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
 
-	<message-bus channel-factory="stubChannelFactory"/>
-
-    <beans:bean id="stubChannelFactory" class="org.springframework.integration.channel.factory.StubChannelFactory"/>
+    <beans:bean id="channelFactory" class="org.springframework.integration.channel.factory.StubChannelFactory"/>
 
     <channel id="defaultChannel"/>
     

org.springframework.integration/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java

@@ -30,7 +30,6 @@ import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.context.support.StaticApplicationContext;
 import org.springframework.integration.bus.DefaultChannelFactoryBean;
-import org.springframework.integration.bus.DefaultMessageBus;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.ChannelInterceptor;
 import org.springframework.integration.channel.DirectChannel;
@@ -39,7 +38,6 @@ import org.springframework.integration.channel.PriorityChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.channel.RendezvousChannel;
 import org.springframework.integration.channel.ThreadLocalChannel;
-import org.springframework.integration.channel.config.ChannelParserTests;
 import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageBuilder;
@@ -105,20 +103,16 @@ public class ChannelFactoryTests {
 
 	@Test
 	public void testDefaultChannelFactoryBean() throws Exception{
-		DefaultMessageBus messageBus = new DefaultMessageBus();
-		ChannelFactory channelFactory = new StubChannelFactory();
-		messageBus.setChannelFactory(channelFactory);
 		StaticApplicationContext applicationContext = new StaticApplicationContext();
-		BeanDefinitionBuilder messageBusDefinitionBuilder = BeanDefinitionBuilder.rootBeanDefinition(DefaultMessageBus.class);
-		messageBusDefinitionBuilder.getBeanDefinition().getPropertyValues().addPropertyValue("channelFactory", channelFactory);
-		applicationContext.registerBeanDefinition("messageBus", messageBusDefinitionBuilder.getBeanDefinition());
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(StubChannelFactory.class);
+		applicationContext.registerBeanDefinition("channelFactory", builder.getBeanDefinition());
 		DefaultChannelFactoryBean channelFactoryBean =  new DefaultChannelFactoryBean();
 		channelFactoryBean.setBeanName("testChannel");
 		channelFactoryBean.setApplicationContext(applicationContext);
 		channelFactoryBean.setInterceptors(interceptors);
 		MessageChannel channel = (MessageChannel) channelFactoryBean.getObject();      
 		channel.getName();
-		assertEquals(StubChannel.class, ChannelParserTests.extractProxifiedChannel(channel).getClass());
+		assertEquals(StubChannel.class, channel.getClass());
 		assertEquals("testChannel", channel.getName());
 		channel.send(MessageBuilder.fromPayload("").build());
 		assertTrue(appliedInterceptors.get(0) == interceptors.get(0));

org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java

@@ -40,7 +40,6 @@ import org.springframework.integration.bus.TestMessageBusStopInterceptor;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.channel.config.ChannelParserTests;
 import org.springframework.integration.endpoint.DefaultEndpoint;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.handler.TestHandlers;
@@ -135,8 +134,7 @@ public class MessageBusParserTests {
 		ApplicationContext context = new ClassPathXmlApplicationContext("messageBusWithChannelFactory.xml",
 				this.getClass());
 		((MessageChannel)context.getBean("defaultTypeChannel")).getName();
-		assertEquals(DirectChannel.class,
-				ChannelParserTests.extractProxifiedChannel(context.getBean("defaultTypeChannel")).getClass());
+		assertEquals(DirectChannel.class, context.getBean("defaultTypeChannel").getClass());
 		assertEquals(QueueChannel.class, context.getBean("specifiedTypeChannel").getClass());
 	}
 

org.springframework.integration/src/test/java/org/springframework/integration/config/messageBusWithChannelFactory.xml

@@ -7,9 +7,9 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
 
-	<message-bus channel-factory="directChannelFactory"/>
+	<message-bus/>
 
-	<beans:bean id="directChannelFactory" class="org.springframework.integration.channel.factory.DirectChannelFactory"/>	
+	<beans:bean id="channelFactory" class="org.springframework.integration.channel.factory.DirectChannelFactory"/>	
 
     <channel id="defaultTypeChannel"/>