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

INT-3006: Resolve 'maxSubscribers' from INT Props

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

* Add integration properties `channels.maxUnicastSubscribers` and `channels.maxBroadcastSubscribers`
with default value to `Integer.MAX_VALUE`
* Add SpEL value resolution for those default values from parsers
* Deprecate similar properties from `ChannelInitializer`

INT-3006 `maxSubscribers` from channels' onInit()

* Move default init for `maxSubscribers` to the channels' `onInit()`
from `integrationProperties`, not from parsers
* extract `default` `integrationProperties` from static block in the `IntegrationProperties`
and use them as `default` on `IntegrationContextUtils#getIntegrationProperties`
* Add bean definition expression building to get the default values for `integration properties`
on bean building phase.

INT-3006: Add `taskScheduler.poolSize` property

INT-3006: Polishing
This commit is contained in:
Artem Bilan
2013-11-22 18:23:20 +02:00
committed by Gary Russell
parent 14c966c1fa
commit db32486d35
29 changed files with 270 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
View File

@@ -33,6 +33,7 @@ import org.springframework.integration.Message;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.dispatcher.AbstractDispatcher;
@@ -43,6 +44,7 @@ import org.springframework.util.Assert;
/**
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
* @since 2.1
*/
abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel implements SubscribableChannel, SmartLifecycle, DisposableBean {
@@ -51,11 +53,11 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
private final SimpleMessageListenerContainer container;
private volatile MessageDispatcher dispatcher;
private volatile AbstractDispatcher dispatcher;
private final boolean isPubSub;
private volatile int maxSubscribers = Integer.MAX_VALUE;
private volatile Integer maxSubscribers;
public AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) {
this(channelName, container, amqpTemplate, false);
@@ -79,6 +81,9 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
*/
public void setMaxSubscribers(int maxSubscribers) {
this.maxSubscribers = maxSubscribers;
if (this.dispatcher != null) {
this.dispatcher.setMaxSubscribers(this.maxSubscribers);
}
}
public boolean subscribe(MessageHandler handler) {
@@ -93,9 +98,13 @@ 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);
if (this.maxSubscribers == null) {
this.maxSubscribers = this.getIntegrationProperty(this.isPubSub ?
IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS :
IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS,
Integer.class);
}
this.setMaxSubscribers(this.maxSubscribers);
AmqpAdmin admin = new RabbitAdmin(this.container.getConnectionFactory());
Queue queue = this.initializeQueue(admin, this.channelName);
this.container.setQueues(queue);
@@ -110,7 +119,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
}
}
protected abstract MessageDispatcher createDispatcher();
protected abstract AbstractDispatcher createDispatcher();
protected abstract Queue initializeQueue(AmqpAdmin admin, String channelName);

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2013 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,7 @@ import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.dispatcher.AbstractDispatcher;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
@@ -57,7 +57,7 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq
}
@Override
protected MessageDispatcher createDispatcher() {
protected AbstractDispatcher createDispatcher() {
UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher();
unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy());
return unicastingDispatcher;

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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,8 +23,8 @@ import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.integration.dispatcher.AbstractDispatcher;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.dispatcher.MessageDispatcher;
/**
* @author Mark Fisher
@@ -65,7 +65,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
}
@Override
protected MessageDispatcher createDispatcher() {
protected AbstractDispatcher createDispatcher() {
return new BroadcastingDispatcher(true);
}

10
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
View File

@@ -121,7 +121,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
private volatile Integer txSize;
private volatile int maxSubscribers = Integer.MAX_VALUE;
private volatile Integer maxSubscribers;
public AmqpChannelFactoryBean() {
@@ -307,7 +307,9 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
if (this.exchange != null) {
pubsub.setExchange(this.exchange);
}
pubsub.setMaxSubscribers(this.maxSubscribers);
if (this.maxSubscribers != null) {
pubsub.setMaxSubscribers(this.maxSubscribers);
}
this.channel = pubsub;
}
else {
@@ -316,7 +318,9 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
if (StringUtils.hasText(this.queueName)) {
p2p.setQueueName(this.queueName);
}
p2p.setMaxSubscribers(this.maxSubscribers);
if (this.maxSubscribers != null) {
p2p.setMaxSubscribers(this.maxSubscribers);
}
this.channel = p2p;
}
}

17
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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,12 +16,13 @@
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
@@ -29,6 +30,7 @@ import org.w3c.dom.Element;
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
* @since 2.1
*/
public class AmqpChannelParser extends AbstractChannelParser {
@@ -45,15 +47,10 @@ public class AmqpChannelParser extends AbstractChannelParser {
connectionFactory = "rabbitConnectionFactory";
}
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);
}
builder.addPropertyValue("pubSub", "publish-subscribe-channel".equals(element.getLocalName()));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-subscribers");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge-mode");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "advice-chain");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "amqp-admin");

