diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index e528ae0677..0d7f25fc52 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java new file mode 100644 index 0000000000..0320f63913 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java @@ -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. + *

+ * 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"; + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 2ee7557f6c..1c65641677 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -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); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/FixedSubscriberChannelBeanFactoryPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/FixedSubscriberChannelBeanFactoryPostProcessor.java new file mode 100644 index 0000000000..1b59d72c19 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/FixedSubscriberChannelBeanFactoryPostProcessor.java @@ -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 candidateFixedChannelHandlerMap; + + private FixedSubscriberChannelBeanFactoryPostProcessor(Map 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 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); + } + } + } + } + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java index 8455297bde..b318e1865f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java @@ -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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java index 2946c30432..76bd955970 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java @@ -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 pollerElementList = DomUtils.getChildElementsByTagName(element, "poller"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractOutboundChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractOutboundChannelAdapterParser.java index 39d997187f..053b6c5f2c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractOutboundChannelAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractOutboundChannelAdapterParser.java @@ -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( diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java index 39413fdc56..2e0ef14bb5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java @@ -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 candidates; + ValueHolder argumentValue = bfppd.getConstructorArgumentValues().getArgumentValue(0, Map.class); + if (argumentValue == null) { + candidates = new ManagedMap(); + bfppd.getConstructorArgumentValues().addIndexedArgumentValue(0, candidates); + } + else { + candidates = (ManagedMap) argumentValue.getValue(); + } + candidates.put(handlerBeanName, channelName); + } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java index 6991836820..4d06581eb4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java @@ -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 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 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"); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java index b7de144c9e..81770a4eb5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java @@ -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". diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index 562ffbdd1b..d37db72fa2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -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 handlers = new OrderedAwareCopyOnWriteArraySet(); + 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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java index 96dd34bb68..ad7819d95a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java @@ -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 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 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(); } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd index 7d672a2bc9..96824e4531 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd @@ -181,6 +181,17 @@ + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java index 4d8be8dbe0..9346afc210 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java @@ -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 message = new GenericMessage("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 message = new GenericMessage("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 message = new GenericMessage("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 message = new GenericMessage("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) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTests-context.xml new file mode 100644 index 0000000000..f5f39d26da --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTests-context.xml @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTests.java new file mode 100644 index 0000000000..6d7bee6e20 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTests.java @@ -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("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 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 child element is present.")); + } + if (context != null) { + context.close(); + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsConverter-fail-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsConverter-fail-context.xml new file mode 100644 index 0000000000..3eda380334 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsConverter-fail-context.xml @@ -0,0 +1,14 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsDatatype-fail-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsDatatype-fail-context.xml new file mode 100644 index 0000000000..09f870d2b9 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsDatatype-fail-context.xml @@ -0,0 +1,14 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsDispatcher-fail-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsDispatcher-fail-context.xml new file mode 100644 index 0000000000..bd00f3216e --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsDispatcher-fail-context.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsInterceptors-fail-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsInterceptors-fail-context.xml new file mode 100644 index 0000000000..bea5313997 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsInterceptors-fail-context.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsNoSubs-fail-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsNoSubs-fail-context.xml new file mode 100644 index 0000000000..f17db81a87 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsNoSubs-fail-context.xml @@ -0,0 +1,14 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsQueue-fail-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsQueue-fail-context.xml new file mode 100644 index 0000000000..9cf1428eec --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsQueue-fail-context.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsTwoSubs-fail-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsTwoSubs-fail-context.xml new file mode 100644 index 0000000000..e6d585f0ef --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsTwoSubs-fail-context.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsTwoSubsAfter-fail-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsTwoSubsAfter-fail-context.xml new file mode 100644 index 0000000000..6c1d686ec1 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/FixedSubscriberChannelTestsTwoSubsAfter-fail-context.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java index e81fa6a847..92a610435a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java @@ -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(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java b/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java index ed66493418..b1ecd63363 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java @@ -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("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 { diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests.java index 16da50f033..24a46946f3 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests.java @@ -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(); } diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractRequestResponseScenarioTests.java b/spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractRequestResponseScenarioTests.java index a33ee7165d..0fecf9431f 100644 --- a/spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractRequestResponseScenarioTests.java +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractRequestResponseScenarioTests.java @@ -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: *

* @author David Turanski + * @author Gary Russell */ @RunWith(SpringJUnit4ClassRunner.class) public abstract class AbstractRequestResponseScenarioTests { private List 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 defineRequestResponseScenarios(); - - + + } diff --git a/spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTests.java b/spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTests.java index d06d0d2817..98cb5428d3 100644 --- a/spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTests.java +++ b/spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTests.java @@ -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