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

INT-3308 SubscribableChannel Optimization

Browse Source 
JIRA: https://jira.springsource.org/browse/INT-3308

INT-3308 Move Optimization to AbstractDispatcher

- avoids the need for `canShortCutDispatcher()`
- optimizes all framework channels, including AMQP, JMS
- slight additional performance improvement

INT-3308 FinalSingleHandlerChannel

This is still a work in process.

TODO:Parser, parser tests.

See the DirectChannelTests for perforance results.

INT-3308 'Final' Channel Parser Changes, Tests

INT-3308 Polishing; PR Comments

INT-3308 Polishing

* Rename 'final' attribute to 'fixed-subscriber'
* Rename `BasicSingleFixedSubscriberChannel` to `FixedSubscriberChannel`

INT-3308 Polishing

* Rename test cases to match channel
* Fix parser error messages
* Add more exclusive attribute/element tests

INT-3308 More Polishing

* Remove remaining textual references to 'final'
* WARN log if the single subscriber has auto-startup="false"
* Handle the case when a fixed subscriber channel is declared after its handler
* Support the use of a fixed subscriber channel for elements created by an AbstractOutboundChannelAdapterParser
This commit is contained in:
Gary Russell
2014-03-06 13:27:38 +02:00
committed by Artem Bilan
parent c7fb83259f
commit 776705e6c4
29 changed files with 915 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -53,7 +53,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
private volatile boolean shouldTrack = false;
private volatile Class<?>[] datatypes = new Class<?>[] { Object.class };
private volatile Class<?>[] datatypes = new Class<?>[0];
private final ChannelInterceptorList interceptors = new ChannelInterceptorList();
@@ -86,7 +86,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
*/
public void setDatatypes(Class<?>... datatypes) {
this.datatypes = (datatypes != null && datatypes.length > 0)
? datatypes : new Class<?>[] { Object.class };
? datatypes : new Class<?>[0];
}
/**
@@ -245,12 +245,14 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
if (this.shouldTrack) {
message = MessageHistory.write(message, this);
}
message = this.convertPayloadIfNecessary(message);
message = this.interceptors.preSend(message, this);
if (message == null) {
return false;
}
try {
if (this.datatypes.length > 0) {
message = this.convertPayloadIfNecessary(message);
}
message = this.interceptors.preSend(message, this);
if (message == null) {
return false;
}
boolean sent = this.doSend(message, timeout);
this.interceptors.postSend(message, this, sent);
return sent;
@@ -332,10 +334,12 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
if (logger.isDebugEnabled()) {
logger.debug("preSend on channel '" + channel + "', message: " + message);
}
for (ChannelInterceptor interceptor : interceptors) {
message = interceptor.preSend(message, channel);
if (message == null) {
return null;
if (this.interceptors.size() > 0) {
for (ChannelInterceptor interceptor : this.interceptors) {
message = interceptor.preSend(message, channel);
if (message == null) {
return null;
}
}
}
return message;
@@ -345,8 +349,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
if (logger.isDebugEnabled()) {
logger.debug("postSend (sent=" + sent + ") on channel '" + channel + "', message: " + message);
}
for (ChannelInterceptor interceptor : interceptors) {
interceptor.postSend(message, channel, sent);
if (this.interceptors.size() > 0) {
for (ChannelInterceptor interceptor : interceptors) {
interceptor.postSend(message, channel, sent);
}
}
}
@@ -354,9 +360,11 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
if (logger.isTraceEnabled()) {
logger.trace("preReceive on channel '" + channel + "'");
}
for (ChannelInterceptor interceptor : interceptors) {
if (!interceptor.preReceive(channel)) {
return false;
if (this.interceptors.size() > 0) {
for (ChannelInterceptor interceptor : interceptors) {
if (!interceptor.preReceive(channel)) {
return false;
}
}
}
return true;
@@ -369,10 +377,12 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
else if (logger.isTraceEnabled()) {
logger.trace("postReceive on channel '" + channel + "', message is null");
}
for (ChannelInterceptor interceptor : interceptors) {
message = interceptor.postReceive(message, channel);
if (message == null) {
return null;
if (this.interceptors.size() > 0) {
for (ChannelInterceptor interceptor : interceptors) {
message = interceptor.postReceive(message, channel);
if (message == null) {
return null;
}
}
}
return message;

117
spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java Normal file
View File

@@ -0,0 +1,117 @@
/*
* Copyright 2014 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 org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
/**
* Specialized {@link SubscribableChannel} for a single final subscriber set up during bean instantiation (unlike
* other {@link SubscribableChannel}s where the {@link MessageHandler} is subscribed when the endpoint
* is started). This channel does not support interceptors or data types.
* <p>
* <b>Note: Stopping ({@link #unsubscribe(MessageHandler)}) the subscribed ({@link MessageHandler}) has no effect.
*
* @author Gary Russell
* @since 4.0
*
*/
public final class FixedSubscriberChannel implements SubscribableChannel, BeanNameAware, NamedComponent {
private final Log logger = LogFactory.getLog(FixedSubscriberChannel.class);
private final MessageHandler handler;
private volatile String beanName;
public FixedSubscriberChannel() {
throw new IllegalArgumentException("Cannot instantiate a " + this.getClass().getSimpleName()
+ " without a MessageHandler.");
}
public FixedSubscriberChannel(MessageHandler handler) {
this.handler = handler;
}
@Override
public void setBeanName(String name) {
this.beanName = name;
}
@Override
public boolean send(Message<?> message) {
return this.send(message, 0);
}
@Override
public boolean send(Message<?> message, long timeout) {
try {
this.handler.handleMessage(message);
return true;
}
catch (Exception e) {
RuntimeException runtimeException = (e instanceof RuntimeException)
? (RuntimeException) e
: new MessageDeliveryException(message,
this.getComponentName() + " failed to deliver Message.", e);
if (e instanceof MessagingException &&
((MessagingException) e).getFailedMessage() == null) {
runtimeException = new MessagingException(message, e);
}
throw runtimeException;
}
}
@Override
public boolean subscribe(MessageHandler handler) {
if (handler != this.handler && logger.isDebugEnabled()) {
logger.debug(this.getComponentName() + ": cannot be subscribed to (it has a fixed single subscriber).");
}
return false;
}
@Override
public boolean unsubscribe(MessageHandler handler) {
if (logger.isDebugEnabled()) {
logger.debug(this.getComponentName() + ": cannot be unsubscribed from (it has a fixed single subscriber).");
}
return false;
}
@Override
public String getComponentType() {
return "Fixed Subscriber Channel";
}
@Override
public String getComponentName() {
if (this.beanName != null) {
return this.beanName;
}
else {
return "Unnamed fixed subscriber channel";
}
}
}