13
spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.channel;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
@@ -34,6 +35,8 @@ public class DirectChannel extends AbstractSubscribableChannel {
private final UnicastingDispatcher dispatcher = new UnicastingDispatcher();
private volatile Integer maxSubscribers;
/**
* Create a channel with default {@link RoundRobinLoadBalancingStrategy}
*/
@@ -64,6 +67,7 @@ public class DirectChannel extends AbstractSubscribableChannel {
* @param maxSubscribers
*/
public void setMaxSubscribers(int maxSubscribers) {
this.maxSubscribers = maxSubscribers;
this.dispatcher.setMaxSubscribers(maxSubscribers);
}
@@ -72,4 +76,13 @@ public class DirectChannel extends AbstractSubscribableChannel {
return this.dispatcher;
}
@Override
protected void onInit() throws Exception {
super.onInit();
if (this.maxSubscribers == null) {
Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class);
this.setMaxSubscribers(maxSubscribers);
}
}
}

9
spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.channel;
import java.util.concurrent.Executor;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
@@ -41,6 +42,7 @@ import org.springframework.util.ErrorHandler;
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
* @since 1.0.3
*/
public class ExecutorChannel extends AbstractSubscribableChannel {
@@ -51,7 +53,7 @@ public class ExecutorChannel extends AbstractSubscribableChannel {
private volatile boolean failover = true;
private volatile int maxSubscribers = Integer.MAX_VALUE;
private volatile Integer maxSubscribers;
private volatile LoadBalancingStrategy loadBalancingStrategy;
@@ -116,7 +118,10 @@ public class ExecutorChannel extends AbstractSubscribableChannel {
}
this.dispatcher = new UnicastingDispatcher(this.executor);
this.dispatcher.setFailover(this.failover);
this.dispatcher.setMaxSubscribers(maxSubscribers);
if (this.maxSubscribers == null) {
this.maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class);
}
this.dispatcher.setMaxSubscribers(this.maxSubscribers);
if (this.loadBalancingStrategy != null) {
this.dispatcher.setLoadBalancingStrategy(this.loadBalancingStrategy);
}

8
spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.channel;
import java.util.concurrent.Executor;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
@@ -44,7 +45,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
private volatile int minSubscribers;
private volatile int maxSubscribers = Integer.MAX_VALUE;
private volatile Integer maxSubscribers;
@Override
public String getComponentType(){
@@ -147,7 +148,10 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
this.dispatcher.setIgnoreFailures(this.ignoreFailures);
this.dispatcher.setApplySequence(this.applySequence);
this.dispatcher.setMinSubscribers(this.minSubscribers);
this.dispatcher.setMaxSubscribers(this.maxSubscribers);
}
if (this.maxSubscribers == null) {
Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class);
this.setMaxSubscribers(maxSubscribers);
}
}

13
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelAdapterParser.java
View File

@@ -20,10 +20,8 @@ import org.w3c.dom.Element;
import org.springframework.beans.MutablePropertyValues;
import org.springframework.beans.factory.BeanDefinitionStoreException;
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.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
@@ -82,15 +80,8 @@ public abstract class AbstractChannelAdapterParser extends AbstractBeanDefinitio
if (parserContext.isNested()) {
return null;
}
String channelId = element.getAttribute(ID_ATTRIBUTE);
if (!StringUtils.hasText(channelId)) {
parserContext.getReaderContext().error("The channel-adapter's 'id' attribute is required when no 'channel' "
+ "reference has been provided, because that 'id' would be used for the created channel.", element);
}
BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
BeanDefinitionHolder holder = new BeanDefinitionHolder(channelBuilder.getBeanDefinition(), channelId);
BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry());
return channelId;
return IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
}
/**

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2013 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.
@@ -19,11 +19,7 @@ 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;
@@ -35,9 +31,10 @@ import org.springframework.util.xml.DomUtils;
/**
* Base class for channel parsers.
*
*
* @author Mark Fisher
* @author Dave Syer
* @author Artem Bilan
*/
public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser {
@@ -67,7 +64,7 @@ public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser
beanDefinition.setSource(parserContext.extractSource(element));
return beanDefinition;
}
/* (non-Javadoc)
* @see org.springframework.beans.factory.xml.AbstractBeanDefinitionParser#registerBeanDefinition(org.springframework.beans.factory.config.BeanDefinitionHolder, org.springframework.beans.factory.support.BeanDefinitionRegistry)
*/
@@ -89,37 +86,4 @@ 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;
}
}

