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

Added 'queueCapacity' and 'keepAliveSeconds' properties to ConcurrencyPolicy. Refactored DefaultMessageEndpoint and ConcurrentHandler to delegate to ConcurrencyPolicy and avoid duplication of concurrency properties. Added namespace support for 'dispatcher-policy' as a sub-element of 'channel', and also added namespace support for the new concurrency properties.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-20 20:03:54 +00:00
parent 945bc9511f
commit 35b098b164
7 changed files with 176 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java
View File

@@ -17,6 +17,8 @@
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
@@ -40,6 +42,8 @@ public class ChannelParser implements BeanDefinitionParser {
private static final String PUBLISH_SUBSCRIBE_ATTRIBUTE = "publish-subscribe";
private static final String DISPATCHER_POLICY_ELEMENT = "dispatcher-policy";
public BeanDefinition parse(Element element, ParserContext parserContext) {
RootBeanDefinition channelDef = new RootBeanDefinition(SimpleChannel.class);
@@ -48,10 +52,19 @@ public class ChannelParser implements BeanDefinitionParser {
if (StringUtils.hasText(capacity)) {
channelDef.getConstructorArgumentValues().addGenericArgumentValue(Integer.parseInt(capacity));
}
String isPublishSubscribe = element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE);
if ("true".equals(isPublishSubscribe)) {
channelDef.getConstructorArgumentValues().addGenericArgumentValue(new DispatcherPolicy(true));
boolean isPublishSubscribe = "true".equals(element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE));
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(isPublishSubscribe);
NodeList childNodes = element.getChildNodes();
for (int i = 0; i < childNodes.getLength(); i++) {
Node child = childNodes.item(i);
if (child.getNodeType() == Node.ELEMENT_NODE) {
String localName = child.getLocalName();
if (DISPATCHER_POLICY_ELEMENT.equals(localName)) {
configureDispatcherPolicy((Element) child, dispatcherPolicy);
}
}
}
channelDef.getConstructorArgumentValues().addGenericArgumentValue(dispatcherPolicy);
String beanName = element.getAttribute(ID_ATTRIBUTE);
if (!StringUtils.hasText(beanName)) {
beanName = parserContext.getReaderContext().generateBeanName(channelDef);
@@ -60,4 +73,27 @@ public class ChannelParser implements BeanDefinitionParser {
return channelDef;
}
private void configureDispatcherPolicy(Element element, DispatcherPolicy dispatcherPolicy) {
String maxMessagesPerTask = element.getAttribute("max-messages-per-task");
if (StringUtils.hasText(maxMessagesPerTask)) {
dispatcherPolicy.setMaxMessagesPerTask(Integer.parseInt(maxMessagesPerTask));
}
String receiveTimeout = element.getAttribute("receive-timeout");
if (StringUtils.hasText(receiveTimeout)) {
dispatcherPolicy.setReceiveTimeout(Long.parseLong(receiveTimeout));
}
String rejectionLimit = element.getAttribute("rejection-limit");
if (StringUtils.hasText(rejectionLimit)) {
dispatcherPolicy.setRejectionLimit(Integer.parseInt(rejectionLimit));
}
String retryInterval = element.getAttribute("retry-interval");
if (StringUtils.hasText(retryInterval)) {
dispatcherPolicy.setRetryInterval(Long.parseLong(retryInterval));
}
String shouldFailOnRejectionLimit = element.getAttribute("should-fail-on-rejection-limit");
if (StringUtils.hasText(shouldFailOnRejectionLimit)) {
dispatcherPolicy.setShouldFailOnRejectionLimit("true".equals(shouldFailOnRejectionLimit));
}
}
}

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

@@ -88,6 +88,10 @@ public class EndpointParser implements BeanDefinitionParser {
private static final String MAX_SIZE_ATTRIBUTE = "max";
private static final String QUEUE_CAPACITY_ATTRIBUTE = "queue-capacity";
private static final String KEEP_ALIVE_ATTRIBUTE = "keep-alive";
private static final String CONCURRENCY_POLICY_PROPERTY = "concurrencyPolicy";
@@ -166,15 +170,23 @@ public class EndpointParser implements BeanDefinitionParser {
}
private void parseConcurrencyPolicy(Element concurrencyElement, RootBeanDefinition endpointDef) {
ConcurrencyPolicy policy = new ConcurrencyPolicy();
String coreSize = concurrencyElement.getAttribute(CORE_SIZE_ATTRIBUTE);
String maxSize = concurrencyElement.getAttribute(MAX_SIZE_ATTRIBUTE);
String queueCapacity = concurrencyElement.getAttribute(QUEUE_CAPACITY_ATTRIBUTE);
String keepAlive = concurrencyElement.getAttribute(KEEP_ALIVE_ATTRIBUTE);
ConcurrencyPolicy policy = new ConcurrencyPolicy();
if (StringUtils.hasText(coreSize)) {
policy.setCoreSize(Integer.parseInt(coreSize));
}
if (StringUtils.hasText(maxSize)) {
policy.setMaxSize(Integer.parseInt(maxSize));
}
if (StringUtils.hasText(queueCapacity)) {
policy.setQueueCapacity(Integer.parseInt(queueCapacity));
}
if (StringUtils.hasText(keepAlive)) {
policy.setKeepAliveSeconds(Integer.parseInt(keepAlive));
}
endpointDef.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy);
}

20
spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd
View File

@@ -49,6 +49,9 @@
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:sequence>
<xsd:element ref="dispatcher-policy" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="capacity" type="xsd:integer"/>
<xsd:attribute name="publish-subscribe" type="xsd:boolean" default="false"/>
</xsd:extension>
@@ -134,6 +137,8 @@
</xsd:annotation>
<xsd:attribute name="core" type="xsd:int"/>
<xsd:attribute name="max" type="xsd:int"/>
<xsd:attribute name="queue-capacity" type="xsd:int"/>
<xsd:attribute name="keep-alive" type="xsd:int"/>
</xsd:complexType>
</xsd:element>
@@ -149,6 +154,21 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="dispatcher-policy">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a dispatcher policy.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="max-messages-per-task" type="xsd:int"/>
<xsd:attribute name="receive-timeout" type="xsd:long"/>
<xsd:attribute name="rejection-limit" type="xsd:int"/>
<xsd:attribute name="retry-interval" type="xsd:long"/>
<xsd:attribute name="should-fail-on-rejection-limit" type="xsd:boolean"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="file-source">
<xsd:complexType>
<xsd:annotation>

36
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java
View File

@@ -25,15 +25,29 @@ import org.springframework.util.Assert;
*/
public class ConcurrencyPolicy implements EndpointPolicy {
private int coreSize;
public static final int DEFAULT_CORE_SIZE = 1;
private int maxSize;
public static final int DEFAULT_MAX_SIZE = 10;
public static final int DEFAULT_QUEUE_CAPACITY = 0;
public static final int DEFAULT_KEEP_ALIVE_SECONDS = 60;
private int coreSize = DEFAULT_CORE_SIZE;
private int maxSize = DEFAULT_MAX_SIZE;
private int queueCapacity = DEFAULT_QUEUE_CAPACITY;
private int keepAliveSeconds = DEFAULT_KEEP_ALIVE_SECONDS;
public ConcurrencyPolicy() {
}
public ConcurrencyPolicy(int coreSize, int maxSize) {
Assert.isTrue(maxSize >= coreSize, "'coreSize' must not exceed 'maxSize'");
this.setCoreSize(coreSize);
this.setMaxSize(maxSize);
}
@@ -57,4 +71,22 @@ public class ConcurrencyPolicy implements EndpointPolicy {
this.maxSize = maxSize;
}
public int getQueueCapacity() {
return this.queueCapacity;
}
public void setQueueCapacity(int queueCapacity) {
Assert.isTrue(queueCapacity >= 0, "'queueCapacity' must not be negative");
this.queueCapacity = queueCapacity;
}
public int getKeepAliveSeconds() {
return this.keepAliveSeconds;
}
public void setKeepAliveSeconds(int keepAliveSeconds) {
Assert.isTrue(keepAliveSeconds >= 0, "'keepAliveSeconds' must not be negative");
this.keepAliveSeconds = keepAliveSeconds;
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java
View File

@@ -147,11 +147,9 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
public void afterPropertiesSet() {
if (this.concurrencyPolicy != null) {
if (!(this.handler instanceof ConcurrentHandler)) {
this.handler = new ConcurrentHandler(this.handler);
this.handler = new ConcurrentHandler(this.handler, this.concurrencyPolicy);
}
ConcurrentHandler concurrentHandler = (ConcurrentHandler) this.handler;
concurrentHandler.setCorePoolSize(this.concurrencyPolicy.getCoreSize());
concurrentHandler.setMaxPoolSize(this.concurrencyPolicy.getMaxSize());
if (this.errorHandler != null) {
concurrentHandler.setErrorHandler(this.errorHandler);
}

103
spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java
View File

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.dispatcher.MessageHandlerNotRunningException;
import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.message.Message;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
@@ -42,15 +43,11 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin
private MessageHandler handler;
private ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
private ThreadPoolTaskExecutor executor;
private int corePoolSize = 1;
private volatile int currentQueueCapacity;
private int maxPoolSize = 5;
private int queueCapacity = 0;
private int keepAliveSeconds = 60;
private final ConcurrencyPolicy concurrencyPolicy;
private ErrorHandler errorHandler;
@@ -60,18 +57,20 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin
public ConcurrentHandler(MessageHandler handler) {
Assert.notNull(handler, "'handler' must not be null");
this.handler = handler;
this(handler, null);
}
public ConcurrentHandler(MessageHandler handler, int corePoolSize, int maxPoolSize) {
public ConcurrentHandler(MessageHandler handler, ConcurrencyPolicy concurrencyPolicy) {
Assert.notNull(handler, "'handler' must not be null");
Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1");
Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1");
Assert.isTrue(maxPoolSize >= corePoolSize, "'corePoolSize' cannot exceed 'maxPoolSize'");
if (concurrencyPolicy != null) {
Assert.isTrue(concurrencyPolicy.getMaxSize() >= concurrencyPolicy.getCoreSize(),
"'coreSize' must not exceed 'maxSize'");
this.concurrencyPolicy = concurrencyPolicy;
}
else {
this.concurrencyPolicy = new ConcurrencyPolicy();
}
this.handler = handler;
this.corePoolSize = corePoolSize;
this.maxPoolSize = maxPoolSize;
}
@@ -80,58 +79,54 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin
this.executor = executor;
}
public void setCorePoolSize(int corePoolSize) {
Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1");
this.corePoolSize = corePoolSize;
if (this.executor != null) {
this.executor.setCorePoolSize(corePoolSize);
}
}
public void setMaxPoolSize(int maxPoolSize) {
Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1");
this.maxPoolSize = maxPoolSize;
if (this.executor != null) {
this.executor.setMaxPoolSize(maxPoolSize);
}
}
public void setQueueCapacity(int queueCapacity) {
this.queueCapacity = queueCapacity;
if (this.executor != null) {
this.executor.setQueueCapacity(queueCapacity);
}
}
public void setKeepAliveSeconds(int keepAliveSeconds) {
this.keepAliveSeconds = keepAliveSeconds;
if (this.executor != null) {
this.executor.setKeepAliveSeconds(keepAliveSeconds);
}
}
public void setErrorHandler(ErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
public void afterPropertiesSet() {
initializeExecutor();
refreshExecutor();
}
public void refreshExecutor() {
if (this.executor == null || currentQueueCapacity != this.concurrencyPolicy.getQueueCapacity()) {
this.initializeExecutor();
}
else {
this.refreshRuntimeModifiableExecutorProperties();
}
}
private void initializeExecutor() {
if (this.executor == null) {
synchronized (this.lifecycleMonitor) {
if (this.executor != null) {
this.executor.shutdown();
}
this.executor = new ThreadPoolTaskExecutor();
this.currentQueueCapacity = this.concurrencyPolicy.getQueueCapacity();
this.executor.setQueueCapacity(this.currentQueueCapacity);
CustomizableThreadFactory threadFactory = new CustomizableThreadFactory();
threadFactory.setThreadNamePrefix("handler-");
this.executor.setThreadFactory(threadFactory);
this.refreshRuntimeModifiableExecutorProperties();
}
this.executor.setCorePoolSize(this.corePoolSize);
this.executor.setMaxPoolSize(this.maxPoolSize);
this.executor.setQueueCapacity(this.queueCapacity);
this.executor.setKeepAliveSeconds(this.keepAliveSeconds);
CustomizableThreadFactory threadFactory = new CustomizableThreadFactory();
threadFactory.setThreadNamePrefix("handler-");
this.executor.setThreadFactory(threadFactory);
this.executor.afterPropertiesSet();
}
private void refreshRuntimeModifiableExecutorProperties() {
int coreSize = this.concurrencyPolicy.getCoreSize();
int maxSize = this.concurrencyPolicy.getMaxSize();
int keepAlive = this.concurrencyPolicy.getKeepAliveSeconds();
if (this.executor.getCorePoolSize() != coreSize) {
this.executor.setCorePoolSize(coreSize);
}
if (this.executor.getMaxPoolSize() != maxSize) {
this.executor.setMaxPoolSize(maxSize);
}
if (this.executor.getKeepAliveSeconds() != keepAlive) {
this.executor.setKeepAliveSeconds(keepAlive);
}
}
public boolean isRunning() {
return this.running;
}