diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java index 7e8eb8fc64..c04b04c8c3 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java @@ -18,7 +18,6 @@ package org.springframework.integration.amqp.channel; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.MessageListener; @@ -36,6 +35,7 @@ import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; @@ -56,6 +56,8 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple private final boolean isPubSub; + private volatile int maxSubscribers = Integer.MAX_VALUE; + public AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) { this(channelName, container, amqpTemplate, false); } @@ -71,6 +73,15 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple this.isPubSub = isPubSub; } + /** + * Specify the maximum number of subscribers supported by the + * channel's dispatcher (if it is an {@link AbstractDispatcher}). + * @param maxSubscribers + */ + public void setMaxSubscribers(int maxSubscribers) { + this.maxSubscribers = maxSubscribers; + } + public boolean subscribe(MessageHandler handler) { return this.dispatcher.addHandler(handler); } @@ -83,6 +94,9 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple public void onInit() throws Exception { super.onInit(); this.dispatcher = this.createDispatcher(); + if (this.dispatcher instanceof AbstractDispatcher) { + ((AbstractDispatcher) this.dispatcher).setMaxSubscribers(this.maxSubscribers); + } AmqpAdmin admin = new RabbitAdmin(this.container.getConnectionFactory()); Queue queue = this.initializeQueue(admin, this.channelName); this.container.setQueues(queue); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java index 3098f713d1..d09a04070b 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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,7 +20,6 @@ import java.util.List; import java.util.concurrent.Executor; import org.aopalliance.aop.Advice; - import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; @@ -56,8 +55,9 @@ import org.springframework.util.StringUtils; * a FanoutExchange named "si.fanout.[beanName]" and we send to that without any * routing key, and on the receiving side, we create an anonymous Queue that is * bound to that exchange. - * + * * @author Mark Fisher + * @author Gary Russell * @since 2.1 */ public class AmqpChannelFactoryBean extends AbstractFactoryBean implements SmartLifecycle, DisposableBean, BeanNameAware { @@ -121,6 +121,8 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean getObjectType() { return (this.channel != null) ? this.channel.getClass() : AbstractAmqpChannel.class; @@ -301,6 +307,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml index 310b6a5320..c2f654a9a4 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml @@ -17,4 +17,6 @@ + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java index 9c8c7cca2e..ee1fa56b8e 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -22,7 +22,6 @@ import java.util.List; import org.junit.Test; import org.junit.runner.RunWith; - import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.integration.MessageChannel; @@ -33,6 +32,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Mark Fisher + * @author Gary Russell * @since 2.1 */ @ContextConfiguration @@ -48,6 +48,15 @@ public class AmqpChannelParserTests { List interceptorList = TestUtils.getPropertyValue(channel, "interceptors.interceptors", List.class); assertEquals(1, interceptorList.size()); assertEquals(TestInterceptor.class, interceptorList.get(0).getClass()); + assertEquals(Integer.MAX_VALUE, TestUtils.getPropertyValue( + TestUtils.getPropertyValue(channel, "dispatcher"), "maxSubscribers", Integer.class).intValue()); + } + + @Test + public void subscriberLimit() { + MessageChannel channel = context.getBean("channelWithSubscriberLimit", MessageChannel.class); + assertEquals(1, TestUtils.getPropertyValue( + TestUtils.getPropertyValue(channel, "dispatcher"), "maxSubscribers", Integer.class).intValue()); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java index 1fb17e354c..d0ed88b842 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java @@ -32,15 +32,15 @@ import org.springframework.util.StringUtils; /** * Base implementation of {@link MessageChannel} that invokes the subscribed * {@link MessageHandler handler(s)} by delegating to a {@link MessageDispatcher}. - * + * * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell */ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel implements SubscribableChannel { - + private final AtomicInteger handlerCounter = new AtomicInteger(); - + public boolean subscribe(MessageHandler handler) { MessageDispatcher dispatcher = this.getRequiredDispatcher(); boolean added = dispatcher.addHandler(handler); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java index 51a53bd01b..52ad830685 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2012 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. @@ -23,11 +23,12 @@ import org.springframework.integration.dispatcher.UnicastingDispatcher; /** * A channel that invokes a single subscriber for each sent Message. * The invocation will occur in the sender's thread. - * + * * @author Dave Syer * @author Mark Fisher * @author Iwein Fuld * @author Oleg Zhurakousky + * @author Gary Russell */ public class DirectChannel extends AbstractSubscribableChannel { @@ -57,6 +58,15 @@ public class DirectChannel extends AbstractSubscribableChannel { this.dispatcher.setFailover(failover); } + /** + * Specify the maximum number of subscribers supported by the + * channel's dispatcher. + * @param maxSubscribers + */ + public void setMaxSubscribers(int maxSubscribers) { + this.dispatcher.setMaxSubscribers(maxSubscribers); + } + @Override protected UnicastingDispatcher getDispatcher() { return this.dispatcher; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java index abbfef8ce5..2dae5e45b9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. @@ -37,9 +37,10 @@ import org.springframework.util.ErrorHandler; * {@link Executor} typically does not block the sender's Thread since it * uses another Thread for the dispatch. (SyncTaskExecutor is an * exception but would provide no value for this channel. If synchronous - * dispatching is required, a DirectChannel should be used instead). - * + * dispatching is required, a DirectChannel should be used instead). + * * @author Mark Fisher + * @author Gary Russell * @since 1.0.3 */ public class ExecutorChannel extends AbstractSubscribableChannel { @@ -50,6 +51,8 @@ public class ExecutorChannel extends AbstractSubscribableChannel { private volatile boolean failover = true; + private volatile int maxSubscribers = Integer.MAX_VALUE; + private volatile LoadBalancingStrategy loadBalancingStrategy; @@ -89,6 +92,16 @@ public class ExecutorChannel extends AbstractSubscribableChannel { this.dispatcher.setFailover(failover); } + /** + * Specify the maximum number of subscribers supported by the + * channel's dispatcher. + * @param maxSubscribers + */ + public void setMaxSubscribers(int maxSubscribers) { + this.maxSubscribers = maxSubscribers; + this.dispatcher.setMaxSubscribers(maxSubscribers); + } + @Override protected UnicastingDispatcher getDispatcher() { return this.dispatcher; @@ -103,6 +116,7 @@ public class ExecutorChannel extends AbstractSubscribableChannel { } this.dispatcher = new UnicastingDispatcher(this.executor); this.dispatcher.setFailover(this.failover); + this.dispatcher.setMaxSubscribers(maxSubscribers); if (this.loadBalancingStrategy != null) { this.dispatcher.setLoadBalancingStrategy(this.loadBalancingStrategy); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index 4b5430f96a..992cc2637f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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,10 +24,11 @@ import org.springframework.integration.util.ErrorHandlingTaskExecutor; import org.springframework.util.ErrorHandler; /** - * A channel that sends Messages to each of its subscribers. - * + * A channel that sends Messages to each of its subscribers. + * * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ public class PublishSubscribeChannel extends AbstractSubscribableChannel { @@ -41,6 +42,9 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { private volatile boolean applySequence; + private volatile int maxSubscribers = Integer.MAX_VALUE; + + @Override public String getComponentType(){ return "publish-subscribe-channel"; } @@ -56,7 +60,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { /** * Create a PublishSubscribeChannel that will invoke the handlers in the - * message sender's thread. + * message sender's thread. */ public PublishSubscribeChannel() { this(null); @@ -103,6 +107,15 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { this.getDispatcher().setApplySequence(applySequence); } + /** + * Specify the maximum number of subscribers supported by the + * channel's dispatcher. + * @param maxSubscribers + */ + public void setMaxSubscribers(int maxSubscribers) { + this.maxSubscribers = maxSubscribers; + this.getDispatcher().setMaxSubscribers(maxSubscribers); + } /** * Callback method for initialization. */ @@ -119,6 +132,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { this.dispatcher = new BroadcastingDispatcher(this.executor); this.dispatcher.setIgnoreFailures(this.ignoreFailures); this.dispatcher.setApplySequence(this.applySequence); + this.dispatcher.setMaxSubscribers(this.maxSubscribers); } } 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 726a89749a..37703086ac 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 @@ -19,7 +19,11 @@ package org.springframework.integration.config.xml; import org.w3c.dom.Element; import org.springframework.aop.scope.ScopedProxyUtils; +import org.springframework.beans.PropertyValue; +import org.springframework.beans.PropertyValues; +import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.BeanDefinitionHolder; +import org.springframework.beans.factory.config.TypedStringValue; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionRegistry; @@ -85,4 +89,37 @@ public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser */ protected abstract BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext); + protected void setMaxSubscribersProperty(ParserContext parserContext, BeanDefinitionBuilder builder, Element element, String channelInitializerPropertyName) { + String maxSubscribers = element.getAttribute("max-subscribers"); + if (!StringUtils.hasText(maxSubscribers)) { + maxSubscribers = getDefaultMaxSubscribers(parserContext, channelInitializerPropertyName); + } + if (StringUtils.hasText(maxSubscribers)) { + builder.addPropertyValue("maxSubscribers", maxSubscribers); + } + } + + protected String getDefaultMaxSubscribers(ParserContext parserContext, String channelInitializerPropertyName) { + String maxSubscribers = null; + BeanDefinition channelInitializer = parserContext.getRegistry().getBeanDefinition( + AbstractIntegrationNamespaceHandler.CHANNEL_INITIALIZER_BEAN_NAME); + if (channelInitializer != null) { + PropertyValues propertyValues = channelInitializer.getPropertyValues(); + if (propertyValues != null) { + PropertyValue propertyValue = propertyValues + .getPropertyValue(channelInitializerPropertyName); + if (propertyValue != null) { + Object propertyValueValue = propertyValue.getValue(); + if (propertyValueValue instanceof TypedStringValue) { + maxSubscribers = ((TypedStringValue) propertyValueValue).getValue(); + } + else if (propertyValueValue instanceof String) { + maxSubscribers = (String) propertyValueValue; + } + } + } + } + return maxSubscribers; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ChannelInitializer.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ChannelInitializer.java index b9ad775235..f00baeb651 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ChannelInitializer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ChannelInitializer.java @@ -19,7 +19,6 @@ import java.util.Collection; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; @@ -41,11 +40,13 @@ import org.springframework.util.Assert; * {@link AbstractIntegrationNamespaceHandler}. * * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.1.1 */ final class ChannelInitializer implements BeanFactoryAware, InitializingBean { public static String AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME = "$autoCreateChannelCandidates"; + public static String CHANNEL_NAMES_ATTR = "channelNames"; private Log logger = LogFactory.getLog(this.getClass()); @@ -54,6 +55,10 @@ final class ChannelInitializer implements BeanFactoryAware, InitializingBean { private volatile boolean autoCreate = true; + private volatile int defaultMaxUnicastSubscribers = Integer.MAX_VALUE; + + private volatile int defaultMaxBroadcastSubscribers = Integer.MAX_VALUE; + public void setAutoCreate(boolean autoCreate) { this.autoCreate = autoCreate; } @@ -62,6 +67,32 @@ final class ChannelInitializer implements BeanFactoryAware, InitializingBean { this.beanFactory = beanFactory; } + public int getDefaultMaxUnicastSubscribers() { + return defaultMaxUnicastSubscribers; + } + + /** + * Set the default max-subscribers for all unicasting channels that don't have the + * attribute set on their dispatcher. Default {@link Integer#MAX_VALUE}. + * @param defaultMaxUnicastSubscribers + */ + public void setDefaultMaxUnicastSubscribers(int defaultMaxUnicastSubscribers) { + this.defaultMaxUnicastSubscribers = defaultMaxUnicastSubscribers; + } + + public int getDefaultMaxBroadcastSubscribers() { + return defaultMaxBroadcastSubscribers; + } + + /** + * Set the default max-subscribers for all broadcasting (pub-sub) channels that don't have the + * attribute set. Default {@link Integer#MAX_VALUE}. + * @param defaultMaxBroadcastSubscribers + */ + public void setDefaultMaxBroadcastSubscribers(int defaultMaxBroadcastSubscribers) { + this.defaultMaxBroadcastSubscribers = defaultMaxBroadcastSubscribers; + } + public void afterPropertiesSet() throws Exception { Assert.notNull(this.beanFactory, "'beanFactory' must not be null"); if (!autoCreate){ 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 d3cb1dd8c2..eb86a327ad 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 @@ -55,6 +55,20 @@ public abstract class IntegrationNamespaceUtils { static final String EXPRESSION_ATTRIBUTE = "expression"; public static final String HANDLER_ALIAS_SUFFIX = ".handler"; + /** + * Property name on ChannelInitializer used to configure the default max subscribers for + * unicast channels. + */ + public static String DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME = "defaultMaxUnicastSubscribers"; + + /** + * Property name on ChannelInitializer used to configure the default max subscribers for + * broadcast channels. + */ + public static String DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME = "defaultMaxBroadcastSubscribers"; + + + /** * Configures the provided bean definition builder with a property value corresponding to the attribute whose name 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 be8934d43f..3ff5d6fd2c 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -18,8 +18,6 @@ package org.springframework.integration.config.xml; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.w3c.dom.Element; - import org.springframework.beans.factory.config.TypedStringValue; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; @@ -31,13 +29,15 @@ import org.springframework.integration.channel.RendezvousChannel; import org.springframework.integration.store.MessageGroupQueue; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Element; /** * Parser for the <channel> element. - * + * * @author Mark Fisher * @author Iwein Fuld * @author Oleg Zhurakousky + * @author Gary Russell */ public class PointToPointChannelParser extends AbstractChannelParser { @@ -116,11 +116,16 @@ public class PointToPointChannelParser extends AbstractChannelParser { if ("failover".equals(dispatcherAttribute)) { // round-robin dispatcher is used by default, the "failover" value simply disables it builder.addConstructorArgValue(null); - } + } } else if (dispatcherElement == null) { // configure the default DirectChannel with a RoundRobinLoadBalancingStrategy builder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class); + String maxSubscribers = this.getDefaultMaxSubscribers(parserContext, + IntegrationNamespaceUtils.DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME); + if (maxSubscribers != null) { + builder.addPropertyValue("maxSubscribers", maxSubscribers); + } } else { // configure either an ExecutorChannel or DirectChannel based on existence of 'task-executor' @@ -137,8 +142,10 @@ public class PointToPointChannelParser extends AbstractChannelParser { String loadBalancer = dispatcherElement.getAttribute("load-balancer"); if ("none".equals(loadBalancer)) { builder.addConstructorArgValue(null); - } + } IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, dispatcherElement, "failover"); + this.setMaxSubscribersProperty(parserContext, builder, dispatcherElement, + IntegrationNamespaceUtils.DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME); } return builder; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java index 754e8b160a..6c4cc255ed 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2012 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,23 +16,24 @@ package org.springframework.integration.config.xml; -import org.w3c.dom.Element; - import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; /** * Parser for the <publish-subscribe-channel> element. - * + * * @author Mark Fisher + * @author Gary Russell */ public class PublishSubscribeChannelParser extends AbstractChannelParser { @Override protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".channel.PublishSubscribeChannel"); + PublishSubscribeChannel.class); String taskExecutorRef = element.getAttribute("task-executor"); if (StringUtils.hasText(taskExecutorRef)) { builder.addConstructorArgReference(taskExecutorRef); @@ -40,6 +41,8 @@ public class PublishSubscribeChannelParser extends AbstractChannelParser { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-handler"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ignore-failures"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence"); + this.setMaxSubscribersProperty(parserContext, builder, element, + IntegrationNamespaceUtils.DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME); return builder; } 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 e4cca9e14a..59c9ab5fe5 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 @@ -20,7 +20,6 @@ import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.integration.core.MessageHandler; import org.springframework.util.Assert; @@ -43,9 +42,18 @@ public abstract class AbstractDispatcher implements MessageDispatcher { protected final Log logger = LogFactory.getLog(this.getClass()); + private volatile int maxSubscribers = Integer.MAX_VALUE; + private final OrderedAwareCopyOnWriteArraySet handlers = new OrderedAwareCopyOnWriteArraySet(); + /** + * Set the maximum subscribers allowed by this dispatcher. + * @param maxSubscribers + */ + public void setMaxSubscribers(int maxSubscribers) { + this.maxSubscribers = maxSubscribers; + } /** * Returns an unmodifiable {@link Set} of this dispatcher's handlers. This @@ -62,6 +70,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher { */ public 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); } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd index a5c67678f1..8c2efb737e 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd @@ -302,6 +302,7 @@ + @@ -388,6 +389,7 @@ + @@ -3729,4 +3731,15 @@ endpoint itself is a Polling Consumer for a channel with a queue. + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests-context.xml new file mode 100644 index 0000000000..a568ad147f --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests-context.xml @@ -0,0 +1,34 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests.java new file mode 100644 index 0000000000..0fc069d8a3 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests.java @@ -0,0 +1,38 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.config.xml; + + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 2.2 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class DispatcherMaxSubscribersDefaultConfigurationTests extends DispatcherMaxSubscribersTests { + + @Test + public void test() { + doTestUnicast(Integer.MAX_VALUE, Integer.MAX_VALUE, 123, Integer.MAX_VALUE, 234); + doTestMulticast(Integer.MAX_VALUE, 2); + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDontOverrideDefaultTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDontOverrideDefaultTests-context.xml new file mode 100644 index 0000000000..7cbef70ba3 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDontOverrideDefaultTests-context.xml @@ -0,0 +1,14 @@ + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDontOverrideDefaultTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDontOverrideDefaultTests.java new file mode 100644 index 0000000000..8572565b49 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDontOverrideDefaultTests.java @@ -0,0 +1,38 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.config.xml; + + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 2.2 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class DispatcherMaxSubscribersDontOverrideDefaultTests extends DispatcherMaxSubscribersTests { + + @Test + public void test() { + doTestUnicast(Integer.MAX_VALUE, Integer.MAX_VALUE, 123, Integer.MAX_VALUE, 234); + doTestMulticast(Integer.MAX_VALUE, 2); + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests-context.xml new file mode 100644 index 0000000000..29356c9a2e --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests-context.xml @@ -0,0 +1,20 @@ + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests.java new file mode 100644 index 0000000000..dd46d510d8 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests.java @@ -0,0 +1,66 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.config.xml; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.core.SubscribableChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 2.2 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class DispatcherMaxSubscribersOverrideDefaultTests extends DispatcherMaxSubscribersTests { + + @Autowired + private SubscribableChannel oneSub; + + @Test + public void test() { + this.doTestUnicast(456, 456, 123, 456, 234); + doTestMulticast(789, 2); + } + + @Test + public void testExceed() { + oneSub.subscribe(new MessageHandler() { + public void handleMessage(Message message) throws MessagingException { + } + }); + try { + oneSub.subscribe(new MessageHandler() { + public void handleMessage(Message message) throws MessagingException { + } + }); + fail("Expected Exception"); + } + catch (IllegalArgumentException e) { + assertEquals("Maximum subscribers exceeded", e.getMessage()); + } + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersTests.java new file mode 100644 index 0000000000..2fa8b66694 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersTests.java @@ -0,0 +1,82 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.config.xml; + +import static org.junit.Assert.assertEquals; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.test.util.TestUtils; + +/** + * @author Gary Russell + * @since 2.2 + * + */ +public abstract class DispatcherMaxSubscribersTests { + + @Autowired + private MessageChannel defaultChannel; + + @Autowired + private MessageChannel defaultChannel2; + + @Autowired + private MessageChannel explicitChannel; + + @Autowired + private MessageChannel executorChannel; + + @Autowired + private MessageChannel explicitExecutorChannel; + + @Autowired + private MessageChannel pubSubDefaultChannel; + + @Autowired + private MessageChannel pubSubExplicitChannel; + + public DispatcherMaxSubscribersTests() { + super(); + } + + protected void doTestUnicast(int val1, int val2, int val3, int val4, int val5) { + Integer defaultMax = TestUtils.getPropertyValue( + TestUtils.getPropertyValue(defaultChannel, "dispatcher"), "maxSubscribers", Integer.class); + assertEquals(val1, defaultMax.intValue()); + Integer defaultMax2 = TestUtils.getPropertyValue( + TestUtils.getPropertyValue(defaultChannel2, "dispatcher"), "maxSubscribers", Integer.class); + assertEquals(val2, defaultMax2.intValue()); + Integer explicitMax = TestUtils.getPropertyValue( + TestUtils.getPropertyValue(explicitChannel, "dispatcher"), "maxSubscribers", Integer.class); + assertEquals(val3, explicitMax.intValue()); + Integer execMax = TestUtils.getPropertyValue( + TestUtils.getPropertyValue(executorChannel, "dispatcher"), "maxSubscribers", Integer.class); + assertEquals(val4, execMax.intValue()); + Integer explicitExecMax = TestUtils.getPropertyValue( + TestUtils.getPropertyValue(explicitExecutorChannel, "dispatcher"), "maxSubscribers", Integer.class); + assertEquals(val5, explicitExecMax.intValue()); + } + + protected void doTestMulticast(int val1, int val2) { + Integer defaultMax = TestUtils.getPropertyValue( + TestUtils.getPropertyValue(pubSubDefaultChannel, "dispatcher"), "maxSubscribers", Integer.class); + assertEquals(val1, defaultMax.intValue()); + Integer explicitMax = TestUtils.getPropertyValue( + TestUtils.getPropertyValue(pubSubExplicitChannel, "dispatcher"), "maxSubscribers", Integer.class); + assertEquals(val2, explicitMax.intValue()); + } +} \ No newline at end of file diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java index 6536016497..097af8e279 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java @@ -20,7 +20,6 @@ import javax.jms.MessageListener; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.beans.factory.DisposableBean; import org.springframework.context.SmartLifecycle; import org.springframework.integration.Message; @@ -29,6 +28,7 @@ import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.BroadcastingDispatcher; import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; @@ -48,16 +48,26 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr private final AbstractMessageListenerContainer container; - private volatile MessageDispatcher dispatcher; - + private volatile AbstractDispatcher dispatcher; + private volatile boolean initialized; - + + private volatile int maxSubscribers = Integer.MAX_VALUE; + public SubscribableJmsChannel(AbstractMessageListenerContainer container, JmsTemplate jmsTemplate) { super(jmsTemplate); Assert.notNull(container, "container must not be null"); this.container = container; } + /** + * Specify the maximum number of subscribers supported by the + * channel's dispatcher. + * @param maxSubscribers + */ + public void setMaxSubscribers(int maxSubscribers) { + this.maxSubscribers = maxSubscribers; + } public boolean subscribe(MessageHandler handler) { Assert.state(this.dispatcher != null, "'MessageDispatcher' must not be null. This channel might not have been initialized"); @@ -96,6 +106,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); this.dispatcher = unicastingDispatcher; } + this.dispatcher.setMaxSubscribers(this.maxSubscribers); } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java index 88f3b8bbd4..d5cd3dcd40 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java @@ -74,7 +74,7 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml index 23a9719be5..63834fd9b7 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml @@ -16,7 +16,7 @@ time-to-live="123" priority="12"/> - + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java index 577227488f..46ea321ab2 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java @@ -131,6 +131,8 @@ public class JmsChannelParserTests { assertEquals(DeliveryMode.PERSISTENT, TestUtils.getPropertyValue(jmsTemplate, "deliveryMode")); assertEquals(123L, TestUtils.getPropertyValue(jmsTemplate, "timeToLive")); assertEquals(12, TestUtils.getPropertyValue(jmsTemplate, "priority")); + assertEquals(Integer.MAX_VALUE, TestUtils.getPropertyValue( + TestUtils.getPropertyValue(channel, "dispatcher"), "maxSubscribers", Integer.class).intValue()); } @Test @@ -142,6 +144,8 @@ public class JmsChannelParserTests { AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) accessor.getPropertyValue("container"); assertEquals("test.queue", jmsTemplate.getDefaultDestinationName()); assertEquals("test.queue", container.getDestinationName()); + assertEquals(1, TestUtils.getPropertyValue( + TestUtils.getPropertyValue(channel, "dispatcher"), "maxSubscribers", Integer.class).intValue()); } @Test diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java index 13bd1b2cd2..f2f2a2d980 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java @@ -36,8 +36,8 @@ import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.BroadcastingDispatcher; -import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.integration.support.converter.MessageConverter; import org.springframework.integration.support.converter.SimpleMessageConverter; @@ -58,16 +58,16 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements private final RedisConnectionFactory connectionFactory; private final RedisTemplate redisTemplate; private final String topicName; - - private final MessageDispatcher dispatcher = new BroadcastingDispatcher(true); - + + private final AbstractDispatcher dispatcher = new BroadcastingDispatcher(true); + private volatile boolean initialized; - + // defaults private volatile Executor taskExecutor = new SimpleAsyncTaskExecutor(); private volatile RedisSerializer serializer = new StringRedisSerializer(); private volatile MessageConverter messageConverter = new SimpleMessageConverter(); - + public SubscribableRedisChannel(RedisConnectionFactory connectionFactory, String topicName) { Assert.notNull(connectionFactory, "'connectionFactory' must not be null"); Assert.hasText(topicName, "'topicName' must not be empty"); @@ -80,17 +80,26 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements Assert.notNull(taskExecutor, "'taskExecutor' must not be null"); this.taskExecutor = taskExecutor; } - + public void setMessageConverter(MessageConverter messageConverter) { Assert.notNull(messageConverter, "'messageConverter' must not be null"); this.messageConverter = messageConverter; } - + public void setSerializer(RedisSerializer serializer) { Assert.notNull(serializer, "'serializer' must not be null"); this.serializer = serializer; } + /** + * Specify the maximum number of subscribers supported by the + * channel's dispatcher. + * @param maxSubscribers + */ + public void setMaxSubscribers(int maxSubscribers) { + this.dispatcher.setMaxSubscribers(maxSubscribers); + } + public boolean subscribe(MessageHandler handler) { return this.dispatcher.addHandler(handler); } @@ -98,7 +107,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements public boolean unsubscribe(MessageHandler handler) { return this.dispatcher.removeHandler(handler); } - + @Override protected boolean doSend(Message message, long arg1) { this.redisTemplate.convertAndSend(this.topicName, this.messageConverter.fromMessage(message)); @@ -169,7 +178,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements this.container.destroy(); } } - + private class MessageListenerDelegate { @SuppressWarnings("unused") diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisChannelParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisChannelParser.java index 0d20e7010e..42432169bb 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisChannelParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisChannelParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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,21 +16,21 @@ package org.springframework.integration.redis.config; -import org.w3c.dom.Element; - import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractChannelParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.redis.channel.SubscribableRedisChannel; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; /** * Parser for the 'channel' and 'publish-subscribe-channel' elements of the * Spring Integration Redis namespace. - * + * * @author Oleg Zhurakusky * @author Artem Bilan + * @author Gary Russell * @since 2.1 */ public class RedisChannelParser extends AbstractChannelParser { @@ -45,13 +45,14 @@ public class RedisChannelParser extends AbstractChannelParser { builder.addConstructorArgReference(connectionFactory); String topicName = element.getAttribute("topic-name"); builder.addConstructorArgValue(topicName); - + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer"); // The following 2 attributes should be added once configurable on the RedisMessageListenerContainer // IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase"); // IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + this.setMaxSubscribersProperty(parserContext, builder, element, IntegrationNamespaceUtils.DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME); return builder; } diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd index 2a733d6c23..c7677020a7 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd @@ -137,6 +137,7 @@ + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests-context.xml index c83b991dac..ea85de380e 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests-context.xml @@ -18,4 +18,7 @@ + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java index 6166a955aa..46c30a1ec7 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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,9 +16,10 @@ package org.springframework.integration.redis.config; +import static junit.framework.Assert.assertEquals; + import org.junit.Test; import org.mockito.Mockito; - import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.serializer.RedisSerializer; @@ -31,24 +32,28 @@ import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; import org.springframework.integration.test.util.TestUtils; -import static junit.framework.Assert.assertEquals; - /** * @author Oleg Zhurakousky + * @author Gary Russell */ public class RedisChannelParserTests extends RedisAvailableTests{ - - @Test + + @Test @RedisAvailable public void testPubSubChannelConfig(){ ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("RedisChannelParserTests-context.xml", this.getClass()); SubscribableChannel redisChannel = context.getBean("redisChannel", SubscribableChannel.class); - JedisConnectionFactory connectionFactory = + JedisConnectionFactory connectionFactory = TestUtils.getPropertyValue(redisChannel, "connectionFactory", JedisConnectionFactory.class); RedisSerializer redisSerializer = TestUtils.getPropertyValue(redisChannel, "serializer", RedisSerializer.class); assertEquals(connectionFactory, context.getBean("redisConnectionFactory")); assertEquals(redisSerializer, context.getBean("redisSerializer")); assertEquals("si.test.topic", TestUtils.getPropertyValue(redisChannel, "topicName")); + assertEquals(Integer.MAX_VALUE, TestUtils.getPropertyValue( + TestUtils.getPropertyValue(redisChannel, "dispatcher"), "maxSubscribers", Integer.class).intValue()); + redisChannel = context.getBean("redisChannelWithSubLimit", SubscribableChannel.class); + assertEquals(1, TestUtils.getPropertyValue( + TestUtils.getPropertyValue(redisChannel, "dispatcher"), "maxSubscribers", Integer.class).intValue()); context.stop(); } @@ -58,9 +63,9 @@ public class RedisChannelParserTests extends RedisAvailableTests{ ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("RedisChannelParserTests-context.xml", this.getClass()); SubscribableChannel redisChannel = context.getBean("redisChannel", SubscribableChannel.class); final Message m = new GenericMessage("Hello Redis"); - + final Marker marker = Mockito.mock(Marker.class); - redisChannel.subscribe(new MessageHandler() { + redisChannel.subscribe(new MessageHandler() { public void handleMessage(Message message) throws MessagingException { assertEquals(m.getPayload(), message.getPayload()); marker.mark(); @@ -71,7 +76,7 @@ public class RedisChannelParserTests extends RedisAvailableTests{ Mockito.verify(marker, Mockito.times(1)).mark(); context.stop(); } - + interface Marker { void mark(); }