9
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractIntegrationNamespaceHandler.java
View File

@@ -96,10 +96,10 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa
BeanDefinitionRegistry registry = parserContext.getRegistry();
if (registry instanceof ListableBeanFactory) {
alreadyRegistered = ((ListableBeanFactory) registry)
.containsBean(IntegrationContextUtils.INTEGRATION_PROPERTIES_BEAN_NAME);
.containsBean(IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME);
}
else {
alreadyRegistered = registry.isBeanNameInUse(IntegrationContextUtils.INTEGRATION_PROPERTIES_BEAN_NAME);
alreadyRegistered = registry.isBeanNameInUse(IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME);
}
if (!alreadyRegistered) {
ResourcePatternResolver resourceResolver =
@@ -115,7 +115,7 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa
.genericBeanDefinition(PropertiesFactoryBean.class)
.addPropertyValue("locations", resources);
registry.registerBeanDefinition(IntegrationContextUtils.INTEGRATION_PROPERTIES_BEAN_NAME,
registry.registerBeanDefinition(IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME,
integrationPropertiesBuilder.getBeanDefinition());
}
catch (IOException e) {
@@ -146,8 +146,7 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa
alreadyRegistered = parserContext.getRegistry().isBeanNameInUse(CHANNEL_INITIALIZER_BEAN_NAME);
}
if (!alreadyRegistered) {
String channelsAutoCreateExpression = "#{@" +IntegrationContextUtils.INTEGRATION_PROPERTIES_BEAN_NAME +
"['" + IntegrationProperties.CHANNELS_AUTOCREATE + "']}";
String channelsAutoCreateExpression = IntegrationProperties.getExpressionFor(IntegrationProperties.CHANNELS_AUTOCREATE);
BeanDefinitionBuilder channelDef = BeanDefinitionBuilder.genericBeanDefinition(ChannelInitializer.class)
.addPropertyValue("autoCreate", channelsAutoCreateExpression);
BeanDefinitionHolder channelCreatorHolder = new BeanDefinitionHolder(channelDef.getBeanDefinition(), CHANNEL_INITIALIZER_BEAN_NAME);

45
spring-integration-core/src/main/java/org/springframework/integration/config/xml/ChannelInitializer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -19,15 +19,12 @@ 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;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.util.Assert;
/**
@@ -47,17 +44,12 @@ 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());
private volatile BeanFactory beanFactory;
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;
@@ -67,32 +59,6 @@ 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){
@@ -111,10 +77,7 @@ final class ChannelInitializer implements BeanFactoryAware, InitializingBean {
if (this.logger.isDebugEnabled()){
this.logger.debug("Auto-creating channel '" + channelName + "' as DirectChannel");
}
RootBeanDefinition messageChannel = new RootBeanDefinition();
messageChannel.setBeanClass(DirectChannel.class);
BeanDefinitionHolder messageChannelHolder = new BeanDefinitionHolder(messageChannel, channelName);
BeanDefinitionReaderUtils.registerBeanDefinition(messageChannelHolder, (BeanDefinitionRegistry) this.beanFactory);
IntegrationNamespaceUtils.autoCreateDirectChannel(channelName, (BeanDefinitionRegistry) this.beanFactory);
}
}
}
@@ -136,4 +99,4 @@ final class ChannelInitializer implements BeanFactoryAware, InitializingBean {
return channelNames;
}
}
}
}

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

@@ -33,6 +33,7 @@ import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationProperties;
/**
* A {@link BeanFactoryPostProcessor} implementation that provides default beans for the error handling and task
@@ -150,7 +151,8 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce
}
BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler");
schedulerBuilder.addPropertyValue("poolSize", 10);
String taskSchedulerPoolSizeExpression = IntegrationProperties.getExpressionFor(IntegrationProperties.TASKSCHEDULER_POOLSIZE);
schedulerBuilder.addPropertyValue("poolSize", taskSchedulerPoolSizeExpression);
schedulerBuilder.addPropertyValue("threadNamePrefix", "task-scheduler-");
schedulerBuilder.addPropertyValue("rejectedExecutionHandler", new CallerRunsPolicy());
BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(

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

@@ -70,21 +70,6 @@ public abstract class IntegrationNamespaceUtils {
public static final String AUTO_STARTUP = "auto-startup";
public static final String PHASE = "phase";
/**
* 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
* is provided if that attribute is defined in the given element.
@@ -507,10 +492,16 @@ public abstract class IntegrationNamespaceUtils {
parserContext.getReaderContext().error("The channel-adapter's 'id' attribute is required when no 'channel' "
+ "reference has been provided, because that 'id' would be used for the created channel.", element);
}
BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
BeanDefinitionHolder holder = new BeanDefinitionHolder(channelBuilder.getBeanDefinition(), channelId);
BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry());
autoCreateDirectChannel(channelId, parserContext.getRegistry());
return channelId;
}
public static void autoCreateDirectChannel(String channelName, BeanDefinitionRegistry registry) {
BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
BeanDefinitionHolder holder = new BeanDefinitionHolder(channelBuilder.getBeanDefinition(), channelName);
BeanDefinitionReaderUtils.registerBeanDefinition(holder, registry);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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,6 +18,8 @@ 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;
@@ -29,7 +31,6 @@ 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 &lt;channel&gt; element.
@@ -38,6 +39,7 @@ import org.w3c.dom.Element;
* @author Iwein Fuld
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*/
public class PointToPointChannelParser extends AbstractChannelParser {
@@ -121,11 +123,6 @@ public class PointToPointChannelParser extends AbstractChannelParser {
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'
@@ -144,8 +141,7 @@ public class PointToPointChannelParser extends AbstractChannelParser {
builder.addConstructorArgValue(null);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, dispatcherElement, "failover");
this.setMaxSubscribersProperty(parserContext, builder, dispatcherElement,
IntegrationNamespaceUtils.DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, dispatcherElement, "max-subscribers");
}
return builder;
}

4
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java
View File

@@ -28,6 +28,7 @@ import org.springframework.util.StringUtils;
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*/
public class PublishSubscribeChannelParser extends AbstractChannelParser {
@@ -42,8 +43,7 @@ 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);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-subscribers");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "min-subscribers");
return builder;
}

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

@@ -49,9 +49,7 @@ public abstract class IntegrationContextUtils {
public static final String INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME = "integrationHeaderChannelRegistry";
public static final String INTEGRATION_PROPERTIES_BEAN_NAME = "integrationProperties";
private static final Properties EMPTY_PROPERTIES = new Properties();
public static final String INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME = "integrationGlobalProperties";
/**
* Return the {@link MetadataStore} bean whose name is "metadataStore".
@@ -113,21 +111,23 @@ public abstract class IntegrationContextUtils {
}
/**
* @return the global {@link IntegrationContextUtils#INTEGRATION_PROPERTIES_BEAN_NAME}
* @return the global {@link IntegrationContextUtils#INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME}
* bean from provided {@code #beanFactory}, which represents the merged
* properties values from all 'META-INF/spring.integration.default.properties'
* and 'META-INF/spring.integration.properties'.
* May return {@link IntegrationContextUtils#EMPTY_PROPERTIES} if there is no
* {@link IntegrationContextUtils#INTEGRATION_PROPERTIES_BEAN_NAME} bean within
* Or user-defined {@link Properties} bean.
* May return only {@link IntegrationProperties#defaults()} if there is no
* {@link IntegrationContextUtils#INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME} bean within
* provided {@code #beanFactory} or provided {@code #beanFactory} is null.
*/
public static Properties getIntegrationProperties(BeanFactory beanFactory) {
Properties properties = null;
Properties properties = new Properties();
properties.putAll(IntegrationProperties.defaults());
if (beanFactory != null) {
properties = getBeanOfType(beanFactory, INTEGRATION_PROPERTIES_BEAN_NAME, Properties.class);
}
if (properties == null) {
properties = EMPTY_PROPERTIES;
Properties userProperties = getBeanOfType(beanFactory, INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME, Properties.class);
if (userProperties != null) {
properties.putAll(userProperties);
}
}
return properties;
}

17
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -30,6 +30,7 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
@@ -59,6 +60,8 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
*/
protected final Log logger = LogFactory.getLog(getClass());
private final ConversionService defaultConversionService = new DefaultConversionService();
private volatile String beanName;
private volatile String componentName;
@@ -67,6 +70,8 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
private volatile TaskScheduler taskScheduler;
private volatile Properties integrationProperties = IntegrationProperties.defaults();
private volatile ConversionService conversionService;
private volatile ApplicationContext applicationContext;
@@ -102,6 +107,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
public final void setBeanFactory(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "'beanFactory' must not be null");
this.beanFactory = beanFactory;
this.integrationProperties = IntegrationContextUtils.getIntegrationProperties(this.beanFactory);
}
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
@@ -176,7 +182,16 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
* @see IntegrationContextUtils#getIntegrationProperties
*/
protected Properties getIntegrationProperties() {
return IntegrationContextUtils.getIntegrationProperties(this.beanFactory);
return this.integrationProperties;
}
/**
* @param key Integration property.
* @param tClass the class to convert a value of Integration property.
* @return the value of the Integration property converted to the provide type.
*/
protected <T> T getIntegrationProperty(String key, Class<T> tClass) {
return this.defaultConversionService.convert(this.integrationProperties.getProperty(key), tClass);
}
@Override

88
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java
View File

@@ -16,16 +16,96 @@
package org.springframework.integration.context;
import java.io.IOException;
import java.util.Properties;
import org.springframework.beans.factory.config.PropertiesFactoryBean;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
/**
* Convention Enumeration to represent keys from 'META-INF/spring.integration.properties'.
* Utility class to encapsulate infrastructure Integration properties constants and
* their default values from resources 'META-INF/spring.integration.default.properties'.
*
* @author Artem Bilan
* @since 3.0
*/
public interface IntegrationProperties {
public final class IntegrationProperties {
String LATE_REPLY_LOGGING_LEVEL = "messagingTemplate.lateReply.logging.level";
/**
* Specifies whether to allow create automatically {@link org.springframework.integration.channel.DirectChannel}
* beans for non-declared channels or not.
*/
public static final String CHANNELS_AUTOCREATE = "channels.autoCreate";
String CHANNELS_AUTOCREATE = "channels.autoCreate";
/**
* Specifies the value for {@link org.springframework.integration.dispatcher.UnicastingDispatcher#maxSubscribers}
* in case of point-to-point channels (e.g. {@link org.springframework.integration.channel.ExecutorChannel}),
* if the attribute {@code max-subscribers} isn't configured on the channel component.
*/
public static final String CHANNELS_MAX_UNICAST_SUBSCRIBERS = "channels.maxUnicastSubscribers";
/**
* Specifies the value for {@link org.springframework.integration.dispatcher.BroadcastingDispatcher#maxSubscribers}
* in case of point-to-point channels (e.g. {@link org.springframework.integration.channel.PublishSubscribeChannel}),
* if the attribute {@code max-subscribers} isn't configured on the channel component.
*/
public static final String CHANNELS_MAX_BROADCAST_SUBSCRIBERS = "channels.maxBroadcastSubscribers";
/**
* Specifies the value of {@link org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler#poolSize}
* for {@code taskScheduler} bean initialized but Integration infrastructure.
* @see org.springframework.integration.config.xml.DefaultConfiguringBeanFactoryPostProcessor#registerTaskScheduler
*/
public static final String TASKSCHEDULER_POOLSIZE = "taskScheduler.poolSize";
// TODO public static final String LATE_REPLY_LOGGING_LEVEL = "messagingTemplate.lateReply.logging.level";
private static Properties defaults;
static {
String resourcePattern = "classpath*:META-INF/spring.integration.default.properties";
try {
ResourcePatternResolver resourceResolver = new PathMatchingResourcePatternResolver(IntegrationProperties.class.getClassLoader());
Resource[] defaultResources = resourceResolver.getResources(resourcePattern);
PropertiesFactoryBean propertiesFactoryBean = new PropertiesFactoryBean();
propertiesFactoryBean.setLocations(defaultResources);
propertiesFactoryBean.afterPropertiesSet();
defaults = propertiesFactoryBean.getObject();
}
catch (IOException e) {
throw new IllegalStateException("Can't load '" + resourcePattern + "' resources.", e);
}
}
/**
* @return {@link Properties} with default values for Integration properties
* from resources 'META-INF/spring.integration.default.properties'.
*/
public static Properties defaults() {
return defaults;
}
/**
* Build the bean property definition expression to resolve the value
* from Integration properties within the bean building phase.
*
* @param key the Integration property key.
* @return the bean property definition expression.
* @throws IllegalArgumentException if provided {@code key} isn't an Integration property.
*/
public static String getExpressionFor(String key) {
if (defaults.containsKey(key)) {
return "#{T(" + IntegrationContextUtils.class.getName() + ").getIntegrationProperties(beanFactory).getProperty('" + key + "')}";
}
else {
throw new IllegalArgumentException("The provided key [" + key + "] isn't the one of Integration properties: " + defaults.keySet());
}
}
private IntegrationProperties() {
}
}

4
spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties
View File

@@ -1,2 +1,4 @@
channels.autoCreate=true
messagingTemplate.lateReply.logging.level=warn
channels.maxUnicastSubscribers=0x7fffffff
channels.maxBroadcastSubscribers=0x7fffffff
taskScheduler.poolSize=10

2
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests-context.xml
View File

@@ -7,6 +7,8 @@
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:service-activator input-channel="autoCreateChannel" expression="'foo'"/>
<int:channel id="defaultChannel" />
<int:channel id="defaultChannel2">

19
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests-context.xml
View File

@@ -1,15 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
<bean id="channelInitializer" class="org.springframework.integration.config.xml.ChannelInitializer">
<property name="autoCreate" value="true" />
<property name="defaultMaxUnicastSubscribers" value="456" />
<property name="defaultMaxBroadcastSubscribers" value="789" />
</bean>
<util:properties id="integrationGlobalProperties">
<prop key="channels.maxUnicastSubscribers">456</prop>
<prop key="channels.maxBroadcastSubscribers">789</prop>
</util:properties>
<import resource="DispatcherMaxSubscribersDefaultConfigurationTests-context.xml" />

23
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersTests.java
View File

@@ -23,11 +23,15 @@ import org.springframework.integration.test.util.TestUtils;
/**
* @author Gary Russell
* @author Artem Bilan
* @since 2.2
*
*/
public abstract class DispatcherMaxSubscribersTests {
@Autowired
private MessageChannel autoCreateChannel;
@Autowired
private MessageChannel defaultChannel;
@@ -54,20 +58,17 @@ public abstract class DispatcherMaxSubscribersTests {
}
protected void doTestUnicast(int val1, int val2, int val3, int val4, int val5) {
Integer defaultMax = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(defaultChannel, "dispatcher"), "maxSubscribers", Integer.class);
Integer autoCreateMax = TestUtils.getPropertyValue(autoCreateChannel, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val1, autoCreateMax.intValue());
Integer defaultMax = TestUtils.getPropertyValue(defaultChannel, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val1, defaultMax.intValue());
Integer defaultMax2 = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(defaultChannel2, "dispatcher"), "maxSubscribers", Integer.class);
Integer defaultMax2 = TestUtils.getPropertyValue(defaultChannel2, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val2, defaultMax2.intValue());
Integer explicitMax = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(explicitChannel, "dispatcher"), "maxSubscribers", Integer.class);
Integer explicitMax = TestUtils.getPropertyValue(explicitChannel, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val3, explicitMax.intValue());
Integer execMax = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(executorChannel, "dispatcher"), "maxSubscribers", Integer.class);
Integer execMax = TestUtils.getPropertyValue(executorChannel, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val4, execMax.intValue());
Integer explicitExecMax = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(explicitExecutorChannel, "dispatcher"), "maxSubscribers", Integer.class);
Integer explicitExecMax = TestUtils.getPropertyValue(explicitExecutorChannel, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val5, explicitExecMax.intValue());
}
@@ -81,4 +82,4 @@ public abstract class DispatcherMaxSubscribersTests {
Integer explicitMin = TestUtils.getPropertyValue(pubSubExplicitChannel, "dispatcher.minSubscribers", Integer.class);
assertEquals(1, explicitMin.intValue());
}
}
}

