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 ce93d232e3..41ba925e86 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 @@ -29,6 +29,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.BeansException; +import org.springframework.beans.factory.InitializingBean; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationEvent; @@ -47,18 +48,19 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.factory.ChannelFactory; import org.springframework.integration.channel.factory.QueueChannelFactory; -import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultEndpointRegistry; import org.springframework.integration.endpoint.EndpointRegistry; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.endpoint.MessagingGateway; +import org.springframework.integration.endpoint.MessageProducingEndpoint; import org.springframework.integration.endpoint.SourceEndpoint; +import org.springframework.integration.endpoint.MessageConsumingEndpoint; import org.springframework.integration.endpoint.TargetEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.CommandMessage; -import org.springframework.integration.message.PollCommand; import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.PollCommand; import org.springframework.integration.scheduling.MessagePublishingErrorHandler; import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.MessagingTaskScheduler; @@ -99,8 +101,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio private volatile ScheduledExecutorService executor; - private volatile ConcurrencyPolicy defaultConcurrencyPolicy; - private volatile boolean configureAsyncEventMulticaster = false; private volatile boolean autoCreateChannels = false; @@ -144,14 +144,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio this.executor = executor; } - /** - * Specify the default concurrency policy to be used for any endpoint that - * is registered without an explicitly provided policy of its own. - */ - public void setDefaultConcurrencyPolicy(ConcurrencyPolicy defaultConcurrencyPolicy) { - this.defaultConcurrencyPolicy = defaultConcurrencyPolicy; - } - /** * Set whether to automatically start the bus after initialization. *

Default is 'true'; set this to 'false' to allow for manual startup @@ -220,7 +212,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio if (this.getErrorChannel() == null) { this.setErrorChannel(new DefaultErrorChannel()); } - this.taskScheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel())); this.initialized = true; this.initializing = false; } @@ -263,32 +254,20 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio } public void registerHandler(String name, MessageHandler handler, Subscription subscription) { - this.registerHandler(name, handler, subscription, this.defaultConcurrencyPolicy); - } - - public void registerHandler(String name, MessageHandler handler, Subscription subscription, - ConcurrencyPolicy concurrencyPolicy) { Assert.notNull(handler, "'handler' must not be null"); HandlerEndpoint endpoint = new HandlerEndpoint(handler); - this.doRegisterEndpoint(name, endpoint, subscription, concurrencyPolicy); + this.doRegisterEndpoint(name, endpoint, subscription); } public void registerTarget(String name, MessageTarget target, Subscription subscription) { - this.registerTarget(name, target, subscription, this.defaultConcurrencyPolicy); - } - - public void registerTarget(String name, MessageTarget target, Subscription subscription, - ConcurrencyPolicy concurrencyPolicy) { Assert.notNull(target, "'target' must not be null"); TargetEndpoint endpoint = new TargetEndpoint(target); - this.doRegisterEndpoint(name, endpoint, subscription, concurrencyPolicy); + this.doRegisterEndpoint(name, endpoint, subscription); } - private void doRegisterEndpoint(String name, TargetEndpoint endpoint, Subscription subscription, - ConcurrencyPolicy concurrencyPolicy) { + private void doRegisterEndpoint(String name, TargetEndpoint endpoint, Subscription subscription) { endpoint.setName(name); endpoint.setSubscription(subscription); - endpoint.setConcurrencyPolicy(concurrencyPolicy); this.registerEndpoint(name, endpoint); } @@ -299,10 +278,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio if (endpoint instanceof ChannelRegistryAware) { ((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry); } - if (endpoint instanceof TargetEndpoint) { - this.registerTargetEndpoint((TargetEndpoint) endpoint); - } - else if (endpoint instanceof SourceEndpoint) { + if (endpoint instanceof SourceEndpoint) { this.registerSourceEndpoint(name, (SourceEndpoint) endpoint); } this.endpointRegistry.registerEndpoint(name, endpoint); @@ -314,12 +290,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio } } - private void registerTargetEndpoint(TargetEndpoint endpoint) { - if (endpoint.getConcurrencyPolicy() == null && this.defaultConcurrencyPolicy != null) { - endpoint.setConcurrencyPolicy(this.defaultConcurrencyPolicy); - } - } - public MessageEndpoint unregisterEndpoint(String name) { MessageEndpoint endpoint = this.endpointRegistry.unregisterEndpoint(name); if (endpoint == null) { @@ -357,22 +327,41 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio } private void activateEndpoint(MessageEndpoint endpoint) { - if (endpoint instanceof TargetEndpoint) { - this.activateTargetEndpoint((TargetEndpoint) endpoint); + if (endpoint instanceof MessageProducingEndpoint) { + String channelName = ((MessageProducingEndpoint) endpoint).getOutputChannelName(); + if (channelName != null && this.lookupChannel(channelName) == null) { + if (!this.autoCreateChannels) { + throw new ConfigurationException("Unknown channel '" + channelName + + "' configured as output channel for endpoint '" + endpoint + + "'. Consider enabling the 'autoCreateChannels' option for the message bus."); + } + this.registerChannel(channelName, new QueueChannel()); + } + } + if (endpoint instanceof InitializingBean) { + try { + ((InitializingBean) endpoint).afterPropertiesSet(); + } + catch (Exception e) { + throw new ConfigurationException("failed to initialize endpoint", e); + } + } + if (endpoint instanceof MessageConsumingEndpoint && endpoint instanceof MessageTarget) { + this.activateSubscriber((MessageConsumingEndpoint) endpoint); } } - private void activateTargetEndpoint(TargetEndpoint endpoint) { - Subscription subscription = endpoint.getSubscription(); + private void activateSubscriber(MessageConsumingEndpoint subscriber) { + Subscription subscription = subscriber.getSubscription(); if (subscription == null) { - throw new ConfigurationException("Unable to register endpoint '" + endpoint + throw new ConfigurationException("Unable to register endpoint '" + subscriber + "'. No subscription information is available."); } MessageChannel channel = subscription.getChannel(); if (channel == null) { String channelName = subscription.getChannelName(); if (channelName == null) { - throw new ConfigurationException("endpoint '" + endpoint + throw new ConfigurationException("endpoint '" + subscriber + "' must provide either 'channel' or 'channelName' in its subscription metadata"); } channel = this.lookupChannel(channelName); @@ -388,27 +377,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio this.registerChannel(channelName, channel); } } - if (endpoint instanceof HandlerEndpoint) { - HandlerEndpoint handlerEndpoint = (HandlerEndpoint) endpoint; - String outputChannelName = handlerEndpoint.getOutputChannelName(); - if (outputChannelName != null && this.lookupChannel(outputChannelName) == null) { - if (!this.autoCreateChannels) { - throw new ConfigurationException("Unknown channel '" + outputChannelName - + "' configured as output channel for endpoint '" + endpoint - + "'. Consider enabling the 'autoCreateChannels' option for the message bus."); - } - this.registerChannel(outputChannelName, new QueueChannel()); - } - } - if (!endpoint.hasErrorHandler() && this.getErrorChannel() != null && !this.getErrorChannel().equals(channel)) { - endpoint.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel())); - } - endpoint.afterPropertiesSet(); - this.activateSubscription(channel, endpoint, subscription.getSchedule()); + this.activateSubscription(channel, (MessageTarget) subscriber, subscription.getSchedule()); if (logger.isInfoEnabled()) { - logger - .info("activated subscription to channel '" + channel.getName() + "' for endpoint '" + endpoint - + "'"); + logger.info("activated subscription to channel '" + channel.getName() + + "' for endpoint '" + subscriber + "' of type '" + subscriber.getClass() + "'"); } } @@ -447,7 +419,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio } } - private void activateSubscription(MessageChannel channel, TargetEndpoint targetEndpoint, Schedule schedule) { + private void activateSubscription(MessageChannel channel, MessageTarget target, Schedule schedule) { SubscriptionManager manager = this.subscriptionManagers.get(channel); if (manager == null) { if (logger.isWarnEnabled()) { @@ -456,7 +428,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio } return; } - manager.addTarget(targetEndpoint, schedule); + manager.addTarget(target, schedule); if (this.isRunning() && !manager.isRunning()) { manager.start(); } @@ -479,6 +451,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio this.starting = true; synchronized (this.lifecycleMonitor) { this.activateEndpoints(); + this.taskScheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel())); this.taskScheduler.start(); for (SubscriptionManager manager : this.subscriptionManagers.values()) { manager.start(); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java index 724b2d511c..064b1fddcd 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java @@ -44,6 +44,9 @@ public class MessageBusAwareBeanPostProcessor implements BeanPostProcessor { } public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + if (bean instanceof MessageBusAware) { + ((MessageBusAware) bean).setMessageBus(messageBus); + } return bean; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java index 5b45eec68a..c8fcb6cb41 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java @@ -30,13 +30,10 @@ import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.dispatcher.DirectChannel; import org.springframework.integration.dispatcher.PollingDispatcherTask; -import org.springframework.integration.endpoint.TargetEndpoint; -import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.scheduling.MessagingTaskScheduler; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Schedule; -import org.springframework.integration.util.ErrorHandler; import org.springframework.util.Assert; /** @@ -88,7 +85,7 @@ public class SubscriptionManager { } else if (this.channel instanceof DirectChannel) { if (logger.isInfoEnabled()) { - logger.info("Subscribing to a SynchronousChannel. The provided schedule will be ignored."); + logger.info("Subscribing to a DirectChannel. The provided schedule will be ignored."); } } else if (this.channel.getDispatcherPolicy().isPublishSubscribe()) { @@ -107,16 +104,6 @@ public class SubscriptionManager { } if (this.channel instanceof DirectChannel) { ((DirectChannel) this.channel).subscribe(target); - if (target instanceof TargetEndpoint) { - ((TargetEndpoint) target).setErrorHandler(new ErrorHandler() { - public void handle(Throwable t) { - if (t instanceof MessagingException) { - throw (MessagingException) t; - } - throw new MessagingException("error occurred in handler", t); - } - }); - } return; } PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java index be5369f9ef..14a03a5a7e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java @@ -32,7 +32,9 @@ import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.ConfigurationException; import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor; import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; import org.springframework.transaction.interceptor.NoRollbackRuleAttribute; import org.springframework.transaction.interceptor.RollbackRuleAttribute; @@ -102,6 +104,10 @@ public abstract class IntegrationNamespaceUtils { String txInterceptorBeanName = parseTransactionInterceptor(childElement, parserContext); interceptors.add(new RuntimeBeanReference(txInterceptorBeanName)); } + else if ("concurrency-interceptor".equals(localName)) { + String concurrencyInterceptorBeanName = parseConcurrencyInterceptor(childElement, parserContext); + interceptors.add(new RuntimeBeanReference(concurrencyInterceptorBeanName)); + } } } return interceptors; @@ -161,6 +167,24 @@ public abstract class IntegrationNamespaceUtils { return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry()); } + private static String parseConcurrencyInterceptor(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ConcurrencyInterceptor.class); + String taskExecutorRef = element.getAttribute("task-executor"); + if (StringUtils.hasText(taskExecutorRef)) { + if (element.getAttributes().getLength() != 1) { + parserContext.getReaderContext().error("No other attributes are permitted when " + + "specifying a 'task-executor' reference on the element.", + parserContext.extractSource(element)); + } + builder.addConstructorArgReference(taskExecutorRef); + } + else { + ConcurrencyPolicy policy = IntegrationNamespaceUtils.parseConcurrencyPolicy(element); + builder.addConstructorArgValue(policy); + } + return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry()); + } + /** * Populates the property identified by propertyName on the bean definition * to the value of the attribute specified by attributeName, if that diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageEndpointBeanPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageEndpointBeanPostProcessor.java index 26afeb68a8..7b8761b689 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageEndpointBeanPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageEndpointBeanPostProcessor.java @@ -46,7 +46,7 @@ public class MessageEndpointBeanPostProcessor implements BeanPostProcessor { if (interceptors.size() > 0) { ProxyFactory proxyFactory = new ProxyFactory(endpoint); for (Advice interceptor : interceptors) { - proxyFactory.addAdvisor(new EndpointInvokeMethodAdvisor(interceptor)); + proxyFactory.addAdvisor(new EndpointMethodAdvisor(interceptor)); } bean = proxyFactory.getProxy(); } @@ -56,16 +56,16 @@ public class MessageEndpointBeanPostProcessor implements BeanPostProcessor { @SuppressWarnings("serial") - private static class EndpointInvokeMethodAdvisor extends StaticMethodMatcherPointcutAdvisor { + private static class EndpointMethodAdvisor extends StaticMethodMatcherPointcutAdvisor { - EndpointInvokeMethodAdvisor(Advice advice) { + EndpointMethodAdvisor(Advice advice) { super(advice); } @SuppressWarnings("unchecked") public boolean matches(Method method, Class clazz) { - return method.getName().equals("invoke") + return (method.getName().equals("invoke") || method.getName().equals("send")) && method.getParameterTypes().length == 1 && method.getParameterTypes()[0].equals(Message.class); } 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 2e1d8e893e..47e9218237 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 @@ -41,6 +41,7 @@ import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerChain; import org.springframework.integration.handler.config.DefaultMessageHandlerCreator; @@ -111,6 +112,7 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost return handler; } + @SuppressWarnings("unchecked") protected MessageHandler processResults(List results) { MessageHandlerChain handlerChain = new MessageHandlerChain(); for (MessageHandler handler : results) { @@ -146,7 +148,7 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost concurrencyAnnotation.coreSize(), concurrencyAnnotation.maxSize()); concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds()); concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity()); - endpoint.setConcurrencyPolicy(concurrencyPolicy); + endpoint.addInterceptor(new ConcurrencyInterceptor(concurrencyPolicy)); } 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 633aa8e1b6..4fdfa5079a 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 @@ -28,6 +28,7 @@ import org.springframework.integration.bus.MessageBus; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.endpoint.TargetEndpoint; +import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor; import org.springframework.integration.handler.MethodInvokingTarget; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.scheduling.Subscription; @@ -70,7 +71,7 @@ public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostP concurrencyAnnotation.maxSize()); concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds()); concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity()); - endpoint.setConcurrencyPolicy(concurrencyPolicy); + endpoint.addInterceptor(new ConcurrencyInterceptor(concurrencyPolicy)); } return endpoint; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index 670f50110f..b6bb72f416 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -25,7 +25,6 @@ - @@ -304,12 +303,13 @@ - + - Defines a concurrency policy. + Defines a ConcurrencyInterceptor for endpoints. + @@ -415,11 +415,9 @@ - - @@ -453,6 +451,7 @@ + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index d4f48c7b33..8d94a5dec3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -59,18 +59,23 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware return (this.name != null) ? this.name : super.toString(); } + public void addInterceptor(Object interceptor) { + if (interceptor instanceof Advice) { + this.interceptors.add((Advice) interceptor); + } + else if (interceptor instanceof EndpointInterceptor) { + this.interceptors.add(new EndpointMethodInterceptor((EndpointInterceptor) interceptor)); + } + else { + throw new ConfigurationException("Interceptor must implement either " + + "'" + Advice.class.getName() + "' or '" + EndpointInterceptor.class.getName() + "'."); + } + } + public void setInterceptors(List interceptors) { + this.interceptors.clear(); for (Object interceptor : interceptors) { - if (interceptor instanceof Advice) { - this.interceptors.add((Advice) interceptor); - } - else if (interceptor instanceof EndpointInterceptor) { - this.interceptors.add(new EndpointMethodInterceptor((EndpointInterceptor) interceptor)); - } - else { - throw new ConfigurationException("Each interceptor element must implement either " - + "'" + Advice.class.getName() + "' or '" + EndpointInterceptor.class.getName() + "'."); - } + this.addInterceptor(interceptor); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ConcurrentTarget.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ConcurrentTarget.java deleted file mode 100644 index 90ec55dbf9..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ConcurrentTarget.java +++ /dev/null @@ -1,108 +0,0 @@ -/* - * Copyright 2002-2008 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.endpoint; - -import java.util.concurrent.ExecutorService; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.factory.DisposableBean; -import org.springframework.integration.handler.MessageHandlerNotRunningException; -import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.util.ErrorHandler; -import org.springframework.util.Assert; - -/** - * A {@link MessageTarget} implementation that encapsulates an Executor and delegates - * to a wrapped target for concurrent, asynchronous message handling. - * - * @author Mark Fisher - */ -public class ConcurrentTarget implements MessageTarget, DisposableBean { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private final MessageTarget target; - - private final ExecutorService executor; - - private volatile ErrorHandler errorHandler; - - - public ConcurrentTarget(MessageTarget target, ExecutorService executor) { - Assert.notNull(target, "'target' must not be null"); - Assert.notNull(executor, "'executor' must not be null"); - this.target = target; - this.executor = executor; - } - - - public void setErrorHandler(ErrorHandler errorHandler) { - this.errorHandler = errorHandler; - } - - public void destroy() { - this.executor.shutdown(); - } - - public boolean send(Message message) { - if (this.executor.isShutdown()) { - throw new MessageHandlerNotRunningException(message); - } - try { - this.executor.execute(new TargetTask(message)); - return true; - } - catch (RuntimeException e) { - throw new MessageHandlerRejectedExecutionException(message, e); - } - } - - - private class TargetTask implements Runnable { - - private Message message; - - TargetTask(Message message) { - this.message = message; - } - - public void run() { - try { - if (!target.send(this.message)) { - throw new MessageDeliveryException(message, "failed to send message to target"); - } - } - catch (Throwable t) { - if (logger.isDebugEnabled()) { - logger.debug("error occurred in handler execution", t); - } - if (errorHandler != null) { - errorHandler.handle(t); - } - else if (logger.isWarnEnabled() && !logger.isDebugEnabled()) { - logger.warn("error occurred in handler execution", t); - } - } - } - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java index 061467c8ef..e492220894 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java @@ -35,7 +35,7 @@ import org.springframework.util.StringUtils; * * @author Mark Fisher */ -public class HandlerEndpoint extends TargetEndpoint { +public class HandlerEndpoint extends TargetEndpoint implements MessageProducingEndpoint { private volatile MessageHandler handler; @@ -81,10 +81,6 @@ public class HandlerEndpoint extends TargetEndpoint { this.outputChannelName = outputChannelName; } - public String getOutputChannelName() { - return this.outputChannelName; - } - public void setReturnAddressOverrides(boolean returnAddressOverrides) { this.returnAddressOverrides = returnAddressOverrides; } @@ -132,7 +128,7 @@ public class HandlerEndpoint extends TargetEndpoint { return null; } - private MessageChannel getOutputChannel() { + public MessageChannel getOutputChannel() { ChannelRegistry registry = this.getChannelRegistry(); if (this.outputChannelName != null && registry != null) { return registry.lookupChannel(this.outputChannelName); @@ -140,6 +136,10 @@ public class HandlerEndpoint extends TargetEndpoint { return null; } + public String getOutputChannelName() { + return this.outputChannelName; + } + private static class HandlerInvokingTarget implements MessageTarget { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageConsumingEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageConsumingEndpoint.java new file mode 100644 index 0000000000..ce721e0633 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageConsumingEndpoint.java @@ -0,0 +1,28 @@ +/* + * Copyright 2002-2008 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.endpoint; + +import org.springframework.integration.scheduling.Subscription; + +/** + * @author Mark Fisher + */ +public interface MessageConsumingEndpoint { + + Subscription getSubscription(); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducingEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducingEndpoint.java new file mode 100644 index 0000000000..d661232512 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducingEndpoint.java @@ -0,0 +1,30 @@ +/* + * Copyright 2002-2008 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.endpoint; + +import org.springframework.integration.channel.MessageChannel; + +/** + * @author Mark Fisher + */ +public interface MessageProducingEndpoint { + + String getOutputChannelName(); + + MessageChannel getOutputChannel(); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java index 3e37dc1148..811c4101c9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java @@ -16,26 +16,16 @@ package org.springframework.integration.endpoint; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; - import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.Lifecycle; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.handler.MessageHandlerNotRunningException; -import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.selector.MessageSelector; import org.springframework.integration.scheduling.Subscription; -import org.springframework.integration.util.ErrorHandler; import org.springframework.util.Assert; /** @@ -43,16 +33,13 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class TargetEndpoint extends AbstractEndpoint implements MessageTarget, ChannelRegistryAware, InitializingBean, Lifecycle { +public class TargetEndpoint extends AbstractEndpoint + implements MessageTarget, MessageConsumingEndpoint, ChannelRegistryAware, InitializingBean, Lifecycle { private volatile MessageTarget target; private volatile Subscription subscription; - private volatile ConcurrencyPolicy concurrencyPolicy; - - private volatile ErrorHandler errorHandler; - private volatile MessageSelector selector; private volatile ChannelRegistry channelRegistry; @@ -92,22 +79,6 @@ public class TargetEndpoint extends AbstractEndpoint implements MessageTarget, C this.subscription = subscription; } - public ConcurrencyPolicy getConcurrencyPolicy() { - return this.concurrencyPolicy; - } - - public void setConcurrencyPolicy(ConcurrencyPolicy concurrencyPolicy) { - this.concurrencyPolicy = concurrencyPolicy; - } - - public void setErrorHandler(ErrorHandler errorHandler) { - this.errorHandler = errorHandler; - } - - public boolean hasErrorHandler() { - return (this.errorHandler != null); - } - /** * Set the channel registry to use for looking up channels by name. */ @@ -123,18 +94,6 @@ public class TargetEndpoint extends AbstractEndpoint implements MessageTarget, C if (this.target instanceof ChannelRegistryAware && this.channelRegistry != null) { ((ChannelRegistryAware) this.target).setChannelRegistry(this.channelRegistry); } - if (this.concurrencyPolicy != null && !(this.target instanceof ConcurrentTarget)) { - int capacity = this.concurrencyPolicy.getQueueCapacity(); - BlockingQueue queue = (capacity < 1) ? new SynchronousQueue() : new ArrayBlockingQueue(capacity); - ExecutorService executor = new ThreadPoolExecutor(this.concurrencyPolicy.getCoreSize(), this.concurrencyPolicy.getMaxSize(), - this.concurrencyPolicy.getKeepAliveSeconds(), TimeUnit.SECONDS, queue); - this.target = new ConcurrentTarget(this.target, executor); - } - if (this.target instanceof ConcurrentTarget) { - if (this.errorHandler != null) { - ((ConcurrentTarget) this.target).setErrorHandler(this.errorHandler); - } - } this.initialized = true; } @@ -180,23 +139,7 @@ public class TargetEndpoint extends AbstractEndpoint implements MessageTarget, C if (this.selector != null && !this.selector.accept(message)) { return false; } - try { - return this.target.send(message); - } - catch (MessageHandlerRejectedExecutionException e) { - throw e; - } - catch (Throwable t) { - if (this.errorHandler == null) { - if (t instanceof MessageHandlingException) { - throw (MessageHandlingException) t; - } - throw new MessageHandlingException(message, - "error occurred in endpoint, and no 'errorHandler' available", t); - } - this.errorHandler.handle(t); - return false; - } + return this.target.send(message); } @Override 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 new file mode 100644 index 0000000000..81ca3f742e --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java @@ -0,0 +1,137 @@ +/* + * Copyright 2002-2008 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.endpoint.interceptor; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.aopalliance.intercept.MethodInvocation; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.Lifecycle; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.bus.MessageBusAware; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.EndpointInterceptor; +import org.springframework.integration.handler.MessageHandlerNotRunningException; +import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; +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.util.Assert; + +/** + * An {@link EndpointInterceptor} implementation that delegates to a + * {@link TaskExecutor} for concurrent, asynchronous message handling. + * + * @author Mark Fisher + */ +public class ConcurrencyInterceptor extends EndpointInterceptorAdapter + implements DisposableBean, InitializingBean, MessageBusAware { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final TaskExecutor executor; + + private volatile ErrorHandler errorHandler; + + private volatile MessageBus messageBus; + + + public ConcurrencyInterceptor(TaskExecutor executor) { + Assert.notNull(executor, "TaskExecutor must not be null"); + this.executor = executor; + } + + public ConcurrencyInterceptor(ConcurrencyPolicy concurrencyPolicy) { + Assert.notNull(concurrencyPolicy, "ConcurrencyPolicy must not be null"); + int core = concurrencyPolicy.getCoreSize(); + int max = concurrencyPolicy.getMaxSize(); + int keepAlive = concurrencyPolicy.getKeepAliveSeconds(); + int capacity = concurrencyPolicy.getQueueCapacity(); + BlockingQueue queue = (capacity > 0) ? + new LinkedBlockingQueue(capacity) : new SynchronousQueue(); + ThreadPoolExecutor tpe = new ThreadPoolExecutor(core, max, keepAlive, TimeUnit.SECONDS, queue); + this.executor = new ConcurrentTaskExecutor(tpe); + } + + + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + + public void setMessageBus(MessageBus messageBus) { + this.messageBus = messageBus; + } + + public void afterPropertiesSet() throws Exception { + if (this.errorHandler == null) { + MessageChannel errorChannel = this.messageBus.getErrorChannel(); + if (errorChannel != null) { + this.errorHandler = new MessagePublishingErrorHandler(errorChannel); + } + } + } + + public void destroy() throws Exception { + if (this.executor instanceof DisposableBean) { + ((DisposableBean) this.executor).destroy(); + } + } + + @Override + public boolean aroundInvoke(final MethodInvocation invocation) { + final Message message = (Message) invocation.getArguments()[0]; + if (invocation.getThis() instanceof Lifecycle && !((Lifecycle) invocation.getThis()).isRunning()) { + throw new MessageHandlerNotRunningException(message); + } + try { + this.executor.execute(new Runnable() { + public void run() { + try { + invocation.proceed(); + } + catch (Throwable t) { + if (logger.isDebugEnabled()) { + logger.debug("error occurred in handler execution", t); + } + if (errorHandler != null) { + errorHandler.handle(t); + } + else if (logger.isWarnEnabled() && !logger.isDebugEnabled()) { + logger.warn("error occurred in handler execution", t); + } + } + } + }); + return true; + } + catch (RuntimeException e) { + throw new MessageHandlerRejectedExecutionException(message, e); + } + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java index 30945c5822..473aa96b05 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java @@ -80,7 +80,7 @@ public class DirectChannelSubscriptionTests { bus.stop(); } - @Test(expected=MessagingException.class) + @Test(expected=RuntimeException.class) public void testExceptionThrownFromRegisteredEndpoint() { QueueChannel errorChannel = new QueueChannel(); bus.setErrorChannel(errorChannel); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java index e3df0d90d7..0d8bb08236 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -31,14 +31,11 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.channel.RendezvousChannel; -import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.SourceEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.PollingSchedule; @@ -184,41 +181,6 @@ public class MessageBusTests { bus.stop(); } - @Test - public void testDefaultConcurrencyPolicy() throws InterruptedException { - MessageBus bus = new MessageBus(); - bus.setDefaultConcurrencyPolicy(new ConcurrencyPolicy(1, 3)); - final CountDownLatch latch = new CountDownLatch(3); - MessageHandler testHandler = new MessageHandler() { - public Message handle(Message message) { - latch.countDown(); - try { - Thread.sleep(5000); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - return null; - } - }; - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - dispatcherPolicy.setRejectionLimit(1); - dispatcherPolicy.setRetryInterval(0); - RendezvousChannel testChannel = new RendezvousChannel(dispatcherPolicy); - bus.registerChannel("testChannel", testChannel); - bus.registerHandler("testHandler", testHandler, new Subscription(testChannel)); - bus.start(); - for (int i = 0; i < 4; i++) { - assertTrue(testChannel.send(new StringMessage("test-"+ i), 1000)); - } - latch.await(1000, TimeUnit.MILLISECONDS); - assertEquals(0, latch.getCount()); - MessageChannel errorChannel = bus.getErrorChannel(); - Message errorMessage = errorChannel.receive(500); - assertNotNull(errorMessage); - assertEquals(MessageDeliveryException.class, errorMessage.getPayload().getClass()); - } - @Test(expected = BeanCreationException.class) public void testMultipleMessageBusBeans() { new ClassPathXmlApplicationContext("multipleMessageBusBeans.xml", this.getClass()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java index 4175d8a382..7632a26934 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java @@ -27,19 +27,23 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; -import org.springframework.integration.bus.SubscriptionManager; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.Lifecycle; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.MessageEndpointBeanPostProcessor; +import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; +import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.selector.PayloadTypeSelector; import org.springframework.integration.scheduling.MessagePublishingErrorHandler; import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; @@ -99,12 +103,12 @@ public class SubscriptionManagerTests { MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); QueueChannel channel = new QueueChannel(); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - HandlerEndpoint inactiveEndpoint = createEndpoint(handler1, true); + MessageTarget inactiveEndpoint = createEndpoint(handler1, true); manager.addTarget(inactiveEndpoint); manager.addTarget(createEndpoint(handler2, true)); manager.addTarget(createEndpoint(handler3, true)); manager.start(); - inactiveEndpoint.stop(); + ((Lifecycle) inactiveEndpoint).stop(); channel.send(new StringMessage(1, "test")); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); @@ -123,12 +127,12 @@ public class SubscriptionManagerTests { MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - HandlerEndpoint inactiveEndpoint = createEndpoint(handler2, true); + MessageTarget inactiveEndpoint = createEndpoint(handler2, true); manager.addTarget(createEndpoint(handler1, true)); manager.addTarget(inactiveEndpoint); manager.addTarget(createEndpoint(handler3, true)); manager.start(); - inactiveEndpoint.stop(); + ((Lifecycle) inactiveEndpoint).stop(); channel.send(new StringMessage(1, "test")); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); @@ -434,9 +438,7 @@ public class SubscriptionManagerTests { channel.send(new StringMessage(1, "test")); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); - endpoint1.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); - endpoint2.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); endpoint1.setMessageSelector(new PayloadTypeSelector(Integer.class)); endpoint2.setMessageSelector(new PayloadTypeSelector(String.class)); manager.addTarget(endpoint1); @@ -449,12 +451,19 @@ public class SubscriptionManagerTests { } - private static HandlerEndpoint createEndpoint(MessageHandler handler, boolean asynchronous) { - HandlerEndpoint endpoint = new HandlerEndpoint(handler); + private static MessageTarget createEndpoint(MessageHandler handler, boolean asynchronous) { + MessageTarget endpoint = new HandlerEndpoint(handler); if (asynchronous) { - endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); + MessageEndpointBeanPostProcessor postProcessor = new MessageEndpointBeanPostProcessor(); + ((AbstractEndpoint) endpoint).addInterceptor(new ConcurrencyInterceptor(new ConcurrencyPolicy(1, 1))); + endpoint = (MessageTarget) postProcessor.postProcessAfterInitialization(endpoint, "test-endpoint"); + } + try { + ((InitializingBean) endpoint).afterPropertiesSet(); + } + catch (Exception e) { + throw new RuntimeException("failed to initialize endpoint", e); } - endpoint.afterPropertiesSet(); return endpoint; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java index ed40dc501b..fb11e444e0 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -30,11 +30,8 @@ import org.springframework.context.Lifecycle; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.endpoint.ConcurrencyPolicy; -import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.MessageTarget; @@ -81,30 +78,6 @@ public class EndpointParserTests { assertEquals("test-1-2-3", reply.getPayload()); } - @Test - public void testDefaultConcurrency() throws InterruptedException { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "endpointConcurrencyTests.xml", this.getClass()); - HandlerEndpoint endpoint = (HandlerEndpoint) context.getBean("defaultConcurrencyEndpoint"); - ConcurrencyPolicy concurrencyPolicy = endpoint.getConcurrencyPolicy(); - assertEquals(ConcurrencyPolicy.DEFAULT_CORE_SIZE, concurrencyPolicy.getCoreSize()); - assertEquals(ConcurrencyPolicy.DEFAULT_MAX_SIZE, concurrencyPolicy.getMaxSize()); - assertEquals(ConcurrencyPolicy.DEFAULT_QUEUE_CAPACITY, concurrencyPolicy.getQueueCapacity()); - assertEquals(ConcurrencyPolicy.DEFAULT_KEEP_ALIVE_SECONDS, concurrencyPolicy.getKeepAliveSeconds()); - } - - @Test - public void testConfiguredConcurrency() throws InterruptedException { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "endpointConcurrencyTests.xml", this.getClass()); - HandlerEndpoint endpoint = (HandlerEndpoint) context.getBean("configuredConcurrencyEndpoint"); - ConcurrencyPolicy concurrencyPolicy = endpoint.getConcurrencyPolicy(); - assertEquals(7, concurrencyPolicy.getCoreSize()); - assertEquals(77, concurrencyPolicy.getMaxSize()); - assertEquals(777, concurrencyPolicy.getQueueCapacity()); - assertEquals(7777, concurrencyPolicy.getKeepAliveSeconds()); - } - @Test public void testEndpointWithSelectorAccepts() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( @@ -132,21 +105,6 @@ public class EndpointParserTests { assertFalse(endpoint.send(message)); } - @Test - public void testCustomErrorHandler() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "endpointWithErrorHandler.xml", this.getClass()); - MessageTarget endpoint = (MessageTarget) context.getBean("endpoint"); - TestErrorHandler errorHandler = (TestErrorHandler) context.getBean("errorHandler"); - assertNull(errorHandler.getLastError()); - Message message = new StringMessage("test"); - endpoint.send(message); - Throwable error = errorHandler.getLastError(); - assertEquals(MessageHandlingException.class, error.getClass()); - MessageHandlingException exception = (MessageHandlingException) error; - assertEquals(message, exception.getFailedMessage()); - } - @Test public void testCustomReplyHandler() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java index 02b3f491a3..d661cad0b9 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java @@ -38,7 +38,6 @@ import org.springframework.integration.bus.interceptor.TestMessageBusStartInterc import org.springframework.integration.bus.interceptor.TestMessageBusStopInterceptor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.dispatcher.DirectChannel; -import org.springframework.integration.endpoint.TargetEndpoint; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; import org.springframework.integration.scheduling.Subscription; @@ -128,24 +127,6 @@ public class MessageBusParserTests { bus.stop(); } - @Test - public void testDefaultConcurrency() { - ApplicationContext context = new ClassPathXmlApplicationContext( - "messageBusWithDefaultConcurrencyTests.xml", this.getClass()); - TargetEndpoint endpoint1 = (TargetEndpoint) context.getBean("endpoint1"); - assertEquals(4, endpoint1.getConcurrencyPolicy().getCoreSize()); - assertEquals(7, endpoint1.getConcurrencyPolicy().getMaxSize()); - } - - @Test - public void testExplicitConcurrencyTakesPrecedence() { - ApplicationContext context = new ClassPathXmlApplicationContext( - "messageBusWithDefaultConcurrencyTests.xml", this.getClass()); - TargetEndpoint endpoint2 = (TargetEndpoint) context.getBean("endpoint2"); - assertEquals(14, endpoint2.getConcurrencyPolicy().getCoreSize()); - assertEquals(17, endpoint2.getConcurrencyPolicy().getMaxSize()); - } - @Test public void testMessageBusAwareAutomaticallyAddedByNamespace() { ApplicationContext context = new ClassPathXmlApplicationContext( diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java index 2213dade1f..0f12c8286d 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java @@ -26,11 +26,14 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import org.aopalliance.aop.Advice; import org.junit.Test; import org.springframework.aop.framework.ProxyFactory; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationContext; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -49,13 +52,14 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; +import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Schedule; +import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; /** * @author Mark Fisher @@ -178,11 +182,18 @@ public class MessagingAnnotationPostProcessorTests { ConcurrencyAnnotationTestBean testBean = new ConcurrencyAnnotationTestBean(); postProcessor.postProcessAfterInitialization(testBean, "testBean"); HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean.MessageHandler.endpoint"); - ConcurrencyPolicy concurrencyPolicy = endpoint.getConcurrencyPolicy(); - assertEquals(17, concurrencyPolicy.getCoreSize()); - assertEquals(42, concurrencyPolicy.getMaxSize()); - assertEquals(11, concurrencyPolicy.getQueueCapacity()); - assertEquals(123, concurrencyPolicy.getKeepAliveSeconds()); + assertEquals(1, endpoint.getInterceptors().size()); + Advice interceptor = endpoint.getInterceptors().get(0); + DirectFieldAccessor accessor = new DirectFieldAccessor(interceptor); + ConcurrencyInterceptor concurrencyInterceptor = (ConcurrencyInterceptor) + accessor.getPropertyValue("interceptor"); + accessor = new DirectFieldAccessor(concurrencyInterceptor); + ConcurrentTaskExecutor cte = (ConcurrentTaskExecutor) accessor.getPropertyValue("executor"); + ThreadPoolExecutor executor = (ThreadPoolExecutor) cte.getConcurrentExecutor(); + assertEquals(17, executor.getCorePoolSize()); + assertEquals(42, executor.getMaximumPoolSize()); + assertEquals(123, executor.getKeepAliveTime(TimeUnit.SECONDS)); + assertEquals(11, executor.getQueue().remainingCapacity()); } @Test(expected=IllegalArgumentException.class) diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java index 48e2a08c00..d2d73d6f93 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java @@ -36,11 +36,9 @@ import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerNotRunningException; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.selector.MessageSelector; import org.springframework.integration.message.selector.MessageSelectorChain; -import org.springframework.integration.util.ErrorHandler; /** * @author Mark Fisher @@ -141,48 +139,6 @@ public class HandlerEndpointTests { endpoint.stop(); } - @Test - public void testCustomErrorHandler() throws InterruptedException { - final CountDownLatch latch = new CountDownLatch(2); - HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.rejectingCountDownHandler(latch)); - endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); - endpoint.setErrorHandler(new ErrorHandler() { - public void handle(Throwable t) { - latch.countDown(); - } - }); - endpoint.start(); - endpoint.send(new StringMessage("test")); - latch.await(500, TimeUnit.MILLISECONDS); - assertEquals("both handler and errorHandler should have been invoked", 0, latch.getCount()); - } - - @Test - public void testConcurrentHandlerWithDefaultReplyChannel() throws InterruptedException { - MessageChannel replyChannel = new QueueChannel(); - ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - channelRegistry.registerChannel("replyChannel", replyChannel); - final CountDownLatch latch = new CountDownLatch(1); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { - latch.countDown(); - return new StringMessage("123", "hello " + message.getPayload()); - } - }; - HandlerEndpoint endpoint = new HandlerEndpoint(handler); - endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); - endpoint.setChannelRegistry(channelRegistry); - endpoint.setOutputChannelName("replyChannel"); - endpoint.start(); - endpoint.send(new StringMessage(1, "test")); - latch.await(500, TimeUnit.MILLISECONDS); - endpoint.stop(); - assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); - Message reply = replyChannel.receive(100); - assertNotNull(reply); - assertEquals("hello test", reply.getPayload()); - } - @Test public void testHandlerReturnsNull() throws InterruptedException { MessageChannel replyChannel = new QueueChannel(); @@ -207,111 +163,6 @@ public class HandlerEndpointTests { assertNull(reply); } - @Test - public void testConcurrentHandlerReturnsNull() throws InterruptedException { - MessageChannel replyChannel = new QueueChannel(); - ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - channelRegistry.registerChannel("replyChannel", replyChannel); - final CountDownLatch latch = new CountDownLatch(1); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { - latch.countDown(); - return null; - } - }; - HandlerEndpoint endpoint = new HandlerEndpoint(handler); - endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); - endpoint.setChannelRegistry(channelRegistry); - endpoint.setOutputChannelName("replyChannel"); - endpoint.start(); - endpoint.send(new StringMessage(1, "test")); - latch.await(500, TimeUnit.MILLISECONDS); - endpoint.stop(); - assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); - Message reply = replyChannel.receive(0); - assertNull(reply); - } - - @Test - public void testConcurrentHandlerWithExplicitReplyChannel() throws InterruptedException { - MessageChannel replyChannel = new QueueChannel(); - ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - channelRegistry.registerChannel("replyChannel", replyChannel); - final CountDownLatch latch = new CountDownLatch(1); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { - latch.countDown(); - return new StringMessage("123", "hello " + message.getPayload()); - } - }; - HandlerEndpoint endpoint = new HandlerEndpoint(handler); - endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); - endpoint.setChannelRegistry(channelRegistry); - endpoint.start(); - StringMessage message = new StringMessage(1, "test"); - message.getHeader().setReturnAddress("replyChannel"); - endpoint.send(message); - latch.await(500, TimeUnit.MILLISECONDS); - assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); - Message reply = replyChannel.receive(100); - endpoint.stop(); - assertNotNull(reply); - assertEquals("hello test", reply.getPayload()); - } - - @Test - public void testGeneratedConcurrentHandlerWithDefaultReplyChannel() throws InterruptedException { - MessageChannel replyChannel = new QueueChannel(); - ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - channelRegistry.registerChannel("replyChannel", replyChannel); - final CountDownLatch latch = new CountDownLatch(1); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { - latch.countDown(); - return new StringMessage("123", "hello " + message.getPayload()); - } - }; - HandlerEndpoint endpoint = new HandlerEndpoint(handler); - endpoint.setChannelRegistry(channelRegistry); - endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(3, 14)); - endpoint.setOutputChannelName("replyChannel"); - endpoint.start(); - endpoint.send(new StringMessage(1, "test")); - latch.await(500, TimeUnit.MILLISECONDS); - endpoint.stop(); - assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); - Message reply = replyChannel.receive(100); - assertNotNull(reply); - assertEquals("hello test", reply.getPayload()); - } - - @Test - public void testGeneratedConcurrentHandlerWithExplicitReplyChannel() throws InterruptedException { - MessageChannel replyChannel = new QueueChannel(); - ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - channelRegistry.registerChannel("replyChannel", replyChannel); - final CountDownLatch latch = new CountDownLatch(1); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { - latch.countDown(); - return new StringMessage("123", "hello " + message.getPayload()); - } - }; - HandlerEndpoint endpoint = new HandlerEndpoint(handler); - endpoint.setChannelRegistry(channelRegistry); - endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(3, 14)); - endpoint.start(); - StringMessage message = new StringMessage(1, "test"); - message.getHeader().setReturnAddress("replyChannel"); - endpoint.send(message); - latch.await(500, TimeUnit.MILLISECONDS); - endpoint.stop(); - assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); - Message reply = replyChannel.receive(100); - assertNotNull(reply); - assertEquals("hello test", reply.getPayload()); - } - @Test(expected=MessageHandlerNotRunningException.class) public void testEndpointDoesNotHandleMessagesWhenNotYetStarted() { HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.nullHandler()); @@ -438,84 +289,6 @@ public class HandlerEndpointTests { endpoint.stop(); } - @Test - public void testDefaultOutputChannelTimeoutSendsToErrorHandler() { - QueueChannel output = new QueueChannel(1); - ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - channelRegistry.registerChannel("output", output); - HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { - public Message handle(Message message) { - return message; - } - }); - endpoint.setOutputChannelName("output"); - endpoint.setChannelRegistry(channelRegistry); - TestErrorHandler errorHandler = new TestErrorHandler(); - endpoint.setErrorHandler(errorHandler); - endpoint.setReplyTimeout(0); - endpoint.start(); - endpoint.send(new StringMessage("test1")); - assertNull(errorHandler.getLastError()); - endpoint.send(new StringMessage("test2")); - Throwable error = errorHandler.getLastError(); - assertNotNull(error); - assertEquals(MessageDeliveryException.class, error.getClass()); - assertEquals("test2", ((MessageDeliveryException) error).getFailedMessage().getPayload()); - } - - @Test - public void testReturnAddressChannelTimeoutSendsToErrorHandler() { - QueueChannel replyChannel = new QueueChannel(1); - HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { - public Message handle(Message message) { - return message; - } - }); - TestErrorHandler errorHandler = new TestErrorHandler(); - endpoint.setErrorHandler(errorHandler); - endpoint.setReplyTimeout(0); - endpoint.start(); - Message message1 = new StringMessage("test1"); - message1.getHeader().setReturnAddress(replyChannel); - endpoint.send(message1); - assertNull(errorHandler.getLastError()); - Message message2 = new StringMessage("test2"); - message2.getHeader().setReturnAddress(replyChannel); - endpoint.send(message2); - Throwable error = errorHandler.getLastError(); - assertNotNull(error); - assertEquals(MessageDeliveryException.class, error.getClass()); - assertEquals(message2, ((MessageDeliveryException) error).getFailedMessage()); - } - - @Test - public void testReturnAddressChannelNameTimeoutSendsToErrorHandler() { - QueueChannel replyChannel = new QueueChannel(1); - ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - channelRegistry.registerChannel("replyChannel", replyChannel); - HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { - public Message handle(Message message) { - return message; - } - }); - endpoint.setChannelRegistry(channelRegistry); - TestErrorHandler errorHandler = new TestErrorHandler(); - endpoint.setErrorHandler(errorHandler); - endpoint.setReplyTimeout(10); - endpoint.start(); - Message message1 = new StringMessage("test1"); - message1.getHeader().setReturnAddress("replyChannel"); - endpoint.send(message1); - assertNull(errorHandler.getLastError()); - Message message2 = new StringMessage("test2"); - message2.getHeader().setReturnAddress("replyChannel"); - endpoint.send(message2); - Throwable error = errorHandler.getLastError(); - assertNotNull(error); - assertEquals(MessageDeliveryException.class, error.getClass()); - assertEquals(message2, ((MessageDeliveryException) error).getFailedMessage()); - } - @Test public void testCorrelationId() { QueueChannel replyChannel = new QueueChannel(1); @@ -551,19 +324,4 @@ public class HandlerEndpointTests { assertEquals("ABC-123", correlationId); } - - private static class TestErrorHandler implements ErrorHandler { - - private volatile Throwable lastError; - - - public void handle(Throwable t) { - this.lastError = t; - } - - Throwable getLastError() { - return this.lastError; - } - } - } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java index fa370ad608..3da678bbe3 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java @@ -80,7 +80,7 @@ public class ReturnAddressTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "returnAddressTests.xml", this.getClass()); MessageChannel channel3 = (MessageChannel) context.getBean("channel3"); - MessageChannel errorChannel = (MessageChannel) context.getBean("errorChannel"); + MessageChannel errorChannel = (MessageChannel) context.getBean("customErrorChannel"); context.start(); StringMessage message = new StringMessage("*"); channel3.send(message); @@ -93,7 +93,7 @@ public class ReturnAddressTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "returnAddressTests.xml", this.getClass()); MessageChannel channel3 = (MessageChannel) context.getBean("channel3WithOverride"); - MessageChannel errorChannel = (MessageChannel) context.getBean("errorChannel"); + MessageChannel errorChannel = (MessageChannel) context.getBean("customErrorChannel"); context.start(); StringMessage message = new StringMessage("*"); channel3.send(message); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/returnAddressTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/returnAddressTests.xml index 26308dde02..f82040f66f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/returnAddressTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/returnAddressTests.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd"> - + @@ -16,7 +16,7 @@ - +