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

Added CallerRunsPolicy and threadNamePrefix for default MessageBus taskScheduler and ConcurrencyInterceptor.

Browse Source
This commit is contained in:
Mark Fisher
2008-07-04 23:32:27 +00:00
parent 71da892bcb
commit bd5d71120f
7 changed files with 39 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -22,6 +22,7 @@ import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -48,19 +49,20 @@ import org.springframework.integration.channel.factory.QueueChannelFactory;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.DefaultEndpointRegistry;
import org.springframework.integration.endpoint.EndpointRegistry;
import org.springframework.integration.endpoint.EndpointTrigger;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.MessagingGateway;
import org.springframework.integration.endpoint.TargetEndpoint;
import org.springframework.integration.endpoint.EndpointTrigger;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.SimpleTaskScheduler;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.util.Assert;
/**
@@ -200,8 +202,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
}
this.initializing = true;
if (this.taskScheduler == null) {
this.taskScheduler = new SimpleTaskScheduler(
new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE));
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE);
executor.setThreadFactory(new CustomizableThreadFactory("message-bus-"));
executor.setRejectedExecutionHandler(new CallerRunsPolicy());
this.taskScheduler = new SimpleTaskScheduler(executor);
}
if (this.getErrorChannel() == null) {
this.setErrorChannel(new DefaultErrorChannel());

4
org.springframework.integration/src/main/java/org/springframework/integration/config/ConcurrencyInterceptorParser.java
View File

@@ -46,8 +46,10 @@ public class ConcurrencyInterceptorParser implements BeanDefinitionRegisteringPa
else {
ConcurrencyPolicy policy = this.parseConcurrencyPolicy(element);
builder.addConstructorArgValue(policy);
builder.addConstructorArgValue("concurrency-interceptor-");
}
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
return BeanDefinitionReaderUtils.registerWithGeneratedName(
builder.getBeanDefinition(), parserContext.getRegistry());
}
private ConcurrencyPolicy parseConcurrencyPolicy(Element element) {

25
org.springframework.integration/src/main/java/org/springframework/integration/config/EndpointInterceptorParser.java
View File

@@ -19,6 +19,10 @@ package org.springframework.integration.config;
import java.util.HashMap;
import java.util.Map;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
@@ -28,9 +32,6 @@ import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
import org.springframework.beans.factory.xml.NamespaceHandler;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.Assert;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
/**
* A helper class for parsing the sub-elements of an endpoint's
@@ -57,8 +58,8 @@ public class EndpointInterceptorParser {
Element childElement = (Element) child;
String localName = child.getLocalName();
if ("bean".equals(localName)) {
BeanDefinitionParserDelegate beanParser = new BeanDefinitionParserDelegate(parserContext
.getReaderContext());
BeanDefinitionParserDelegate beanParser =
new BeanDefinitionParserDelegate(parserContext.getReaderContext());
beanParser.initDefaults(childElement.getOwnerDocument().getDocumentElement());
BeanDefinitionHolder beanDefinitionHolder = beanParser.parseBeanDefinitionElement(childElement);
parserContext.registerBeanComponent(new BeanComponentDefinition(beanDefinitionHolder));
@@ -70,7 +71,7 @@ public class EndpointInterceptorParser {
}
else {
BeanDefinitionRegisteringParser parser = this.parsers.get(localName);
String interceptorBeanName;
String interceptorBeanName = null;
if (parser == null) {
interceptorBeanName = handleNonstandardInterceptor(childElement, parserContext);
}
@@ -85,13 +86,13 @@ public class EndpointInterceptorParser {
}
protected String handleNonstandardInterceptor(Element childElement, ParserContext parserContext) {
NamespaceHandler handlerFromOtherNamespace = parserContext.getReaderContext().getNamespaceHandlerResolver()
NamespaceHandler handler = parserContext.getReaderContext().getNamespaceHandlerResolver()
.resolve(childElement.getNamespaceURI());
AbstractBeanDefinition interceptorDefintiion = ((AbstractBeanDefinition) handlerFromOtherNamespace.parse(
childElement, parserContext));
String beanName = (String) interceptorDefintiion.getMetadataAttribute("interceptorName").getValue();
Assert.hasText("No value for interceptorName provided by namespace handler for element "
+ childElement.getNodeName());
AbstractBeanDefinition interceptorDefinition =
((AbstractBeanDefinition) handler.parse(childElement, parserContext));
String beanName = (String) interceptorDefinition.getMetadataAttribute("interceptorName").getValue();
Assert.hasText("No value for interceptorName provided by namespace handler for element '"
+ childElement.getNodeName() + "'");
return beanName;
}

2
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java
View File

@@ -150,7 +150,7 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost
concurrencyAnnotation.coreSize(), concurrencyAnnotation.maxSize());
concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds());
concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity());
endpoint.addInterceptor(new ConcurrencyInterceptor(concurrencyPolicy));
endpoint.addInterceptor(new ConcurrencyInterceptor(concurrencyPolicy, beanName));
}
return endpoint;
}

2
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java
View File

@@ -76,7 +76,7 @@ public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostP
concurrencyAnnotation.maxSize());
concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds());
concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity());
endpoint.addInterceptor(new ConcurrencyInterceptor(concurrencyPolicy));
endpoint.addInterceptor(new ConcurrencyInterceptor(concurrencyPolicy, beanName));
}
return endpoint;
}

6
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java
View File

@@ -21,6 +21,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
@@ -41,6 +42,7 @@ import org.springframework.integration.message.Message;
import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.util.Assert;
/**
@@ -66,7 +68,7 @@ public class ConcurrencyInterceptor extends EndpointInterceptorAdapter
this.executor = executor;
}
public ConcurrencyInterceptor(ConcurrencyPolicy concurrencyPolicy) {
public ConcurrencyInterceptor(ConcurrencyPolicy concurrencyPolicy, String threadPrefix) {
Assert.notNull(concurrencyPolicy, "ConcurrencyPolicy must not be null");
int core = concurrencyPolicy.getCoreSize();
int max = concurrencyPolicy.getMaxSize();
@@ -75,6 +77,8 @@ public class ConcurrencyInterceptor extends EndpointInterceptorAdapter
BlockingQueue<Runnable> queue = (capacity > 0) ?
new LinkedBlockingQueue<Runnable>(capacity) : new SynchronousQueue<Runnable>();
ThreadPoolExecutor tpe = new ThreadPoolExecutor(core, max, keepAlive, TimeUnit.SECONDS, queue);
tpe.setThreadFactory(new CustomizableThreadFactory(threadPrefix));
tpe.setRejectedExecutionHandler(new CallerRunsPolicy());
this.executor = new ConcurrentTaskExecutor(tpe);
}

9
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java
View File

@@ -21,6 +21,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.util.ErrorHandler;
/**
@@ -52,7 +53,13 @@ public class MessagePublishingErrorHandler implements ErrorHandler {
public final void handle(Throwable t) {
if (logger.isWarnEnabled()) {
logger.warn("failure occurred in messaging task", t);
if (t instanceof MessagingException) {
logger.warn("failure occurred in messaging task with message: "
+ ((MessagingException) t).getFailedMessage(), t);
}
else {
logger.warn("failure occurred in messaging task", t);
}
}
if (this.errorChannel != null) {
try {