14
spring-integration-core/src/test/java/org/springframework/integration/context/IntegrationContextTests.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.context;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import java.util.Properties;
@@ -26,6 +25,8 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -38,17 +39,22 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
public class IntegrationContextTests {
@Autowired
@Qualifier(IntegrationContextUtils.INTEGRATION_PROPERTIES_BEAN_NAME)
@Qualifier(IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME)
private Properties integrationProperties;
@Autowired
@Qualifier("fooService")
private IntegrationObjectSupport serviceActivator;
@Autowired
private ThreadPoolTaskScheduler taskScheduler;
@Test
public void testIntegrationContextComponents() {
assertEquals("error", this.integrationProperties.get(IntegrationProperties.LATE_REPLY_LOGGING_LEVEL));
assertSame(this.integrationProperties, this.serviceActivator.getIntegrationProperties());
//TODO INT-3005 assertEquals("error", this.integrationProperties.get(IntegrationProperties.LATE_REPLY_LOGGING_LEVEL));
assertEquals("20", this.integrationProperties.get(IntegrationProperties.TASKSCHEDULER_POOLSIZE));
assertEquals(this.integrationProperties, this.serviceActivator.getIntegrationProperties());
assertEquals(20, TestUtils.getPropertyValue(this.taskScheduler, "poolSize"));
}
}

