From bd5d71120f46698476c2106ca8eebcbfbe447133 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 4 Jul 2008 23:32:27 +0000 Subject: [PATCH] Added CallerRunsPolicy and threadNamePrefix for default MessageBus taskScheduler and ConcurrencyInterceptor. --- .../integration/bus/MessageBus.java | 12 ++++++--- .../config/ConcurrencyInterceptorParser.java | 4 ++- .../config/EndpointInterceptorParser.java | 25 ++++++++++--------- .../HandlerAnnotationPostProcessor.java | 2 +- .../TargetAnnotationPostProcessor.java | 2 +- .../interceptor/ConcurrencyInterceptor.java | 6 ++++- .../MessagePublishingErrorHandler.java | 9 ++++++- 7 files changed, 39 insertions(+), 21 deletions(-) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java index 7fdd843850..6a4e83e606 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -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()); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ConcurrencyInterceptorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ConcurrencyInterceptorParser.java index b073d36fe8..ac6de7ec5d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ConcurrencyInterceptorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/ConcurrencyInterceptorParser.java @@ -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) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/EndpointInterceptorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/EndpointInterceptorParser.java index e58967af89..ea9a827521 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/EndpointInterceptorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/EndpointInterceptorParser.java @@ -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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java index 6640207ca9..6b7741e89f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java @@ -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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java index f079499337..673fdffde9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java @@ -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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java index 81ca3f742e..bcbbfb9fab 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java @@ -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 queue = (capacity > 0) ? new LinkedBlockingQueue(capacity) : new SynchronousQueue(); ThreadPoolExecutor tpe = new ThreadPoolExecutor(core, max, keepAlive, TimeUnit.SECONDS, queue); + tpe.setThreadFactory(new CustomizableThreadFactory(threadPrefix)); + tpe.setRejectedExecutionHandler(new CallerRunsPolicy()); this.executor = new ConcurrentTaskExecutor(tpe); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java index 8015d6c144..c0ae88750e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java @@ -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 {