28
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -21,6 +21,7 @@ import java.util.List;
import org.aopalliance.aop.Advice;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
@@ -33,16 +34,17 @@ import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.SmartLifecycle;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
@@ -106,6 +108,7 @@ public class ConsumerEndpointFactoryBean
this.pollerMetadata = pollerMetadata;
}
@Override
public void setBeanClassLoader(ClassLoader classLoader) {
this.beanClassLoader = classLoader;
}
@@ -118,10 +121,12 @@ public class ConsumerEndpointFactoryBean
this.phase = phase;
}
@Override
public void setBeanName(String beanName) {
this.beanName = beanName;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
Assert.isInstanceOf(ConfigurableBeanFactory.class, beanFactory, "a ConfigurableBeanFactory is required");
this.beanFactory = (ConfigurableBeanFactory) beanFactory;
@@ -132,6 +137,7 @@ public class ConsumerEndpointFactoryBean
this.adviceChain = adviceChain;
}
@Override
public void afterPropertiesSet() throws Exception {
try {
if (!this.beanName.startsWith("org.springframework")) {
@@ -181,10 +187,12 @@ public class ConsumerEndpointFactoryBean
this.initializeEndpoint();
}
@Override
public boolean isSingleton() {
return true;
}
@Override
public AbstractEndpoint getObject() throws Exception {
if (!this.initialized) {
this.initializeEndpoint();
@@ -192,6 +200,7 @@ public class ConsumerEndpointFactoryBean
return this.endpoint;
}
@Override
public Class<?> getObjectType() {
if (this.endpoint == null) {
return AbstractEndpoint.class;
@@ -218,6 +227,9 @@ public class ConsumerEndpointFactoryBean
Assert.isNull(this.pollerMetadata, "A poller should not be specified for endpoint '" + this.beanName
+ "', since '" + channel + "' is a SubscribableChannel (not pollable).");
this.endpoint = new EventDrivenConsumer((SubscribableChannel) channel, this.handler);
if (logger.isWarnEnabled() && !this.autoStartup && channel instanceof FixedSubscriberChannel) {
logger.warn("'autoStartup=\"false\"' has no effect when using a FixedSubscriberChannel");
}
}
else if (channel instanceof PollableChannel) {
PollingConsumer pollingConsumer = new PollingConsumer((PollableChannel) channel, this.handler);
@@ -256,30 +268,36 @@ public class ConsumerEndpointFactoryBean
* SmartLifecycle implementation (delegates to the created endpoint)
*/
@Override
public boolean isAutoStartup() {
return (this.endpoint != null) ? this.endpoint.isAutoStartup() : true;
}
@Override
public int getPhase() {
return (this.endpoint != null) ? this.endpoint.getPhase() : 0;
}
@Override
public boolean isRunning() {
return (this.endpoint != null) ? this.endpoint.isRunning() : false;
}
@Override
public void start() {
if (this.endpoint != null) {
this.endpoint.start();
}
}
@Override
public void stop() {
if (this.endpoint != null) {
this.endpoint.stop();
}
}
@Override
public void stop(Runnable callback) {
if (this.endpoint != null) {
this.endpoint.stop(callback);

73
spring-integration-core/src/main/java/org/springframework/integration/config/FixedSubscriberChannelBeanFactoryPostProcessor.java Normal file
View File

@@ -0,0 +1,73 @@
/*
* Copyright 2014 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;
import java.util.Map;
import java.util.Map.Entry;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.config.ConstructorArgumentValues;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
/**
* Used to post process candidates for {@link FixedSubscriberChannel} {@link MessageHandler}s.
* @author Gary Russell
* @since 4.0
*
*/
public class FixedSubscriberChannelBeanFactoryPostProcessor implements BeanFactoryPostProcessor {
private final Map<String, String> candidateFixedChannelHandlerMap;
private FixedSubscriberChannelBeanFactoryPostProcessor(Map<String, String> candidateHandlers) {
this.candidateFixedChannelHandlerMap = candidateHandlers;
}
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
if (beanFactory instanceof BeanDefinitionRegistry) {
BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
if (this.candidateFixedChannelHandlerMap.size() > 0) {
for (Entry<String, String> entry : this.candidateFixedChannelHandlerMap.entrySet()) {
String handlerName = entry.getKey();
String channelName = entry.getValue();
BeanDefinition handlerBeanDefinition = null;
if (registry.containsBeanDefinition(handlerName)) {
handlerBeanDefinition = registry.getBeanDefinition(handlerName);
}
if (handlerBeanDefinition != null && registry.containsBeanDefinition(channelName)) {
BeanDefinition inputChannelDefinition = registry.getBeanDefinition(channelName);
if (FixedSubscriberChannel.class.getName().equals(inputChannelDefinition.getBeanClassName())) {
ConstructorArgumentValues constructorArgumentValues = inputChannelDefinition
.getConstructorArgumentValues();
Assert.isTrue(constructorArgumentValues.isEmpty(),
"Only one subscriber is allowed for a FixedSubscriberChannel.");
constructorArgumentValues.addGenericArgumentValue(handlerBeanDefinition);
}
}
}
}
}
}
}

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

@@ -26,6 +26,7 @@ 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;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
@@ -42,28 +43,41 @@ public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser
@Override
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = this.buildBeanDefinition(element, parserContext);
ManagedList interceptors = null;
Element interceptorsElement = DomUtils.getChildElementByTagName(element, "interceptors");
if (interceptorsElement != null) {
ChannelInterceptorParser interceptorParser = new ChannelInterceptorParser();
interceptors = interceptorParser.parseInterceptors(interceptorsElement, parserContext);
}
if (interceptors == null) {
interceptors = new ManagedList();
}
String datatypeAttr = element.getAttribute("datatype");
if (StringUtils.hasText(datatypeAttr)) {
builder.addPropertyValue("datatypes", datatypeAttr);
}
String messageConverter = element.getAttribute("message-converter");
if (StringUtils.hasText(messageConverter)) {
builder.addPropertyReference("messageConverter", messageConverter);
}
builder.addPropertyValue("interceptors", interceptors);
AbstractBeanDefinition beanDefinition = builder.getBeanDefinition();
String scopeAttr = element.getAttribute("scope");
if (StringUtils.hasText(scopeAttr)) {
builder.setScope(scopeAttr);
Element interceptorsElement = DomUtils.getChildElementByTagName(element, "interceptors");
String datatypeAttr = element.getAttribute("datatype");
String messageConverter = element.getAttribute("message-converter");
if (!FixedSubscriberChannel.class.getName().equals(builder.getBeanDefinition().getBeanClassName())) {
ManagedList interceptors = null;
if (interceptorsElement != null) {
ChannelInterceptorParser interceptorParser = new ChannelInterceptorParser();
interceptors = interceptorParser.parseInterceptors(interceptorsElement, parserContext);
}
if (interceptors == null) {
interceptors = new ManagedList();
}
if (StringUtils.hasText(datatypeAttr)) {
builder.addPropertyValue("datatypes", datatypeAttr);
}
if (StringUtils.hasText(messageConverter)) {
builder.addPropertyReference("messageConverter", messageConverter);
}
builder.addPropertyValue("interceptors", interceptors);
String scopeAttr = element.getAttribute("scope");
if (StringUtils.hasText(scopeAttr)) {
builder.setScope(scopeAttr);
}
}
else {
if (interceptorsElement != null) {
parserContext.getReaderContext().error("Cannot have interceptors when 'fixed-subscriber=\"true\"'", element);
}
if (StringUtils.hasText(datatypeAttr)) {
parserContext.getReaderContext().error("Cannot have 'datatype' when 'fixed-subscriber=\"true\"'", element);
}
if (StringUtils.hasText(messageConverter)) {
parserContext.getReaderContext().error("Cannot have 'message-converter' when 'fixed-subscriber=\"true\"'", element);
}
}
beanDefinition.setSource(parserContext.extractSource(element));
return beanDefinition;

2
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java
View File

@@ -141,6 +141,8 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME + "'", parserContext.getRegistry());
}
}
IntegrationNamespaceUtils.checkAndConfigureFixedSubscriberChannel(element, parserContext, inputChannelName,
handlerBeanName);
builder.addPropertyValue("inputChannelName", inputChannelName);
List<Element> pollerElementList = DomUtils.getChildElementsByTagName(element, "poller");

2
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractOutboundChannelAdapterParser.java
View File

@@ -63,6 +63,8 @@ public abstract class AbstractOutboundChannelAdapterParser extends AbstractChann
Element pollerElement = DomUtils.getChildElementByTagName(element, "poller");
BeanComponentDefinition handlerBeanComponentDefinition = this.doParseAndRegisterConsumer(element, parserContext);
builder.addPropertyReference("handler", handlerBeanComponentDefinition.getBeanName());
IntegrationNamespaceUtils.checkAndConfigureFixedSubscriberChannel(element, parserContext, channelName,
handlerBeanComponentDefinition.getBeanName());
if (pollerElement != null) {
if (!StringUtils.hasText(channelName)) {
parserContext.getReaderContext().error(

48
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -16,6 +16,7 @@ package org.springframework.integration.config.xml;
import static org.springframework.beans.factory.xml.AbstractBeanDefinitionParser.ID_ATTRIBUTE;
import java.util.List;
import java.util.Map;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
@@ -23,18 +24,25 @@ import org.w3c.dom.NodeList;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.config.ConstructorArgumentValues;
import org.springframework.beans.factory.config.ConstructorArgumentValues.ValueHolder;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.config.TypedStringValue;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
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.support.ManagedMap;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.core.Conventions;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.config.FixedSubscriberChannelBeanFactoryPostProcessor;
import org.springframework.integration.config.IntegrationConfigUtils;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.AbstractPollingEndpoint;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
@@ -500,4 +508,44 @@ public abstract class IntegrationNamespaceUtils {
return channelId;
}
@SuppressWarnings("unchecked")
public static void checkAndConfigureFixedSubscriberChannel(Element element, ParserContext parserContext,
String channelName, String handlerBeanName) {
BeanDefinitionRegistry registry = parserContext.getRegistry();
if (registry.containsBeanDefinition(channelName)) {
BeanDefinition inputChannelDefinition = registry.getBeanDefinition(channelName);
if (FixedSubscriberChannel.class.getName().equals(inputChannelDefinition.getBeanClassName())) {
ConstructorArgumentValues constructorArgumentValues = inputChannelDefinition
.getConstructorArgumentValues();
if (constructorArgumentValues.isEmpty()) {
constructorArgumentValues.addGenericArgumentValue(new RuntimeBeanReference(handlerBeanName));
}
else {
parserContext.getReaderContext().error("Only one subscriber is allowed for a FixedSubscriberChannel.",
element);
}
}
}
else {
BeanDefinition bfppd;
if (!registry.containsBeanDefinition(IntegrationContextUtils.INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME)) {
bfppd = new RootBeanDefinition(FixedSubscriberChannelBeanFactoryPostProcessor.class);
registry.registerBeanDefinition(IntegrationContextUtils.INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME, bfppd);
}
else {
bfppd = registry.getBeanDefinition(IntegrationContextUtils.INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME);
}
ManagedMap<String, String> candidates;
ValueHolder argumentValue = bfppd.getConstructorArgumentValues().getArgumentValue(0, Map.class);
if (argumentValue == null) {
candidates = new ManagedMap<String, String>();
bfppd.getConstructorArgumentValues().addIndexedArgumentValue(0, candidates);
}
else {
candidates = (ManagedMap<String, String>) argumentValue.getValue();
}
candidates.put(handlerBeanName, channelName);
}
}
}

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

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.channel.PriorityChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.RendezvousChannel;
@@ -45,6 +46,8 @@ public class PointToPointChannelParser extends AbstractChannelParser {
protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = null;
Element queueElement = null;
String fixedSubscriberChannel = element.getAttribute("fixed-subscriber");
boolean isFixedSubscriber = "true".equals(fixedSubscriberChannel.trim().toLowerCase());
// configure a queue-based channel if any queue sub-element is defined
if ((queueElement = DomUtils.getChildElementByTagName(element, "queue")) != null) {
@@ -87,14 +90,25 @@ public class PointToPointChannelParser extends AbstractChannelParser {
}
if (queueElement != null) {
if (isFixedSubscriber) {
parserContext.getReaderContext().error(
"The 'fixed-subscriber' attribute is not allowed when a <queue/> child element is present.",
element);
}
return builder;
}
if (dispatcherElement == null) {
// configure the default DirectChannel with a RoundRobinLoadBalancingStrategy
builder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
builder = BeanDefinitionBuilder.genericBeanDefinition(isFixedSubscriber ? FixedSubscriberChannel.class
: DirectChannel.class);
}
else {
if (isFixedSubscriber) {
parserContext.getReaderContext().error(
"The 'fixed-subscriber' attribute is not allowed when a <dispatcher/> child element is present.",
element);
}
// configure either an ExecutorChannel or DirectChannel based on existence of 'task-executor'
String taskExecutor = dispatcherElement.getAttribute("task-executor");
if (StringUtils.hasText(taskExecutor)) {
@@ -119,7 +133,7 @@ public class PointToPointChannelParser extends AbstractChannelParser {
builder.addConstructorArgValue(null);
}
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, dispatcherElement, "failover");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, dispatcherElement, "max-subscribers");
}

3
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
View File

@@ -33,6 +33,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Josh Long
* @author Artem Bilan
* @author Gary Russell
*/
public abstract class IntegrationContextUtils {
@@ -70,6 +71,8 @@ public abstract class IntegrationContextUtils {
public static final String INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME = "datatypeChannelMessageConverter";
public static final String INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME = "fixedSubscriberChannelBeanFactoryPostProcessor";
/**
* @param beanFactory BeanFactory for lookup, must not be null.
* @return The {@link MetadataStore} bean whose name is "metadataStore".

53
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java
View File

@@ -21,7 +21,10 @@ import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
/**
@@ -48,6 +51,8 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
private final OrderedAwareCopyOnWriteArraySet<MessageHandler> handlers =
new OrderedAwareCopyOnWriteArraySet<MessageHandler>();
private volatile MessageHandler theOneHandler;
/**
* Set the maximum subscribers allowed by this dispatcher.
* @param maxSubscribers The maximum number of subscribers allowed.
@@ -73,10 +78,17 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
* @return the result of {@link Set#add(Object)}
*/
@Override
public boolean addHandler(MessageHandler handler) {
public synchronized boolean addHandler(MessageHandler handler) {
Assert.notNull(handler, "handler must not be null");
Assert.isTrue(this.handlers.size() < this.maxSubscribers, "Maximum subscribers exceeded");
return this.handlers.add(handler);
boolean added = this.handlers.add(handler);
if (this.handlers.size() == 1) {
this.theOneHandler = handler;
}
else {
this.theOneHandler = null;
}
return added;
}
/**
@@ -85,9 +97,42 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
* @return the result of {@link Set#remove(Object)}
*/
@Override
public boolean removeHandler(MessageHandler handler) {
public synchronized boolean removeHandler(MessageHandler handler) {
Assert.notNull(handler, "handler must not be null");
return this.handlers.remove(handler);
boolean removed = this.handlers.remove(handler);
if (this.handlers.size() == 1) {
this.theOneHandler = this.handlers.iterator().next();
}
else {
this.theOneHandler = null;
}
return removed;
}
protected boolean tryOptimizedDispatch(Message<?> message) {
MessageHandler handler = this.theOneHandler;
if (handler != null) {
try {
handler.handleMessage(message);
return true;
}
catch (Exception e) {
throw wrapExceptionIfNecessary(message, e);
}
}
return false;
}
protected RuntimeException wrapExceptionIfNecessary(Message<?> message, Exception e) {
RuntimeException runtimeException = (e instanceof RuntimeException)
? (RuntimeException) e
: new MessageDeliveryException(message,
"Dispatcher failed to deliver Message.", e);
if (e instanceof MessagingException &&
((MessagingException) e).getFailedMessage() == null) {
runtimeException = new MessagingException(message, e);
}
return runtimeException;
}
@Override

37
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java
View File

@@ -19,15 +19,11 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
/**
* Implementation of {@link MessageDispatcher} that will attempt to send a
@@ -52,7 +48,7 @@ import org.springframework.messaging.MessagingException;
public class UnicastingDispatcher extends AbstractDispatcher {
private volatile boolean failover = true;
private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
private volatile LoadBalancingStrategy loadBalancingStrategy;
private final Executor executor;
@@ -84,14 +80,7 @@ public class UnicastingDispatcher extends AbstractDispatcher {
* @param loadBalancingStrategy The load balancing strategy implementation.
*/
public void setLoadBalancingStrategy(LoadBalancingStrategy loadBalancingStrategy) {
Lock lock = rwLock.writeLock();
lock.lock();
try {
this.loadBalancingStrategy = loadBalancingStrategy;
}
finally {
lock.unlock();
}
this.loadBalancingStrategy = loadBalancingStrategy;
}
@Override
@@ -109,6 +98,9 @@ public class UnicastingDispatcher extends AbstractDispatcher {
}
private boolean doDispatch(Message<?> message) {
if (this.tryOptimizedDispatch(message)) {
return true;
}
boolean success = false;
Iterator<MessageHandler> handlerIterator = this.getHandlerIterator(message);
if (!handlerIterator.hasNext()) {
@@ -122,14 +114,7 @@ public class UnicastingDispatcher extends AbstractDispatcher {
success = true; // we have a winner.
}
catch (Exception e) {
RuntimeException runtimeException = (e instanceof RuntimeException)
? (RuntimeException) e
: new MessageDeliveryException(message,
"Dispatcher failed to deliver Message.", e);
if (e instanceof MessagingException &&
((MessagingException) e).getFailedMessage() == null) {
runtimeException = new MessagingException(message, e);
}
RuntimeException runtimeException = this.wrapExceptionIfNecessary(message, e);
exceptions.add(runtimeException);
this.handleExceptions(exceptions, message, !handlerIterator.hasNext());
}
@@ -143,14 +128,8 @@ public class UnicastingDispatcher extends AbstractDispatcher {
* it simply returns the Iterator for the existing handler List.
*/
private Iterator<MessageHandler> getHandlerIterator(Message<?> message) {
Lock lock = rwLock.readLock();
lock.lock();
try {
if (this.loadBalancingStrategy != null) {
return this.loadBalancingStrategy.getHandlerIterator(message, this.getHandlers());
}
} finally {
lock.unlock();
if (this.loadBalancingStrategy != null) {
return this.loadBalancingStrategy.getHandlerIterator(message, this.getHandlers());
}
return this.getHandlers().iterator();
}

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

@@ -181,6 +181,17 @@
</xsd:annotation>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="fixed-subscriber" default="false" type="xsd:boolean">
<xsd:annotation>
<xsd:documentation><![CDATA[
When true, only one subscriber is allowed; the subscriber must be available at context initialization time,
and will be subscribed during bean initialization rather than when being started; 'auto-startup="false"' will
take no effect on a subscriber to this channel, the subscriber will always be "started".
The subscriber cannot be stopped. When true, no sub elements are allowed; 'datatype' is not allowed;
'message-converter' is not allowed. Default: false.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

109
spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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.
@@ -16,31 +16,34 @@
package org.springframework.integration.channel;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.lang.reflect.Method;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.ReflectionUtils;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
*/
public class DirectChannelTests {
@@ -59,6 +62,92 @@ public class DirectChannelTests {
assertTrue(loadBalancingStrategy instanceof RoundRobinLoadBalancingStrategy);
}
@Test
public void testSendPerfOneHandler() {
/*
* INT-3308 - used to run 12 million/sec
* 1. optimize for single handler 20 million/sec
* 2. Don't iterate over empty datatypes 23 million/sec
* 3. Don't iterate over empty interceptors 31 million/sec
* 4. Move single handler optimization to dispatcher 34 million/sec
*
* 29 million per second with increment counter in the handler
*/
DirectChannel channel = new DirectChannel();
final AtomicInteger count = new AtomicInteger();
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
count.incrementAndGet();
}
});
GenericMessage<String> message = new GenericMessage<String>("test");
assertTrue(channel.send(message));
for (int i = 0; i < 10000000; i++) {
channel.send(message);
}
}
@Test
public void testSendPerfTwoHandlers() {
/*
* INT-3308 - used to run 6.4 million/sec
* 1. Skip empty iterators as above 7.2 million/sec
* 2. optimize for single handler 6.7 million/sec (small overhead added)
* 3. remove LB rwlock from UnicastingDispatcher 7.2 million/sec
* 4. Move single handler optimization to dispatcher 7.3 million/sec
*/
DirectChannel channel = new DirectChannel();
final AtomicInteger count1 = new AtomicInteger();
final AtomicInteger count2 = new AtomicInteger();
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
count1.incrementAndGet();
}
});
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
count2.getAndIncrement();
}
});
GenericMessage<String> message = new GenericMessage<String>("test");
assertTrue(channel.send(message));
for (int i = 0; i < 10000000; i++) {
channel.send(message);
}
assertEquals(5000001, count1.get());
assertEquals(5000000, count2.get());
}
@Test
public void testSendPerfFixedSubscriberChannel() {
/*
* INT-3308 - 96 million/sec
* NOTE: in order to get a measurable time, I had to add some code to the handler -
* presumably the JIT compiler short circuited the call becaues it's a final field
* and he knows the method does nothing.
* Added the same code to the other tests for comparison.
*/
final AtomicInteger count = new AtomicInteger();
FixedSubscriberChannel channel = new FixedSubscriberChannel(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
count.incrementAndGet();
}
});
GenericMessage<String> message = new GenericMessage<String>("test");
assertTrue(channel.send(message));
for (int i = 0; i < 100000000; i++) {
channel.send(message, 0);
}
}
@Test
public void testSendInSeparateThread() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
@@ -67,6 +156,7 @@ public class DirectChannelTests {
channel.subscribe(target);
final GenericMessage<String> message = new GenericMessage<String>("test");
new Thread(new Runnable() {
@Override
public void run() {
channel.send(message);
}
@@ -140,6 +230,7 @@ public class DirectChannelTests {
this.latch = latch;
}
@Override
public void handleMessage(Message<?> message) {
this.threadName = Thread.currentThread().getName();
if (this.latch != null) {

28
spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTests-context.xml Normal file
View File

@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
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">
<int:channel id="in" fixed-subscriber="true" />
<int:transformer input-channel="in" output-channel="out" expression="payload.toUpperCase()" auto-startup="false" />
<int:channel id="out">
<int:queue/>
</int:channel>
<int:channel id="ocaFixedChannel" fixed-subscriber="true" />
<int:outbound-channel-adapter channel="ocaFixedChannel" expression="'foo'" />
<int:transformer input-channel="afterHandler" output-channel="out" expression="payload.toUpperCase()" auto-startup="false" />
<int:channel id="afterHandler" fixed-subscriber="true" />
<int:outbound-channel-adapter channel="ocaFixedChannelAfter" expression="'foo'" />
<int:channel id="ocaFixedChannelAfter" fixed-subscriber="true" />
</beans>

201
spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTests.java Normal file
View File

@@ -0,0 +1,201 @@
/*
* Copyright 2014 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.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.BeanInstantiationException;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 4.0
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class FixedSubscriberChannelTests {
@Autowired
private MessageChannel in;
@Autowired
private PollableChannel out;
@Test
public void testHappyDay() {
this.in.send(new GenericMessage<String>("foo"));
Message<?> out = this.out.receive(0);
assertEquals("FOO", out.getPayload());
assertThat(this.in, instanceOf(FixedSubscriberChannel.class));
}
@Test
public void testNoSubs() {
ConfigurableApplicationContext context = null;
try {
context = new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "NoSubs-fail-context.xml",
this.getClass());
fail("Expected exception");
}
catch (Exception e) {
assertThat(e, instanceOf(BeanCreationException.class));
assertThat(e.getCause(), instanceOf(BeanInstantiationException.class));
assertThat(e.getCause().getCause(), instanceOf(IllegalArgumentException.class));
assertThat(e.getCause().getCause().getMessage(), Matchers.containsString("Cannot instantiate a"));
}
if (context != null) {
context.close();
}
}
@Test
public void testTwoSubs() {
ConfigurableApplicationContext context = null;
try {
context = new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "TwoSubs-fail-context.xml",
this.getClass());
fail("Expected exception");
}
catch (Exception e) {
assertThat(e, instanceOf(BeanDefinitionParsingException.class));
assertThat(e.getMessage(), Matchers.containsString("Only one subscriber is allowed for a FixedSubscriberChannel."));
}
if (context != null) {
context.close();
}
}
@Test
public void testTwoSubsAfter() {
ConfigurableApplicationContext context = null;
try {
context = new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "TwoSubsAfter-fail-context.xml",
this.getClass());
fail("Expected exception");
}
catch (Exception e) {
assertThat(e, instanceOf(IllegalArgumentException.class));
assertThat(e.getMessage(), Matchers.containsString("Only one subscriber is allowed for a FixedSubscriberChannel."));
}
if (context != null) {
context.close();
}
}
@Test
public void testInterceptors() {
ConfigurableApplicationContext context = null;
try {
context = new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "Interceptors-fail-context.xml",
this.getClass());
fail("Expected exception");
}
catch (Exception e) {
assertThat(e, instanceOf(BeanDefinitionParsingException.class));
assertThat(e.getMessage(), Matchers.containsString("Cannot have interceptors when 'fixed-subscriber=\"true\"'"));
}
if (context != null) {
context.close();
}
}
@Test
public void testDatatype() {
ConfigurableApplicationContext context = null;
try {
context = new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "Datatype-fail-context.xml",
this.getClass());
fail("Expected exception");
}
catch (Exception e) {
assertThat(e, instanceOf(BeanDefinitionParsingException.class));
assertThat(e.getMessage(), Matchers.containsString("Cannot have 'datatype' when 'fixed-subscriber=\"true\"'"));
}
if (context != null) {
context.close();
}
}
@Test
public void testConverter() {
ConfigurableApplicationContext context = null;
try {
context = new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "Converter-fail-context.xml",
this.getClass());
fail("Expected exception");
}
catch (Exception e) {
assertThat(e, instanceOf(BeanDefinitionParsingException.class));
assertThat(e.getMessage(), Matchers.containsString("Cannot have 'message-converter' when 'fixed-subscriber=\"true\"'"));
}
if (context != null) {
context.close();
}
}
@Test
public void testQueue() {
ConfigurableApplicationContext context = null;
try {
context = new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "Queue-fail-context.xml",
this.getClass());
fail("Expected exception");
}
catch (Exception e) {
assertThat(e, instanceOf(BeanDefinitionParsingException.class));
assertThat(e.getMessage(), Matchers.containsString("The 'fixed-subscriber' attribute is not allowed when a <queue/> child element is present."));
}
if (context != null) {
context.close();
}
}
@Test
public void testDispatcher() {
ConfigurableApplicationContext context = null;
try {
context = new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "Dispatcher-fail-context.xml",
this.getClass());
fail("Expected exception");
}
catch (Exception e) {
assertThat(e, instanceOf(BeanDefinitionParsingException.class));
assertThat(e.getMessage(), Matchers.containsString("The 'fixed-subscriber' attribute is not allowed when a <dispatcher/> child element is present."));
}
if (context != null) {
context.close();
}
}
}

14
spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsConverter-fail-context.xml Normal file
View File

@@ -0,0 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
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">
<int:channel id="in" fixed-subscriber="true" message-converter="foo" />
<int:channel id="out">
<int:queue/>
</int:channel>
</beans>

14
spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsDatatype-fail-context.xml Normal file
View File

@@ -0,0 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
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">
<int:channel id="in" fixed-subscriber="true" datatype="java.lang.String" />
<int:channel id="out">
<int:queue/>
</int:channel>
</beans>

16
spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsDispatcher-fail-context.xml Normal file
View File

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
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">
<int:channel id="in" fixed-subscriber="true">
<int:dispatcher failover="false" />
</int:channel>
<int:channel id="out">
<int:queue/>
</int:channel>
</beans>

18
spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsInterceptors-fail-context.xml Normal file
View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
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">
<int:channel id="in" fixed-subscriber="true">
<int:interceptors>
<int:wire-tap channel="out"/>
</int:interceptors>
</int:channel>
<int:channel id="out">
<int:queue/>
</int:channel>
</beans>

14
spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsNoSubs-fail-context.xml Normal file
View File

@@ -0,0 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
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">
<int:channel id="in" fixed-subscriber="true" />
<int:channel id="out">
<int:queue/>
</int:channel>
</beans>

16
spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsQueue-fail-context.xml Normal file
View File

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
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">
<int:channel id="in" fixed-subscriber="true">
<int:queue />
</int:channel>
<int:channel id="out">
<int:queue/>
</int:channel>
</beans>

18
spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsTwoSubs-fail-context.xml Normal file
View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
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">
<int:channel id="in" fixed-subscriber="true" />
<int:transformer input-channel="in" output-channel="out" expression="payload.toUpperCase()" />
<int:transformer input-channel="in" output-channel="out" expression="payload.toUpperCase()" />
<int:channel id="out">
<int:queue/>
</int:channel>
</beans>

18
spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsTwoSubsAfter-fail-context.xml Normal file
View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
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">
<int:transformer input-channel="in" output-channel="out" expression="payload.toUpperCase()" />
<int:transformer input-channel="in" output-channel="out" expression="payload.toUpperCase()" />
<int:channel id="in" fixed-subscriber="true" />
<int:channel id="out">
<int:queue/>
</int:channel>
</beans>

11
spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java
View File

@@ -98,7 +98,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
assertEquals("TEST", reply.getPayload());
assertEquals("replyingHandlerTestInputChannel,replyingHandlerTestService", reply.getHeaders().get("history").toString());
StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); // close to the metal
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st)); // close to the metal
}
@Test
@@ -111,7 +111,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
assertEquals("optimizedRefReplyingHandlerTestInputChannel,optimizedRefReplyingHandlerTestService",
reply.getHeaders().get("history").toString());
StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); // close to the metal
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st)); // close to the metal
}
@Test
@@ -123,7 +123,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
assertEquals("TEST", reply.getPayload());
assertEquals("replyingHandlerWithStandardMethodTestInputChannel,replyingHandlerWithStandardMethodTestService", reply.getHeaders().get("history").toString());
StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); // close to the metal
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st)); // close to the metal
}
@Test
@@ -189,7 +189,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
Exception e = new RuntimeException();
StackTraceElement[] st = e.getStackTrace();
// use this to test that StackTraceUtils works as expected and returns false
assertFalse(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st));
assertFalse(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st));
return "bar";
}
@@ -202,7 +202,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
public void handleMessage(Message<?> requestMessage) {
Exception e = new RuntimeException();
StackTraceElement[] st = e.getStackTrace();
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); // close to the metal
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st)); // close to the metal
}
}
@@ -215,6 +215,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
this.prefix = prefix;
}
@Override
public String processMessage(Message<?> message) {
return prefix + ":" + message.getPayload();
}

13
spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -20,15 +20,15 @@ import static org.junit.Assert.assertEquals;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Mark Fisher
@@ -40,7 +40,7 @@ public class TransformerContextTests {
@Test
public void methodInvokingTransformer() {
ApplicationContext context = new ClassPathXmlApplicationContext(
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"transformerContextTests.xml", this.getClass());
MessageChannel input = context.getBean("input", MessageChannel.class);
PollableChannel output = context.getBean("output", PollableChannel.class);
@@ -54,14 +54,15 @@ public class TransformerContextTests {
reply = output.receive(0);
assertEquals("FOO", reply.getPayload());
StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
assertEquals("doDispatch", st[3].getMethodName()); // close to the metal
assertEquals("doSend", st[6].getMethodName()); // no MethodInvokerHelper
input = context.getBean("directRef", MessageChannel.class);
input.send(new GenericMessage<String>("foo"));
reply = output.receive(0);
assertEquals("FOO", reply.getPayload());
st = (StackTraceElement[]) reply.getHeaders().get("callStack");
assertEquals("doDispatch", st[3].getMethodName()); // SpEL
assertEquals("doSend", st[6].getMethodName()); // no MethodInvokerHelper
context.close();
}
public static class FooAdvice extends AbstractRequestHandlerAdvice {

12
spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests.java
View File

@@ -100,7 +100,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
assertEquals("TEST", reply.getPayload());
assertEquals("replyingHandlerTestInputChannel,replyingHandlerTestService", reply.getHeaders().get("history").toString());
StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st));
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st));
}
@Test
@@ -113,7 +113,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
assertEquals("optimizedRefReplyingHandlerTestInputChannel,optimizedRefReplyingHandlerTestService",
reply.getHeaders().get("history").toString());
StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st));
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st));
}
@Test
@@ -125,7 +125,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
assertEquals("TEST", reply.getPayload());
assertEquals("replyingHandlerWithStandardMethodTestInputChannel,replyingHandlerWithStandardMethodTestService", reply.getHeaders().get("history").toString());
StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack");
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st));
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st));
}
@Test
@@ -193,11 +193,12 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
.setHeader("callStack", st);
}
@Override
public String foo(String in) {
Exception e = new RuntimeException();
StackTraceElement[] st = e.getStackTrace();
// use this to test that StackTraceUtils works as expected and returns false
assertFalse(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st));
assertFalse(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st));
return "bar";
}
@@ -210,7 +211,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
public void handleMessage(Message<?> requestMessage) {
Exception e = new RuntimeException();
StackTraceElement[] st = e.getStackTrace();
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st));
assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("AbstractSubscribableChannel", "MethodInvokerHelper", st));
}
}
@@ -223,6 +224,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
this.prefix = prefix;
}
@Override
public String processMessage(Message<?> message) {
return prefix + ":" + message.getPayload();
}

34
spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractRequestResponseScenarioTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2014 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.
@@ -24,6 +24,7 @@ import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.messaging.Message;
@@ -34,7 +35,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* Convenience class for testing Spring Integration request-response message scenarios. Users
* Convenience class for testing Spring Integration request-response message scenarios. Users
* create subclasses to execute on or more {@link RequestResponseScenario} tests. each scenario defines:
* <ul>
* <li>An inputChannelName</li>
@@ -43,23 +44,24 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* <li>A handler to validate the response received on the outputChannel</li>
* </ul>
* @author David Turanski
* @author Gary Russell
*/
@RunWith(SpringJUnit4ClassRunner.class)
public abstract class AbstractRequestResponseScenarioTests {
private List<RequestResponseScenario> scenarios = null;
@Autowired
private ApplicationContext applicationContext;
@Before
public void setUp(){
scenarios = defineRequestResponseScenarios();
}
}
/**
* Execute each scenario. Instantiate the message channels, send the request message on the
* input channel and invoke the validator on the response received on the output channel.
* Execute each scenario. Instantiate the message channels, send the request message on the
* input channel and invoke the validator on the response received on the output channel.
* This can handle subscribable or pollable output channels.
*/
@Test
@@ -73,18 +75,22 @@ public abstract class AbstractRequestResponseScenarioTests {
if (outputChannel instanceof SubscribableChannel){
((SubscribableChannel) outputChannel).subscribe(scenario.getResponseValidator());
}
assertTrue(name + ": message not sent on " + scenario.getInputChannelName()
, inputChannel.send(scenario.getMessage()));
if (outputChannel instanceof PollableChannel){
Message<?> response = ((PollableChannel) outputChannel).receive(10000);
assertNotNull(name + ": receive timeout on " + scenario.getOutputChannelName(),response);
scenario.getResponseValidator().handleMessage(response);
}
assertNotNull("message was not handled on " + outputChannel + " for scenario '" + scenario.getName() + "'.",
assertNotNull("message was not handled on " + outputChannel + " for scenario '" + name + "'.",
scenario.getResponseValidator().getLastMessage());
if (outputChannel instanceof SubscribableChannel){
((SubscribableChannel) outputChannel).unsubscribe(scenario.getResponseValidator());
}
}
}
/**
@@ -92,6 +98,6 @@ public abstract class AbstractRequestResponseScenarioTests {
* @return - A List of {@link RequestResponseScenario}
*/
protected abstract List<RequestResponseScenario> defineRequestResponseScenarios();
}

2
spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTests.java
View File

@@ -24,8 +24,8 @@ import static org.springframework.integration.test.matcher.PayloadMatcher.hasPay
import java.util.ArrayList;
import java.util.List;
import org.springframework.messaging.Message;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.test.context.ContextConfiguration;
@ContextConfiguration