diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultChannelFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultChannelFactoryBean.java index 0f02581dff..0d8951fc81 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultChannelFactoryBean.java +++ b/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). + *

* 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 targetChannelReference = new AtomicReference(); - - - 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(); - } - } - } - - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java index 59ab0697ac..8d78ae6ac5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java +++ b/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"); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java index 4fee6c12f6..ae4a12b585 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/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(); - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index a8371efc7d..4efb75aa40 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -33,7 +33,6 @@ - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java index 59f37c3f0b..99bcfee86b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java +++ b/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 reference = - (AtomicReference) handlerAccessor.getPropertyValue("targetChannelReference"); - return reference.get(); - } - } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/DefaultChannelParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/DefaultChannelParserTests.java index 10d3eae73c..34155b13b9 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/DefaultChannelParserTests.java +++ b/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()); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/defaultChannelParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/defaultChannelParserTests.xml index 4dedb67b75..26a6ec7ef1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/defaultChannelParserTests.xml +++ b/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"> - - - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java index 199946c814..fce44629bb 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java +++ b/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)); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java index b84bd59d41..89dae5b5e5 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java +++ b/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()); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/messageBusWithChannelFactory.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/messageBusWithChannelFactory.xml index 523b5d3b1f..ba147a89f6 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/messageBusWithChannelFactory.xml +++ b/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"> - + - +