3
spring-integration-core/src/test/resources/META-INF/spring.integration.properties
View File

@@ -1,2 +1,5 @@
#channels.autoCreate=false
#channels.maxUnicastSubscribers=1
#channels.maxBroadcastSubscribers=1
messagingTemplate.lateReply.logging.level=error
taskScheduler.poolSize=20

9
spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java
View File

@@ -26,6 +26,7 @@ import org.springframework.integration.Message;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.dispatcher.AbstractDispatcher;
@@ -51,7 +52,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
private volatile boolean initialized;
private volatile int maxSubscribers = Integer.MAX_VALUE;
private volatile Integer maxSubscribers;
public SubscribableJmsChannel(AbstractMessageListenerContainer container, JmsTemplate jmsTemplate) {
super(jmsTemplate);
@@ -105,6 +106,12 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy());
this.dispatcher = unicastingDispatcher;
}
if (this.maxSubscribers == null) {
this.maxSubscribers = this.getIntegrationProperty(isPubSub ?
IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS :
IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS,
Integer.class);
}
this.dispatcher.setMaxSubscribers(this.maxSubscribers);
}

8
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java
View File

@@ -18,12 +18,13 @@ package org.springframework.integration.jms.config;
import javax.jms.Session;
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
@@ -58,12 +59,13 @@ public class JmsChannelParser extends AbstractChannelParser {
builder.addPropertyReference("connectionFactory", connectionFactory);
if ("channel".equals(element.getLocalName())) {
this.parseDestination(element, parserContext, builder, "queue");
this.setMaxSubscribersProperty(parserContext, builder, element, IntegrationNamespaceUtils.DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME);
}
else if ("publish-subscribe-channel".equals(element.getLocalName())) {
this.parseDestination(element, parserContext, builder, "topic");
this.setMaxSubscribersProperty(parserContext, builder, element, IntegrationNamespaceUtils.DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-subscribers");
String containerType = element.getAttribute(CONTAINER_TYPE_ATTRIBUTE);
String containerClass = element.getAttribute(CONTAINER_CLASS_ATTRIBUTE);
if (!StringUtils.hasText(containerClass)) {

8
spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java
View File

@@ -34,6 +34,7 @@ import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.dispatcher.AbstractDispatcher;
@@ -61,6 +62,8 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
private final AbstractDispatcher dispatcher = new BroadcastingDispatcher(true);
private volatile Integer maxSubscribers;
private volatile boolean initialized;
// defaults
@@ -97,6 +100,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
* @param maxSubscribers
*/
public void setMaxSubscribers(int maxSubscribers) {
this.maxSubscribers = maxSubscribers;
this.dispatcher.setMaxSubscribers(maxSubscribers);
}
@@ -120,6 +124,10 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
return;
}
super.onInit();
if (this.maxSubscribers == null) {
Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class);
this.setMaxSubscribers(maxSubscribers);
}
if (this.messageConverter == null){
this.messageConverter = new SimpleMessageConverter();
}

8
spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisChannelParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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,13 +16,14 @@
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
@@ -52,7 +53,8 @@ public class RedisChannelParser extends AbstractChannelParser {
// 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);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-subscribers");
return builder;
}