diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ThreadLocalChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ThreadLocalChannel.java deleted file mode 100644 index 3f45f464f4..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ThreadLocalChannel.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * Copyright 2002-2010 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.channel; - -import java.util.Queue; -import java.util.concurrent.LinkedBlockingQueue; - -import org.springframework.integration.core.Message; - -/** - * A channel implementation that stores messages in a thread-bound queue. In - * other words, send() will put a message at the tail of the queue for the - * current thread, and receive() will retrieve a message from the head of the - * queue. Since, by definition, only one thread will interact with the queue - * at a time, the timeout values on send and receive have no effect. If there - * are no Messages in the queue, the receive operations will return a - * null value immediately, regardless of any timeout value. - * - * @author Dave Syer - * @author Mark Fisher - */ -public class ThreadLocalChannel extends AbstractPollableChannel { - - private final ThreadLocalMessageHolder messageHolder = new ThreadLocalMessageHolder(); - - - @Override - protected boolean doSend(Message message, long timeout) { - if (message == null) { - return false; - } - return messageHolder.get().add(message); - } - - @Override - protected Message doReceive(long timeout) { - return messageHolder.get().poll(); - } - - - /** - * The thread-bound Queue. - */ - private static class ThreadLocalMessageHolder extends ThreadLocal>> { - - @Override - protected Queue> initialValue() { - return new LinkedBlockingQueue>(); - } - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java index 073893c780..9bf28e82e3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java @@ -18,8 +18,11 @@ package org.springframework.integration.config.xml; import org.w3c.dom.Element; +import org.springframework.aop.scope.ScopedProxyUtils; +import org.springframework.beans.factory.config.BeanDefinitionHolder; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; @@ -31,6 +34,7 @@ import org.springframework.util.xml.DomUtils; * Base class for channel parsers. * * @author Mark Fisher + * @author Dave Syer */ public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser { @@ -60,7 +64,24 @@ public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser builder.addPropertyValue("datatypes", datatypes); } builder.addPropertyValue("interceptors", interceptors); - return builder.getBeanDefinition(); + AbstractBeanDefinition beanDefinition = builder.getBeanDefinition(); + String scopeAttr = element.getAttribute("scope"); + if (StringUtils.hasText(scopeAttr)) { + builder.setScope(scopeAttr); + } + return beanDefinition; + } + + /* (non-Javadoc) + * @see org.springframework.beans.factory.xml.AbstractBeanDefinitionParser#registerBeanDefinition(org.springframework.beans.factory.config.BeanDefinitionHolder, org.springframework.beans.factory.support.BeanDefinitionRegistry) + */ + @Override + protected void registerBeanDefinition(BeanDefinitionHolder definition, BeanDefinitionRegistry registry) { + String scope = definition.getBeanDefinition().getScope(); + if (!AbstractBeanDefinition.SCOPE_DEFAULT.equals(scope) && !AbstractBeanDefinition.SCOPE_SINGLETON.equals(scope) && !AbstractBeanDefinition.SCOPE_PROTOTYPE.equals(scope)) { + definition = ScopedProxyUtils.createScopedProxy(definition, registry, false); + } + super.registerBeanDefinition(definition, registry); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java index 3dcd99f305..19c5f9cd56 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java @@ -20,7 +20,6 @@ import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.BeanDefinitionHolder; @@ -35,10 +34,9 @@ import org.springframework.integration.channel.NullChannel; import org.springframework.integration.context.IntegrationContextUtils; /** - * A {@link BeanFactoryPostProcessor} implementation that provides default - * beans for the error handling and task scheduling if those beans have not - * already been explicitly defined within the registry. It also registers a - * single null channel with the bean name "nullChannel". + * A {@link BeanFactoryPostProcessor} implementation that provides default beans for the error handling and task + * scheduling if those beans have not already been explicitly defined within the registry. It also registers a single + * null channel with the bean name "nullChannel". * * @author Mark Fisher * @author Oleg Zhurakousky @@ -49,7 +47,6 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce private Log logger = LogFactory.getLog(this.getClass()); - public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { if (beanFactory instanceof BeanDefinitionRegistry) { BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; @@ -65,78 +62,85 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce } /** - * Register a null channel in the given BeanDefinitionRegistry. The bean name is - * defined by the constant {@link IntegrationContextUtils#NULL_CHANNEL_BEAN_NAME}. + * Register a null channel in the given BeanDefinitionRegistry. The bean name is defined by the constant + * {@link IntegrationContextUtils#NULL_CHANNEL_BEAN_NAME}. */ private void registerNullChannel(BeanDefinitionRegistry registry) { if (registry.isBeanNameInUse(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME)) { BeanDefinition bDef = registry.getBeanDefinition(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME); - if (bDef.getBeanClassName().equals(NullChannel.class.getName())){ + if (bDef.getBeanClassName().equals(NullChannel.class.getName())) { return; - } else { - throw new IllegalStateException("The bean name '" + - IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME + "' is reserved."); } - } else { + else { + throw new IllegalStateException("The bean name '" + IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME + + "' is reserved."); + } + } + else { RootBeanDefinition nullChannelDef = new RootBeanDefinition(); nullChannelDef.setBeanClassName(IntegrationNamespaceUtils.BASE_PACKAGE + ".channel.NullChannel"); - BeanDefinitionHolder nullChannelHolder = new BeanDefinitionHolder( - nullChannelDef, IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME); + BeanDefinitionHolder nullChannelHolder = new BeanDefinitionHolder(nullChannelDef, + IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME); BeanDefinitionReaderUtils.registerBeanDefinition(nullChannelHolder, registry); } } /** - * Register an error channel in the given BeanDefinitionRegistry if not yet present. - * The bean name for which this is checking is defined by the constant - * {@link IntegrationContextUtils#ERROR_CHANNEL_BEAN_NAME}. + * Register an error channel in the given BeanDefinitionRegistry if not yet present. The bean name for which this is + * checking is defined by the constant {@link IntegrationContextUtils#ERROR_CHANNEL_BEAN_NAME}. */ private void registerErrorChannelIfNecessary(BeanDefinitionRegistry registry) { if (!registry.isBeanNameInUse(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) { if (logger.isInfoEnabled()) { - logger.info("No bean named '" + IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME + - "' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created."); + logger + .info("No bean named '" + + IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME + + "' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created."); } RootBeanDefinition errorChannelDef = new RootBeanDefinition(); - errorChannelDef.setBeanClassName(IntegrationNamespaceUtils.BASE_PACKAGE + ".channel.PublishSubscribeChannel"); - BeanDefinitionHolder errorChannelHolder = new BeanDefinitionHolder( - errorChannelDef, IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); + errorChannelDef.setBeanClassName(IntegrationNamespaceUtils.BASE_PACKAGE + + ".channel.PublishSubscribeChannel"); + BeanDefinitionHolder errorChannelHolder = new BeanDefinitionHolder(errorChannelDef, + IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); BeanDefinitionReaderUtils.registerBeanDefinition(errorChannelHolder, registry); - BeanDefinitionBuilder loggingHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".handler.LoggingHandler"); + BeanDefinitionBuilder loggingHandlerBuilder = BeanDefinitionBuilder + .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".handler.LoggingHandler"); loggingHandlerBuilder.addConstructorArgValue("ERROR"); - BeanDefinitionBuilder loggingEndpointBuilder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".endpoint.EventDrivenConsumer"); + BeanDefinitionBuilder loggingEndpointBuilder = BeanDefinitionBuilder + .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".endpoint.EventDrivenConsumer"); loggingEndpointBuilder.addConstructorArgReference(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); loggingEndpointBuilder.addConstructorArgValue(loggingHandlerBuilder.getBeanDefinition()); - BeanComponentDefinition componentDefinition = new BeanComponentDefinition( - loggingEndpointBuilder.getBeanDefinition(), ERROR_LOGGER_BEAN_NAME); + BeanComponentDefinition componentDefinition = new BeanComponentDefinition(loggingEndpointBuilder + .getBeanDefinition(), ERROR_LOGGER_BEAN_NAME); BeanDefinitionReaderUtils.registerBeanDefinition(componentDefinition, registry); } } /** - * Register a TaskScheduler in the given BeanDefinitionRegistry if not yet present. - * The bean name for which this is checking is defined by the constant - * {@link IntegrationContextUtils#TASK_SCHEDULER_BEAN_NAME}. + * Register a TaskScheduler in the given BeanDefinitionRegistry if not yet present. The bean name for which this is + * checking is defined by the constant {@link IntegrationContextUtils#TASK_SCHEDULER_BEAN_NAME}. */ private void registerTaskSchedulerIfNecessary(BeanDefinitionRegistry registry) { if (!registry.isBeanNameInUse(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)) { if (logger.isInfoEnabled()) { - logger.info("No bean named '" + IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME + - "' has been explicitly defined. Therefore, a default ThreadPoolTaskScheduler will be created."); + logger + .info("No bean named '" + + IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME + + "' has been explicitly defined. Therefore, a default ThreadPoolTaskScheduler will be created."); } - BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler"); + BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder + .genericBeanDefinition("org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler"); schedulerBuilder.addPropertyValue("poolSize", 10); schedulerBuilder.addPropertyValue("threadNamePrefix", "task-scheduler-"); schedulerBuilder.addPropertyValue("rejectedExecutionHandler", new CallerRunsPolicy()); - BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".channel.MessagePublishingErrorHandler"); - errorHandlerBuilder.addPropertyReference("defaultErrorChannel", IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); + BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder + .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + + ".channel.MessagePublishingErrorHandler"); + errorHandlerBuilder.addPropertyReference("defaultErrorChannel", + IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); schedulerBuilder.addPropertyValue("errorHandler", errorHandlerBuilder.getBeanDefinition()); - BeanComponentDefinition schedulerComponent = new BeanComponentDefinition( - schedulerBuilder.getBeanDefinition(), IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); + BeanComponentDefinition schedulerComponent = new BeanComponentDefinition(schedulerBuilder + .getBeanDefinition(), IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME); BeanDefinitionReaderUtils.registerBeanDefinition(schedulerComponent, registry); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java index 20474c3acc..b472591759 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java @@ -27,7 +27,6 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan public void init() { registerBeanDefinitionParser("channel", new PointToPointChannelParser()); - registerBeanDefinitionParser("thread-local-channel", new ThreadLocalChannelParser()); registerBeanDefinitionParser("publish-subscribe-channel", new PublishSubscribeChannelParser()); registerBeanDefinitionParser("service-activator", new ServiceActivatorParser()); registerBeanDefinitionParser("transformer", new TransformerParser()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java index 5d9efc2514..636024c55c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java @@ -18,13 +18,12 @@ package org.springframework.integration.config.xml; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.w3c.dom.Element; - import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Element; /** * Parser for the <channel> element. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ThreadLocalChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ThreadLocalChannelParser.java deleted file mode 100644 index 6ea55a3c48..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ThreadLocalChannelParser.java +++ /dev/null @@ -1,37 +0,0 @@ -/* - * Copyright 2002-2008 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.config.xml; - -import org.w3c.dom.Element; - -import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.xml.ParserContext; - -/** - * Parser for the <thread-local-channel> element. - * - * @author Mark Fisher - */ -public class ThreadLocalChannelParser extends AbstractChannelParser { - - @Override - protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) { - return BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".channel.ThreadLocalChannel"); - } - -} diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index 9f279fb7c7..45f5edd5e1 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -325,29 +325,6 @@ - - - - Defines a channel that maintains its Messages on a thread-bound queue. - - - - - - - - - - - - - - - - - - @@ -355,6 +332,7 @@ + diff --git a/spring-integration-core/src/test/java/log4j.properties b/spring-integration-core/src/test/java/log4j.properties new file mode 100644 index 0000000000..941cbe4822 --- /dev/null +++ b/spring-integration-core/src/test/java/log4j.properties @@ -0,0 +1,8 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%c{1}: %m%n + +log4j.category.org.springframework.integration=WARN +log4j.category.org.springframework.integration.file=WARN diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java index 7adc972568..6770fbc3c2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java @@ -20,12 +20,11 @@ import static org.junit.Assert.assertEquals; import org.junit.Before; import org.junit.Test; - import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.channel.ThreadLocalChannel; import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.core.Message; @@ -46,7 +45,7 @@ public class DirectChannelSubscriptionTests { private DirectChannel sourceChannel = new DirectChannel(); - private ThreadLocalChannel targetChannel = new ThreadLocalChannel(); + private PollableChannel targetChannel = new QueueChannel(); @Before diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/ThreadLocalChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/ThreadLocalChannelTests.java deleted file mode 100644 index d3cd1d1dd0..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/ThreadLocalChannelTests.java +++ /dev/null @@ -1,118 +0,0 @@ -/* - * Copyright 2002-2010 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.channel; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; - -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executor; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; - -import org.junit.Before; -import org.junit.Test; - -import org.springframework.integration.core.Message; -import org.springframework.integration.message.StringMessage; - -/** - * @author Mark Fisher - */ -public class ThreadLocalChannelTests { - - @Before - public void clearThreadLocalQueue() { - ThreadLocalChannel channel = new ThreadLocalChannel(); - Message result = null; - do { - result = channel.receive(0); - } while (result != null); - } - - - @Test - public void testSendAndReceive() { - ThreadLocalChannel channel = new ThreadLocalChannel(); - StringMessage message = new StringMessage("test"); - assertNull(channel.receive()); - assertTrue(channel.send(message)); - Message response = channel.receive(); - assertNotNull(response); - assertEquals(response, message); - assertNull(channel.receive()); - } - - @Test - public void testSendAndReceiveMultipleMessages() { - ThreadLocalChannel channel = new ThreadLocalChannel(); - StringMessage message1 = new StringMessage("test1"); - StringMessage message2 = new StringMessage("test2"); - assertNull(channel.receive()); - assertTrue(channel.send(message1)); - assertTrue(channel.send(message2)); - List> receivedMessages = new ArrayList>(); - receivedMessages.add(channel.receive(0)); - receivedMessages.add(channel.receive(0)); - assertEquals(2, receivedMessages.size()); - assertEquals(message1, receivedMessages.get(0)); - assertEquals(message2, receivedMessages.get(1)); - assertNull(channel.receive()); - } - - @Test - public void multipleThreadLocalChannels() throws Exception { - final ThreadLocalChannel channel1 = new ThreadLocalChannel(); - final ThreadLocalChannel channel2 = new ThreadLocalChannel(); - channel1.send(new StringMessage("test-1.1")); - channel1.send(new StringMessage("test-1.2")); - channel1.send(new StringMessage("test-1.3")); - channel2.send(new StringMessage("test-2.1")); - channel2.send(new StringMessage("test-2.2")); - Executor otherThreadExecutor = Executors.newSingleThreadExecutor(); - final List otherThreadResults = new ArrayList(); - final CountDownLatch latch = new CountDownLatch(2); - otherThreadExecutor.execute(new Runnable() { - public void run() { - otherThreadResults.add(channel1.receive(0)); - latch.countDown(); - } - }); - otherThreadExecutor.execute(new Runnable() { - public void run() { - otherThreadResults.add(channel2.receive(0)); - latch.countDown(); - } - }); - latch.await(1, TimeUnit.SECONDS); - assertEquals(2, otherThreadResults.size()); - assertNull(otherThreadResults.get(0)); - assertNull(otherThreadResults.get(1)); - assertEquals("test-1.1", channel1.receive(0).getPayload()); - assertEquals("test-1.2", channel1.receive(0).getPayload()); - assertEquals("test-1.3", channel1.receive(0).getPayload()); - assertNull(channel1.receive(0)); - assertEquals("test-2.1", channel2.receive(0).getPayload()); - assertEquals("test-2.2", channel2.receive(0).getPayload()); - assertNull(channel2.receive(0)); - } - -} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests-context.xml index 96954094f7..55327dbefa 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests-context.xml @@ -1,20 +1,29 @@ - + - + + + + + + + - + + + + + + - + - + - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests.java index 7cfd2e762e..cfb16823c3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests.java @@ -17,46 +17,99 @@ package org.springframework.integration.channel.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.junit.Test; import org.junit.runner.RunWith; - import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.integration.channel.ThreadLocalChannel; +import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.config.TestChannelInterceptor; -import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.StringMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Mark Fisher + * @author Dave Syer */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration public class ThreadLocalChannelParserTests { @Autowired @Qualifier("simpleChannel") - private MessageChannel simpleChannel; + private PollableChannel simpleChannel; @Autowired @Qualifier("channelWithInterceptor") - private MessageChannel channelWithInterceptor; + private PollableChannel channelWithInterceptor; @Autowired private TestChannelInterceptor interceptor; - @Test - public void checkType() { - assertEquals(ThreadLocalChannel.class, simpleChannel.getClass()); + public void testSendInAnotherThread() throws Exception { + simpleChannel.send(new StringMessage("test")); + Executor otherThreadExecutor = Executors.newSingleThreadExecutor(); + final CountDownLatch latch = new CountDownLatch(1); + otherThreadExecutor.execute(new Runnable() { + public void run() { + simpleChannel.send(new StringMessage("crap")); + latch.countDown(); + } + }); + latch.await(1, TimeUnit.SECONDS); + assertEquals("test", simpleChannel.receive(10).getPayload()); + // Message sent on another thread is not collected here + assertEquals(null, simpleChannel.receive(10)); } @Test - public void verifyInterceptor() { - assertEquals(0, interceptor.getSendCount()); + public void testReceiveInAnotherThread() throws Exception { + simpleChannel.send(new StringMessage("test-1.1")); + simpleChannel.send(new StringMessage("test-1.2")); + simpleChannel.send(new StringMessage("test-1.3")); + channelWithInterceptor.send(new StringMessage("test-2.1")); + channelWithInterceptor.send(new StringMessage("test-2.2")); + Executor otherThreadExecutor = Executors.newSingleThreadExecutor(); + final List otherThreadResults = new ArrayList(); + final CountDownLatch latch = new CountDownLatch(2); + otherThreadExecutor.execute(new Runnable() { + public void run() { + otherThreadResults.add(simpleChannel.receive(0)); + latch.countDown(); + } + }); + otherThreadExecutor.execute(new Runnable() { + public void run() { + otherThreadResults.add(channelWithInterceptor.receive(0)); + latch.countDown(); + } + }); + latch.await(1, TimeUnit.SECONDS); + assertEquals(2, otherThreadResults.size()); + assertNull(otherThreadResults.get(0)); + assertNull(otherThreadResults.get(1)); + assertEquals("test-1.1", simpleChannel.receive(0).getPayload()); + assertEquals("test-1.2", simpleChannel.receive(0).getPayload()); + assertEquals("test-1.3", simpleChannel.receive(0).getPayload()); + assertNull(simpleChannel.receive(0)); + assertEquals("test-2.1", channelWithInterceptor.receive(0).getPayload()); + assertEquals("test-2.2", channelWithInterceptor.receive(0).getPayload()); + assertNull(channelWithInterceptor.receive(0)); + } + + @Test + public void testInterceptor() { + int before = interceptor.getSendCount(); channelWithInterceptor.send(new StringMessage("test")); - assertEquals(1, interceptor.getSendCount()); + assertEquals(before+1, interceptor.getSendCount()); } } diff --git a/spring-integration-stream/pom.xml b/spring-integration-stream/pom.xml index b1d4173094..12c1174a89 100644 --- a/spring-integration-stream/pom.xml +++ b/spring-integration-stream/pom.xml @@ -15,7 +15,7 @@ org.springframework - spring-core + spring-context org.springframework.integration diff --git a/spring-integration-test/pom.xml b/spring-integration-test/pom.xml index a4ffd3d2f4..29eb61cf05 100644 --- a/spring-integration-test/pom.xml +++ b/spring-integration-test/pom.xml @@ -32,6 +32,10 @@ org.springframework.integration spring-integration-core + + org.springframework + spring-context + org.easymock