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 4bc2ff7349..411ef9d7c2 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,11 +16,17 @@ 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; @@ -32,45 +38,57 @@ 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}. - * As a {@link FactoryBean}, this class is solely intended to be used within + * single {@link MessageBus} defined in the {@link ApplicationContext}. + * 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 { +public class DefaultChannelFactoryBean implements ApplicationContextAware, FactoryBean, BeanNameAware, InitializingBean { private volatile String beanName; - private volatile ChannelFactory channelFactory; - private volatile List interceptors; + private volatile ApplicationContext applicationContext; + + private volatile boolean initialized; + + private final Object initializationMonitor = new Object(); + + private volatile Object proxyBean; + + + public void setApplicationContext(ApplicationContext applicationContext) { + this.applicationContext = applicationContext; + } public void setBeanName(String beanName) { this.beanName = beanName; } - @SuppressWarnings("unchecked") - public void setApplicationContext(ApplicationContext applicationContext){ - Map map = applicationContext.getBeansOfType(MessageBus.class); - Assert.state(map.size() <= 1, "There is more than one MessageBus in the ApplicationContext"); - if (map.isEmpty()) { - this.channelFactory = new QueueChannelFactory(); - } - else { - this.channelFactory = ((MessageBus) map.values().iterator().next()).getChannelFactory(); - } - } - public void setInterceptors(List interceptors) { this.interceptors = interceptors; } + public void afterPropertiesSet() throws Exception { + synchronized (initializationMonitor) { + if (!initialized) { + this.proxyBean = Proxy.newProxyInstance( + getClass().getClassLoader(), + new Class[]{MessageChannel.class}, + new DefaultChannelInvocationHandler()); + this.initialized = true; + } + } + } + public Object getObject() throws Exception { - Assert.notNull(channelFactory, "ChannelFactory not set on this instance. Is this used within an ApplicationContext?"); - return channelFactory.getChannel(this.beanName, interceptors); + if (!initialized) { + afterPropertiesSet(); + } + return proxyBean; } public Class getObjectType() { @@ -81,4 +99,49 @@ public class DefaultChannelFactoryBean implements ApplicationContextAware, Facto 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/test/java/org/springframework/integration/channel/config/ChannelParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java index b80ff59bf7..2b0f876e08 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,6 +16,10 @@ 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; @@ -64,7 +68,9 @@ public class ChannelParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "channelParserTests.xml", this.getClass()); MessageChannel channel = (MessageChannel) context.getBean("queueChannelByDefault"); - assertEquals(QueueChannel.class, channel.getClass()); + //called to initialize the channel instance + channel.getName(); + assertEquals(QueueChannel.class, extractProxifiedChannel(channel).getClass()); } @Test @@ -207,4 +213,13 @@ public class ChannelParserTests { assertTrue(threwException); } + + 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 345eb0daf0..10d3eae73c 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.*; +import static org.junit.Assert.assertTrue; 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(channel instanceof StubChannel); + assertTrue(StubChannel.class.isAssignableFrom(ChannelParserTests.extractProxifiedChannel(channel).getClass())); } } 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 92aa6c97bc..cb5a838f0c 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 @@ -16,13 +16,13 @@ package org.springframework.integration.channel.factory; +import java.util.ArrayList; +import java.util.List; + import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; -import java.util.ArrayList; -import java.util.List; - import org.junit.Before; import org.junit.Test; @@ -33,13 +33,17 @@ 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.MessageChannel; 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.dispatcher.DirectChannel; import org.springframework.integration.dispatcher.DirectChannelFactory; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageBuilder; /** * @author Marius Bogoevici @@ -49,6 +53,7 @@ public class ChannelFactoryTests { private final ArrayList interceptors = new ArrayList(); + private final ArrayList appliedInterceptors = new ArrayList(); @Before public void initInterceptorsList() { @@ -112,12 +117,15 @@ public class ChannelFactoryTests { channelFactoryBean.setBeanName("testChannel"); channelFactoryBean.setApplicationContext(applicationContext); channelFactoryBean.setInterceptors(interceptors); - StubChannel channel = (StubChannel) channelFactoryBean.getObject(); + MessageChannel channel = (MessageChannel) channelFactoryBean.getObject(); + channel.getName(); + assertEquals(StubChannel.class, ChannelParserTests.extractProxifiedChannel(channel).getClass()); assertEquals("testChannel", channel.getName()); - assertInterceptors(channel); + channel.send(MessageBuilder.fromPayload("").build()); + assertTrue(appliedInterceptors.get(0) == interceptors.get(0)); + assertTrue(appliedInterceptors.get(1) == interceptors.get(1)); } - private void genericChannelFactoryTests(ChannelFactory channelFactory, Class expectedChannelClass) { assertNotNull(interceptors); AbstractMessageChannel channel = (AbstractMessageChannel) channelFactory.getChannel("testChannel", interceptors); @@ -131,12 +139,18 @@ public class ChannelFactoryTests { Object interceptorsWrapper = new DirectFieldAccessor(channel).getPropertyValue("interceptors"); List interceptors = (List) new DirectFieldAccessor(interceptorsWrapper).getPropertyValue("interceptors"); - assertTrue(interceptors.get(0) == interceptors.get(0)); - assertTrue(interceptors.get(1) == interceptors.get(1)); + assertTrue(interceptors.get(0) == this.interceptors.get(0)); + assertTrue(interceptors.get(1) == this.interceptors.get(1)); } - private static class TestChannelInterceptor extends ChannelInterceptorAdapter { + private class TestChannelInterceptor extends ChannelInterceptorAdapter { + + public Message preSend(Message message, MessageChannel channel) { + appliedInterceptors.add(this); + return message; + } + } } 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 b67d201d57..3fc7bbeddb 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 @@ -37,7 +37,9 @@ import org.springframework.integration.bus.MessageBusInterceptorTests; import org.springframework.integration.bus.TestMessageBusAwareImpl; import org.springframework.integration.bus.TestMessageBusStartInterceptor; import org.springframework.integration.bus.TestMessageBusStopInterceptor; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.channel.config.ChannelParserTests; import org.springframework.integration.dispatcher.DirectChannel; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.scheduling.SimpleTaskScheduler; @@ -137,7 +139,9 @@ public class MessageBusParserTests { public void testMessageBusWithChannelFactory() { ApplicationContext context = new ClassPathXmlApplicationContext("messageBusWithChannelFactory.xml", this.getClass()); - assertEquals(DirectChannel.class, context.getBean("defaultTypeChannel").getClass()); + ((MessageChannel)context.getBean("defaultTypeChannel")).getName(); + assertEquals(DirectChannel.class, + ChannelParserTests.extractProxifiedChannel(context.getBean("defaultTypeChannel")).getClass()); assertEquals(QueueChannel.class, context.getBean("specifiedTypeChannel").getClass()); }