INT-2285 Implement max-subscribers on Channels

Add a subscriber limit on both unicast and publish-subscribe
channels. This permits detection of inadvertent channel wiring
during context initialization.

Also add a mechanism to globally set these defaults.

Two properties are added to ChannelInitializer. If the
'channelInitializer' bean is declared before a channel, and
these properties are set, it will globally override the
default (Integer.MAX_VALUE) for these properties.

For example:

    <bean id="channelInitializer" class="org.springframework.integration.config.xml.ChannelInitializer">
        <property name="autoCreate" value="true" />
        <property name="defaultMaxUnicastSubscribers" value="1" />
        <property name="defaultMaxMulticastSubscribers" value="2" />
    </bean>

will make the default max-subscribers 1 and 2 for <channel/> and
<publish-subscribe/> channels respectively.

Also applies to module channels (jms, amqp, redis).

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

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<AbstractAmqpChannel> implements SmartLifecycle, DisposableBean, BeanNameAware {
@@ -121,6 +121,8 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	private volatile Integer txSize;
 
+	private volatile int maxSubscribers = Integer.MAX_VALUE;
+
 
 	public AmqpChannelFactoryBean() {
 		this(true);
@@ -283,6 +285,10 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		this.txSize = txSize;
 	}
 
+	public void setMaxSubscribers(int maxSubscribers) {
+		this.maxSubscribers = maxSubscribers;
+	}
+
 	@Override
 	public Class<?> getObjectType() {
 		return (this.channel != null) ? this.channel.getClass() : AbstractAmqpChannel.class;
@@ -301,6 +307,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 				if (this.exchange != null) {
 					pubsub.setExchange(this.exchange);
 				}
+				pubsub.setMaxSubscribers(this.maxSubscribers);
 				this.channel = pubsub;
 			}
 			else {
@@ -309,6 +316,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 				if (StringUtils.hasText(this.queueName)) {
 					p2p.setQueueName(this.queueName);
 				}
+				p2p.setMaxSubscribers(this.maxSubscribers);
 				this.channel = p2p;
 			}
 		}
@@ -423,6 +431,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		}
 	}
 
+	@Override
 	protected void destroyInstance(AbstractAmqpChannel instance) throws Exception {
 		if (instance instanceof DisposableBean) {
 			((DisposableBean) this.channel).destroy();

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.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,19 +16,19 @@
 
 package org.springframework.integration.amqp.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.util.StringUtils;
+import org.w3c.dom.Element;
 
 /**
  * Parser for the 'channel' and 'publish-subscribe-channel' elements of the
  * Spring Integration AMQP namespace.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.1
  */
 public class AmqpChannelParser extends AbstractChannelParser {
@@ -47,10 +47,13 @@ public class AmqpChannelParser extends AbstractChannelParser {
 		builder.addPropertyReference("connectionFactory", connectionFactory);
 		if ("channel".equals(element.getLocalName())) {
 			builder.addPropertyValue("pubSub", false);
+			this.setMaxSubscribersProperty(parserContext, builder, element, IntegrationNamespaceUtils.DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME);
 		}
 		else if ("publish-subscribe-channel".equals(element.getLocalName())) {
 			builder.addPropertyValue("pubSub", true);
+			this.setMaxSubscribersProperty(parserContext, builder, element, IntegrationNamespaceUtils.DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME);
 		}
+
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge-mode");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "advice-chain");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "amqp-admin");

spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.2.xsd

@@ -336,6 +336,7 @@
 			</xsd:annotation>
 		</xsd:attribute>
 		<xsd:attributeGroup ref="containerAndTemplateAttributes"/>
+		<xsd:attributeGroup ref="integration:subscribersAttributeGroup" />
 	</xsd:complexType>
 
 	<xsd:complexType name="outboundType">

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml

@@ -17,4 +17,6 @@
 
 	<bean id="rabbitConnectionFactory" class="org.springframework.integration.amqp.StubRabbitConnectionFactory"/>
 
+	<amqp:channel id="channelWithSubscriberLimit" max-subscribers="1" />
+
 </beans>

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