Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1176: added scope= attribute to channels and removed thread-local-channel

Browse Source
This commit is contained in:
David Syer
2010-06-17 11:07:37 +00:00
parent 49558ea82f
commit 8df7265abb
14 changed files with 169 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
spring-integration-core/src/main/java/org/springframework/integration/channel/ThreadLocalChannel.java
View File

@@ -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
* <code>null</code> 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<Queue<Message<?>>> {
@Override
protected Queue<Message<?>> initialValue() {
return new LinkedBlockingQueue<Message<?>>();
}
}
}

23
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java
View File

@@ -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);
}
/**

86
spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java
View File

@@ -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);
}
}

1
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java
View File

@@ -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());

3
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java
View File

@@ -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 &lt;channel&gt; element.

37
spring-integration-core/src/main/java/org/springframework/integration/config/xml/ThreadLocalChannelParser.java
View File

@@ -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 &lt;thread-local-channel&gt; 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");
}
}

24
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
View File

@@ -325,29 +325,6 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="thread-local-channel">
<xsd:annotation>
<xsd:documentation>
Defines a channel that maintains its Messages on a thread-bound queue.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports type="org.springframework.integration.core.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="channelType">
<xsd:sequence>
<xsd:element name="interceptors" type="channelInterceptorsType"
minOccurs="0" maxOccurs="1" />
</xsd:sequence>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="channelType">
<xsd:annotation>
<xsd:documentation>
@@ -355,6 +332,7 @@
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:ID" use="required" />
<xsd:attribute name="scope" type="xsd:string"/>
<xsd:attribute name="datatype" type="xsd:string">
<xsd:annotation>
<xsd:documentation>

8
spring-integration-core/src/test/java/log4j.properties Normal file
View File

@@ -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

5
spring-integration-core/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java
View File

@@ -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

118
spring-integration-core/src/test/java/org/springframework/integration/channel/ThreadLocalChannelTests.java
View File

@@ -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<Message<?>> receivedMessages = new ArrayList<Message<?>>();
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<Object> otherThreadResults = new ArrayList<Object>();
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));
}
}

33
spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests-context.xml
View File

@@ -1,20 +1,29 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans" xmlns:aop="http://www.springframework.org/schema/aop"
xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<thread-local-channel id="simpleChannel"/>
<bean class="org.springframework.beans.factory.config.CustomScopeConfigurer" xmlns="http://www.springframework.org/schema/beans">
<property name="scopes">
<map>
<entry key="thread" value="org.springframework.context.support.SimpleThreadScope" />
</map>
</property>
</bean>
<thread-local-channel id="channelWithInterceptor">
<channel id="simpleChannel" scope="thread">
<queue />
</channel>
<channel id="channelWithInterceptor" scope="thread">
<queue />
<interceptors>
<beans:ref bean="interceptor"/>
<beans:ref bean="interceptor" />
</interceptors>
</thread-local-channel>
</channel>
<beans:bean id="interceptor" class="org.springframework.integration.config.TestChannelInterceptor"/>
<beans:bean id="interceptor" class="org.springframework.integration.config.TestChannelInterceptor" />
</beans:beans>

75
spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests.java
View File

@@ -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<Object> otherThreadResults = new ArrayList<Object>();
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());
}
}

2
spring-integration-stream/pom.xml
View File

@@ -15,7 +15,7 @@
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<artifactId>spring-context</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>

4
spring-integration-test/pom.xml
View File

@@ -32,6 +32,10 @@
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
</dependency>
<!-- test-scoped dependencies -->
<dependency>
<groupId>org.easymock</groupId>