From 3a53c35f0bec85c320620595c9fdf135655bf625 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 17 Jan 2008 21:42:51 +0000 Subject: [PATCH] Refactored so that endpoint is reponsible for managing ConcurrentHandlers rather than dispatchers having that responsibility. Also providing errorHandler for endpoints. That errorHandler is also used for asynchronous handler task execution. --- .../integration/bus/MessageBus.java | 51 ++-- .../integration/channel/MessageChannel.java | 5 + .../dispatcher/DefaultMessageDispatcher.java | 66 ++--- .../MessageSelectorRejectedException.java | 4 + .../endpoint/ConcurrencyPolicy.java | 9 + .../endpoint/DefaultMessageEndpoint.java | 146 +++++++++-- .../integration/endpoint/MessageEndpoint.java | 3 +- .../handler/ConcurrentHandler.java | 196 ++++++++++++++ .../handler/InterceptingMessageHandler.java | 2 +- .../handler/PooledMessageHandler.java | 225 ---------------- .../bus/DefaultMessageDispatcherTests.java | 241 +++++++++--------- .../endpoint/DefaultMessageEndpointTests.java | 4 + .../integration/handler/TestHandlers.java | 36 ++- 13 files changed, 555 insertions(+), 433 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/PooledMessageHandler.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index a941bf600a..2f4be97634 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -42,7 +42,6 @@ import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.handler.PooledMessageHandler; import org.springframework.integration.scheduling.MessagePublishingErrorHandler; import org.springframework.integration.scheduling.MessagingTaskScheduler; import org.springframework.integration.scheduling.MessagingTaskSchedulerAware; @@ -187,6 +186,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif if (!this.initialized) { this.initialize(); } + channel.setName(name); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setMessagingTaskScheduler(this.taskScheduler); if (dispatcherPolicy != null) { @@ -213,11 +213,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif Assert.notNull(name, "'name' must not be null"); Assert.notNull(handler, "'handler' must not be null"); Assert.notNull(subscription, "'subscription' must not be null"); - DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handler); endpoint.setName(name); - endpoint.setHandler(handler); endpoint.setSubscription(subscription); endpoint.setConcurrencyPolicy(concurrencyPolicy); + endpoint.afterPropertiesSet(); this.registerEndpoint(name, endpoint); } @@ -260,7 +260,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } if (endpoint instanceof DefaultMessageEndpoint) { - String outputChannelName = ((DefaultMessageEndpoint) endpoint).getDefaultOutputChannelName(); + DefaultMessageEndpoint dme = (DefaultMessageEndpoint) endpoint; + String outputChannelName = dme.getDefaultOutputChannelName(); if (outputChannelName != null && this.lookupChannel(outputChannelName) == null) { if (!this.autoCreateChannels) { throw new MessagingConfigurationException("Unknown channel '" + outputChannelName + @@ -269,8 +270,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } this.registerChannel(outputChannelName, new SimpleChannel()); } + if (!dme.hasErrorHandler() && this.getErrorChannel() != null) { + dme.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel())); + } } - this.registerWithDispatcher(channel, endpoint, subscription.getSchedule(), endpoint.getConcurrencyPolicy()); + this.registerWithDispatcher(channel, endpoint, subscription.getSchedule()); if (logger.isInfoEnabled()) { logger.info("activated subscription to channel '" + channel.getName() + "' for endpoint '" + endpoint.getName() + "'"); @@ -295,16 +299,13 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } - private void registerWithDispatcher(MessageChannel channel, MessageHandler handler, Schedule schedule, ConcurrencyPolicy concurrencyPolicy) { + private void registerWithDispatcher(MessageChannel channel, MessageHandler handler, Schedule schedule) { MessageDispatcher dispatcher = dispatchers.get(channel); if (dispatcher == null) { if (logger.isWarnEnabled()) { logger.warn("no dispatcher available for channel '" + channel.getName() + "', be sure to register the channel"); } } - if (concurrencyPolicy != null) { - handler = new PooledMessageHandler(handler, concurrencyPolicy.getCoreSize(), concurrencyPolicy.getMaxSize()); - } dispatcher.addHandler(handler, schedule); if (this.isRunning() && !dispatcher.isRunning()) { dispatcher.start(); @@ -349,23 +350,27 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public void stop() { + if (!this.isRunning()) { + return; + } synchronized (this.lifecycleMonitor) { - if (this.isRunning()) { - this.running = false; - this.taskScheduler.stop(); - for (Lifecycle adapter : this.lifecycleSourceAdapters) { - adapter.stop(); - if (logger.isInfoEnabled()) { - logger.info("stopped source adapter '" + adapter + "'"); - } - } - for (MessageDispatcher dispatcher : this.dispatchers.values()) { - dispatcher.stop(); - if (logger.isInfoEnabled()) { - logger.info("stopped dispatcher '" + dispatcher + "'"); - } + this.running = false; + this.taskScheduler.stop(); + for (Lifecycle adapter : this.lifecycleSourceAdapters) { + adapter.stop(); + if (logger.isInfoEnabled()) { + logger.info("stopped source adapter '" + adapter + "'"); } } + for (MessageDispatcher dispatcher : this.dispatchers.values()) { + dispatcher.stop(); + if (logger.isInfoEnabled()) { + logger.info("stopped dispatcher '" + dispatcher + "'"); + } + } + } + if (logger.isInfoEnabled()) { + logger.info("message bus stopped"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java index 9ba0ad24d4..ab52c2d55d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java @@ -30,6 +30,11 @@ public interface MessageChannel { */ String getName(); + /** + * Set the name of this channel. + */ + void setName(String name); + /** * Return whether this channel has been designated as a publish-subscribe channel. * If so, any dispatcher retrieving messages from this channel should send each diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java index 6d6b399c42..b31507e3f4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java @@ -156,51 +156,53 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas if (!this.scheduler.isRunning()) { this.scheduler.start(); } + if (this.isRunning()) { + return; + } synchronized (this.lifecycleMonitor) { - if (!this.isRunning()) { - for (Map.Entry> entry : this.scheduledHandlers.entrySet()) { - Schedule schedule = entry.getKey(); - List handlers = entry.getValue(); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setMaxMessagesPerTask(this.maxMessagesPerTask); - retriever.setReceiveTimeout(this.receiveTimeout); - DispatcherTask task = new DispatcherTask(retriever); - task.setSchedule(schedule); - task.setRejectionLimit(this.rejectionLimit); - task.setRetryInterval(this.retryInterval); - task.setPublishSubscribe(channel.isPublishSubscribe()); - task.setShouldFailOnRejectionLimit(this.shouldFailOnRejectionLimit); - for (MessageHandler handler : handlers) { - if (handler instanceof Lifecycle) { - ((Lifecycle) handler).start(); - } - task.addHandler(handler); - } - ScheduledFuture future = this.scheduler.schedule(task); - if (future != null) { - futures.add(future); + for (Map.Entry> entry : this.scheduledHandlers.entrySet()) { + Schedule schedule = entry.getKey(); + List handlers = entry.getValue(); + ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); + retriever.setMaxMessagesPerTask(this.maxMessagesPerTask); + retriever.setReceiveTimeout(this.receiveTimeout); + DispatcherTask task = new DispatcherTask(retriever); + task.setSchedule(schedule); + task.setRejectionLimit(this.rejectionLimit); + task.setRetryInterval(this.retryInterval); + task.setPublishSubscribe(channel.isPublishSubscribe()); + task.setShouldFailOnRejectionLimit(this.shouldFailOnRejectionLimit); + for (MessageHandler handler : handlers) { + if (handler instanceof Lifecycle) { + ((Lifecycle) handler).start(); } + task.addHandler(handler); + } + ScheduledFuture future = this.scheduler.schedule(task); + if (future != null) { + futures.add(future); } - this.running = true; } + this.running = true; } } public void stop() { + if (!this.isRunning()) { + return; + } synchronized (this.lifecycleMonitor) { - if (this.isRunning()) { - for (ScheduledFuture future : this.futures) { - future.cancel(true); - for (List handlerList : scheduledHandlers.values()) { - for (MessageHandler handler : handlerList) { - if (handler instanceof Lifecycle) { - ((Lifecycle) handler).stop(); - } + for (ScheduledFuture future : this.futures) { + future.cancel(true); + for (List handlerList : scheduledHandlers.values()) { + for (MessageHandler handler : handlerList) { + if (handler instanceof Lifecycle) { + ((Lifecycle) handler).stop(); } } } - this.running = false; } + this.running = false; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageSelectorRejectedException.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageSelectorRejectedException.java index 55b7cac638..b67fd89132 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageSelectorRejectedException.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageSelectorRejectedException.java @@ -26,6 +26,10 @@ import org.springframework.integration.MessageHandlingException; */ public class MessageSelectorRejectedException extends MessageHandlingException { + public MessageSelectorRejectedException() { + super(); + } + public MessageSelectorRejectedException(String message) { super(message); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java index 8e72d69009..a59323e50c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java @@ -30,6 +30,15 @@ public class ConcurrencyPolicy implements EndpointPolicy { private int maxSize; + public ConcurrencyPolicy() { + } + + public ConcurrencyPolicy(int coreSize, int maxSize) { + this.setCoreSize(coreSize); + this.setMaxSize(maxSize); + } + + public int getCoreSize() { return this.coreSize; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index 60a5262f96..b9bb695ec2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -16,35 +16,67 @@ package org.springframework.integration.endpoint; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.Lifecycle; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessagingConfigurationException; import org.springframework.integration.bus.Subscription; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.dispatcher.MessageHandlerNotRunningException; +import org.springframework.integration.dispatcher.MessageSelectorRejectedException; +import org.springframework.integration.handler.ConcurrentHandler; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; +import org.springframework.integration.message.selector.MessageSelector; +import org.springframework.integration.util.ErrorHandler; +import org.springframework.util.Assert; /** * Default implementation of the {@link MessageEndpoint} interface. * * @author Mark Fisher */ -public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryAware, BeanNameAware { +public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryAware, InitializingBean, BeanNameAware { + + private final Log logger = LogFactory.getLog(this.getClass()); private String name; private MessageHandler handler; + private List selectors = new CopyOnWriteArrayList(); + private Subscription subscription; private ConcurrencyPolicy concurrencyPolicy; + private ErrorHandler errorHandler; + private String defaultOutputChannelName; private ChannelRegistry channelRegistry; + private volatile boolean initialized; + + private volatile boolean running; + + + public DefaultMessageEndpoint() { + } + + public DefaultMessageEndpoint(MessageHandler handler) { + this.handler = handler; + } + public String getName() { return this.name; @@ -58,28 +90,22 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA this.setName(beanName); } - public String getDefaultOutputChannelName() { - return this.defaultOutputChannelName; - } - - /** - * Set the name of the channel to which this endpoint can send reply messages by default. - */ - public void setDefaultOutputChannelName(String defaultOutputChannelName) { - this.defaultOutputChannelName = defaultOutputChannelName; - } - public MessageHandler getHandler() { return this.handler; } /** - * Set a handler to be invoked for each consumed message. + * Set the handler to be invoked for each consumed message. */ public void setHandler(MessageHandler handler) { this.handler = handler; } + public void addMessageSelector(MessageSelector messageSelector) { + Assert.notNull(messageSelector, "'messageSelector' must not be null"); + this.selectors.add(messageSelector); + } + public Subscription getSubscription() { return this.subscription; } @@ -96,6 +122,25 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA this.concurrencyPolicy = concurrencyPolicy; } + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + + public boolean hasErrorHandler() { + return (this.errorHandler != null); + } + + public String getDefaultOutputChannelName() { + return this.defaultOutputChannelName; + } + + /** + * Set the name of the channel to which this endpoint can send reply messages by default. + */ + public void setDefaultOutputChannelName(String defaultOutputChannelName) { + this.defaultOutputChannelName = defaultOutputChannelName; + } + /** * Set the channel registry to use for looking up channels by name. */ @@ -103,7 +148,55 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA this.channelRegistry = channelRegistry; } - public Message handle(Message message) { + public void afterPropertiesSet() { + if (this.concurrencyPolicy != null) { + if (!(this.handler instanceof ConcurrentHandler)) { + this.handler = new ConcurrentHandler(this.handler); + } + ConcurrentHandler concurrentHandler = (ConcurrentHandler) this.handler; + concurrentHandler.setCorePoolSize(this.concurrencyPolicy.getCoreSize()); + concurrentHandler.setMaxPoolSize(this.concurrencyPolicy.getMaxSize()); + if (this.errorHandler != null) { + concurrentHandler.setErrorHandler(this.errorHandler); + } + concurrentHandler.afterPropertiesSet(); + } + this.initialized = true; + } + + public boolean isRunning() { + return this.running; + } + + public void start() { + if (this.isRunning()) { + return; + } + if (this.handler instanceof Lifecycle) { + ((Lifecycle) handler).start(); + } + this.running = true; + } + + public void stop() { + if (!this.isRunning()) { + return; + } + if (this.handler instanceof Lifecycle) { + ((Lifecycle) handler).stop(); + } + this.running = false; + } + + public final Message handle(Message message) { + if (!this.isRunning()) { + throw new MessageHandlerNotRunningException(); + } + for (MessageSelector selector : this.selectors) { + if (!selector.accept(message)) { + throw new MessageSelectorRejectedException(); + } + } if (this.handler == null) { if (this.defaultOutputChannelName == null) { throw new MessagingConfigurationException( @@ -113,15 +206,24 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA replyChannel.send(message); return null; } - Message replyMessage = handler.handle(message); - if (replyMessage != null) { - MessageChannel replyChannel = this.resolveReplyChannel(message); - if (replyChannel == null) { - throw new MessageHandlingException("Unable to determine reply channel for message. " - + "Provide a 'replyChannelName' in the message header or a 'defaultOutputChannelName' " - + "on the message endpoint."); + try { + Message replyMessage = handler.handle(message); + if (replyMessage != null) { + MessageChannel replyChannel = this.resolveReplyChannel(message); + if (replyChannel == null) { + throw new MessageHandlingException("Unable to determine reply channel for message. " + + "Provide a 'replyChannelName' in the message header or a 'defaultOutputChannelName' " + + "on the message endpoint."); + } + replyChannel.send(replyMessage); } - replyChannel.send(replyMessage); + } + catch (Throwable t) { + if (this.errorHandler == null) { + throw new MessageHandlingException( + "error occurred in endpoint, and no 'errorHandler' available", t); + } + this.errorHandler.handle(t); } return null; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index 7627162581..7d19b6cc53 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -16,6 +16,7 @@ package org.springframework.integration.endpoint; +import org.springframework.context.Lifecycle; import org.springframework.integration.bus.Subscription; import org.springframework.integration.handler.MessageHandler; @@ -24,7 +25,7 @@ import org.springframework.integration.handler.MessageHandler; * * @author Mark Fisher */ -public interface MessageEndpoint extends MessageHandler { +public interface MessageEndpoint extends MessageHandler, Lifecycle { String getName(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java new file mode 100644 index 0000000000..952a8fa335 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java @@ -0,0 +1,196 @@ +/* + * Copyright 2002-2007 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.handler; + +import java.util.concurrent.RejectedExecutionException; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.Lifecycle; +import org.springframework.integration.dispatcher.MessageHandlerNotRunningException; +import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; +import org.springframework.integration.message.Message; +import org.springframework.integration.util.ErrorHandler; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.util.Assert; + +/** + * A {@link MessageHandler} implementation that encapsulates a + * {@link ThreadPoolTaskExecutor} and delegates to a wrapped handler for + * concurrent, asynchronous message handling. + * + * @author Mark Fisher + */ +public class ConcurrentHandler implements MessageHandler, Lifecycle, InitializingBean { + + private Log logger = LogFactory.getLog(this.getClass()); + + private MessageHandler handler; + + private ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + + private int corePoolSize = 1; + + private int maxPoolSize = 5; + + private int queueCapacity = 0; + + private int keepAliveSeconds = 60; + + private ErrorHandler errorHandler; + + private volatile boolean running; + + private Object lifecycleMonitor = new Object(); + + + public ConcurrentHandler(MessageHandler handler) { + Assert.notNull(handler, "'handler' must not be null"); + this.handler = handler; + } + + public ConcurrentHandler(MessageHandler handler, int corePoolSize, int maxPoolSize) { + Assert.notNull(handler, "'handler' must not be null"); + Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1"); + Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1"); + Assert.isTrue(maxPoolSize >= corePoolSize, "'corePoolSize' cannot exceed 'maxPoolSize'"); + this.handler = handler; + this.corePoolSize = corePoolSize; + this.maxPoolSize = maxPoolSize; + } + + + public void setExecutor(ThreadPoolTaskExecutor executor) { + Assert.notNull(executor, "'executor' must not be null"); + this.executor = executor; + } + + public void setCorePoolSize(int corePoolSize) { + Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1"); + this.corePoolSize = corePoolSize; + if (this.executor != null) { + this.executor.setCorePoolSize(corePoolSize); + } + } + + public void setMaxPoolSize(int maxPoolSize) { + Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1"); + this.maxPoolSize = maxPoolSize; + if (this.executor != null) { + this.executor.setMaxPoolSize(maxPoolSize); + } + } + + public void setQueueCapacity(int queueCapacity) { + this.queueCapacity = queueCapacity; + if (this.executor != null) { + this.executor.setQueueCapacity(queueCapacity); + } + } + + public void setKeepAliveSeconds(int keepAliveSeconds) { + this.keepAliveSeconds = keepAliveSeconds; + if (this.executor != null) { + this.executor.setKeepAliveSeconds(keepAliveSeconds); + } + } + + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + + public void afterPropertiesSet() { + initializeExecutor(); + } + + private void initializeExecutor() { + if (this.executor == null) { + this.executor = new ThreadPoolTaskExecutor(); + } + this.executor.setCorePoolSize(this.corePoolSize); + this.executor.setMaxPoolSize(this.maxPoolSize); + this.executor.setQueueCapacity(this.queueCapacity); + this.executor.setKeepAliveSeconds(this.keepAliveSeconds); + CustomizableThreadFactory threadFactory = new CustomizableThreadFactory(); + threadFactory.setThreadNamePrefix("handler-"); + this.executor.setThreadFactory(threadFactory); + this.executor.afterPropertiesSet(); + } + + public boolean isRunning() { + return this.running; + } + + public void start() { + synchronized (this.lifecycleMonitor) { + if (!this.running) { + this.afterPropertiesSet(); + } + this.running = true; + } + } + + public void stop() { + synchronized (this.lifecycleMonitor) { + if (this.isRunning()) { + this.executor.shutdown(); + } + this.running = false; + } + } + + public Message handle(Message message) { + if (!this.isRunning()) { + throw new MessageHandlerNotRunningException(); + } + try { + this.executor.execute(new HandlerTask(message)); + return null; + } + catch (RejectedExecutionException e) { + throw new MessageHandlerRejectedExecutionException(e); + } + } + + + private class HandlerTask implements Runnable { + + private Message message; + + HandlerTask(Message message) { + this.message = message; + } + + public void run() { + try { + handler.handle(this.message); + } + catch (Throwable t) { + if (errorHandler != null) { + errorHandler.handle(t); + } + else if (logger.isWarnEnabled()) { + logger.warn("error occurred in handler execution", t); + } + } + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/InterceptingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/InterceptingMessageHandler.java index 5b3e409c15..ca57f1aabc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/InterceptingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/InterceptingMessageHandler.java @@ -34,7 +34,7 @@ public abstract class InterceptingMessageHandler implements MessageHandler { this.target = target; } - public Message handle(Message message) { + public final Message handle(Message message) { return handle(message, this.target); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/PooledMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/PooledMessageHandler.java deleted file mode 100644 index 249517cc4f..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/PooledMessageHandler.java +++ /dev/null @@ -1,225 +0,0 @@ -/* - * Copyright 2002-2007 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.handler; - -import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.context.Lifecycle; -import org.springframework.integration.dispatcher.MessageHandlerNotRunningException; -import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; -import org.springframework.integration.dispatcher.MessageSelectorRejectedException; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.selector.MessageSelector; -import org.springframework.scheduling.concurrent.CustomizableThreadFactory; -import org.springframework.util.Assert; - -/** - * Encapsulates a {@link ThreadPoolExecutor} with configurable error thresholds. - * - * @author Mark Fisher - */ -public class PooledMessageHandler implements MessageHandler, Lifecycle { - - private Log logger = LogFactory.getLog(this.getClass()); - - private MessageHandler handler; - - private List selectors = new CopyOnWriteArrayList(); - - private ThreadPoolExecutor threadPoolExecutor; - - private int corePoolSize; - - private int maxPoolSize; - - private volatile boolean running; - - private Object lifecycleMonitor = new Object(); - - private int successiveErrorCount; - - private int successiveErrorThreshold = -1; - - private int totalErrorCount; - - private int totalErrorThreshold = -1; - - - public PooledMessageHandler(MessageHandler handler, int corePoolSize, int maxPoolSize) { - Assert.notNull(handler, "'handler' must not be null"); - Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1"); - Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1"); - Assert.isTrue(maxPoolSize >= corePoolSize, "'corePoolSize' cannot exceed 'maxPoolSize'"); - this.handler = handler; - this.corePoolSize = corePoolSize; - this.maxPoolSize = maxPoolSize; - } - - public void setCorePoolSize(int corePoolSize) { - this.corePoolSize = corePoolSize; - } - - public void setMaxPoolSize(int maxPoolSize) { - this.maxPoolSize = maxPoolSize; - } - - public void addMessageSelector(MessageSelector messageSelector) { - Assert.notNull(messageSelector, "'messageSelector' must not be null"); - this.selectors.add(messageSelector); - } - - public boolean isRunning() { - return this.running; - } - - public void start() { - synchronized (this.lifecycleMonitor) { - if (!this.running) { - this.threadPoolExecutor = new MessageHandlerThreadPoolExecutor(this.corePoolSize, this.maxPoolSize); - } - this.running = true; - } - } - - public void stop() { - synchronized (this.lifecycleMonitor) { - if (this.isRunning()) { - this.threadPoolExecutor.shutdown(); - this.threadPoolExecutor = null; - } - this.running = false; - } - } - - public Message handle(Message message) { - if (!this.isRunning()) { - throw new MessageHandlerNotRunningException(); - } - for (MessageSelector selector : this.selectors) { - if (!selector.accept(message)) { - throw new MessageSelectorRejectedException("selector rejected message"); - } - } - try { - this.threadPoolExecutor.execute(new HandlerTask(this.handler, message)); - return null; - } - catch (RejectedExecutionException e) { - throw new MessageHandlerRejectedExecutionException( - "handler executor rejected message", e); - } - } - - /** - * Set the maximum number of errors allowed in successive - * executions. If this threshold is ever exceeded, the executor - * will shutdown. - */ - public void setSuccessiveErrorThreshold(int successiveErrorThreshold) { - this.successiveErrorThreshold = successiveErrorThreshold; - } - - /** - * Set the maximum number of total errors allowed in executions - * If this threshold is ever exceeded, the executor will shutdown. - */ - public void setTotalErrorThreshold(int totalErrorThreshold) { - this.totalErrorThreshold = totalErrorThreshold; - } - - public int getActiveCount() { - if (this.threadPoolExecutor == null) { - return 0; - } - return this.threadPoolExecutor.getActiveCount(); - } - - public boolean isShutdown() { - return this.threadPoolExecutor.isShutdown(); - } - - - private static class HandlerTask implements Runnable { - - private MessageHandler handler; - - private Message message; - - private Throwable error; - - - HandlerTask(MessageHandler handler, Message message) { - this.handler = handler; - this.message = message; - } - - public Throwable getError() { - return this.error; - } - - public void run() { - try { - this.handler.handle(this.message); - } - catch (Throwable t) { - this.error = t; - } - } - } - - - private class MessageHandlerThreadPoolExecutor extends ThreadPoolExecutor { - - public MessageHandlerThreadPoolExecutor(int corePoolSize, int maximumPoolSize) { - super(corePoolSize, maximumPoolSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue()); - CustomizableThreadFactory threadFactory = new CustomizableThreadFactory(); - threadFactory.setThreadNamePrefix("handler-pool-"); - this.setThreadFactory(threadFactory); - } - - @Override - protected void afterExecute(Runnable r, Throwable t) { - HandlerTask task = (HandlerTask) r; - if (task.getError() != null) { - if (logger.isWarnEnabled()) { - logger.warn("Exception occurred during task execution", task.getError()); - } - successiveErrorCount++; - totalErrorCount++; - if ((successiveErrorThreshold >= 0 && successiveErrorCount > successiveErrorThreshold) - || (totalErrorThreshold >= 0 && totalErrorCount > totalErrorThreshold)) { - if (logger.isInfoEnabled()) { - logger.info("error threshold exceeded, shutting down now"); - } - this.shutdownNow(); - } - } - else { - successiveErrorCount = 0; - } - } - } - -} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java index 5fd7d0eadd..06555c7f79 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java @@ -30,8 +30,12 @@ import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.dispatcher.DefaultMessageDispatcher; import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; +import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; +import org.springframework.integration.handler.ConcurrentHandler; +import org.springframework.integration.handler.InterceptingMessageHandler; import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.handler.PooledMessageHandler; +import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -49,20 +53,20 @@ public class DefaultMessageDispatcherTests { final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(1); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); - dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1)); SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); scheduler.start(); dispatcher.setMessagingTaskScheduler(scheduler); dispatcher.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("exactly one endpoint should have received message", 1, counter1.get() + counter2.get()); + assertEquals("exactly one handler should have received message", 1, counter1.get() + counter2.get()); } @Test @@ -70,18 +74,18 @@ public class DefaultMessageDispatcherTests { final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(2); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); - dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1)); dispatcher.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("both endpoints should have received message", 2, counter1.get() + counter2.get()); + assertEquals("both handlers should have received message", 2, counter1.get() + counter2.get()); } @Test @@ -90,24 +94,24 @@ public class DefaultMessageDispatcherTests { final AtomicInteger counter2 = new AtomicInteger(); final AtomicInteger counter3 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(1); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); + MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { @Override public void start() { } }); - dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1)); - dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler3, 1, 1)); dispatcher.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("inactive endpoint should not have received message", 0, counter1.get()); - assertEquals("exactly one endpoint should have received message", 1, counter2.get() + counter3.get()); + assertEquals("inactive handler should not have received message", 0, counter1.get()); + assertEquals("exactly one handler should have received message", 1, counter2.get() + counter3.get()); } @Test @@ -116,25 +120,25 @@ public class DefaultMessageDispatcherTests { final AtomicInteger counter2 = new AtomicInteger(); final AtomicInteger counter3 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(2); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); + MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); SimpleChannel channel = new SimpleChannel(); channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); - dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { @Override public void start() { } }); - dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler3, 1, 1)); dispatcher.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("inactive endpoint should not have received message", 0, counter2.get()); - assertEquals("both active endpoints should have received message", 2, counter1.get() + counter3.get()); + assertEquals("inactive handler should not have received message", 0, counter2.get()); + assertEquals("both active handlers should have received message", 2, counter1.get() + counter3.get()); } @Test @@ -151,23 +155,23 @@ public class DefaultMessageDispatcherTests { final AtomicInteger counter2 = new AtomicInteger(); final AtomicInteger counter3 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(2); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); + MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); SimpleChannel channel = new SimpleChannel(); channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); dispatcher.setRetryInterval(3); - dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); - dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { @Override public Message handle(Message message) { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler3, 1, 1)); SimpleChannel errorChannel = new SimpleChannel(); SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); @@ -187,9 +191,9 @@ public class DefaultMessageDispatcherTests { final AtomicInteger counter2 = new AtomicInteger(); final AtomicInteger counter3 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(2); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); + MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); SimpleChannel channel = new SimpleChannel(); channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); @@ -197,19 +201,19 @@ public class DefaultMessageDispatcherTests { dispatcher.setRejectionLimit(2); dispatcher.setRetryInterval(3); dispatcher.setShouldFailOnRejectionLimit(false); - dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); - dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { @Override public Message handle(Message message) { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler3, 1, 1)); dispatcher.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("rejecting endpoint should not have received message", 0, counter2.get()); - assertEquals("both non-rejecting endpoints should have received message", 2, counter1.get() + counter3.get()); + assertEquals("rejecting handler should not have received message", 0, counter2.get()); + assertEquals("both non-rejecting handlers should have received message", 2, counter1.get() + counter3.get()); } @Test @@ -217,21 +221,21 @@ public class DefaultMessageDispatcherTests { final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(4); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); dispatcher.setRetryInterval(3); - dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { @Override public Message handle(Message message) { latch.countDown(); throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { @Override public Message handle(Message message) { latch.countDown(); @@ -258,15 +262,15 @@ public class DefaultMessageDispatcherTests { final AtomicInteger rejectedCounter1 = new AtomicInteger(); final AtomicInteger rejectedCounter2 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(4); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); dispatcher.setRetryInterval(3); dispatcher.setShouldFailOnRejectionLimit(false); - dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { @Override public Message handle(Message message) { rejectedCounter1.incrementAndGet(); @@ -274,7 +278,7 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { @Override public Message handle(Message message) { rejectedCounter2.incrementAndGet(); @@ -286,9 +290,9 @@ public class DefaultMessageDispatcherTests { latch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("latch should have counted down within allotted time", 0, latch.getCount()); - assertEquals("rejecting endpoints should not have received message", 0, counter1.get() + counter2.get()); - assertEquals("endpoint1 should have rejected two times", 2, rejectedCounter1.get()); - assertEquals("endpoint2 should have rejected two times", 2, rejectedCounter2.get()); + assertEquals("rejecting handlers should not have received message", 0, counter1.get() + counter2.get()); + assertEquals("handler1 should have rejected two times", 2, rejectedCounter1.get()); + assertEquals("handler2 should have rejected two times", 2, rejectedCounter2.get()); } @Test @@ -300,16 +304,16 @@ public class DefaultMessageDispatcherTests { final AtomicInteger rejectedCounter2 = new AtomicInteger(); final AtomicInteger rejectedCounter3 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(5); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); + MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); dispatcher.setRetryInterval(3); dispatcher.setShouldFailOnRejectionLimit(false); - dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { @Override public Message handle(Message message) { rejectedCounter1.incrementAndGet(); @@ -317,7 +321,7 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { @Override public Message handle(Message message) { if (rejectedCounter2.get() == 1) { @@ -328,7 +332,7 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler3, 1, 1) { @Override public Message handle(Message message) { rejectedCounter3.incrementAndGet(); @@ -339,12 +343,12 @@ public class DefaultMessageDispatcherTests { dispatcher.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("endpoint1 should not have received message", 0, counter1.get()); - assertEquals("endpoint2 should have received message the second time", 1, counter2.get()); - assertEquals("endpoint3 should not have received message", 0, counter3.get()); - assertEquals("endpoint1 should have rejected two times", 2, rejectedCounter1.get()); - assertEquals("endpoint2 should have rejected one time", 1, rejectedCounter2.get()); - assertEquals("endpoint3 should have rejected one time", 1, rejectedCounter3.get()); + assertEquals("handler1 should not have received message", 0, counter1.get()); + assertEquals("handler2 should have received message the second time", 1, counter2.get()); + assertEquals("handler3 should not have received message", 0, counter3.get()); + assertEquals("handler1 should have rejected two times", 2, rejectedCounter1.get()); + assertEquals("handler2 should have rejected one time", 1, rejectedCounter2.get()); + assertEquals("handler3 should have rejected one time", 1, rejectedCounter3.get()); } @Test @@ -354,8 +358,8 @@ public class DefaultMessageDispatcherTests { final AtomicInteger rejectedCounter1 = new AtomicInteger(); final AtomicInteger rejectedCounter2 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(8); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); @@ -363,7 +367,7 @@ public class DefaultMessageDispatcherTests { dispatcher.setRejectionLimit(5); dispatcher.setRetryInterval(3); dispatcher.setShouldFailOnRejectionLimit(false); - dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { @Override public Message handle(Message message) { if (rejectedCounter1.get() == 2) { @@ -374,7 +378,7 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { @Override public Message handle(Message message) { if (rejectedCounter2.get() == 4) { @@ -388,10 +392,10 @@ public class DefaultMessageDispatcherTests { dispatcher.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("endpoint1 should have received one message", 1, counter1.get()); - assertEquals("endpoint2 should have received one message", 1, counter2.get()); - assertEquals("endpoint1 should have rejected two times", 2, rejectedCounter1.get()); - assertEquals("endpoint2 should have rejected four times", 4, rejectedCounter2.get()); + assertEquals("handler1 should have received one message", 1, counter1.get()); + assertEquals("handler2 should have received one message", 1, counter2.get()); + assertEquals("handler1 should have rejected two times", 2, rejectedCounter1.get()); + assertEquals("handler2 should have rejected four times", 4, rejectedCounter2.get()); } @Test @@ -399,22 +403,22 @@ public class DefaultMessageDispatcherTests { final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(1); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1); - PooledMessageHandler executor2 = new PooledMessageHandler(endpoint2, 1, 1); - executor1.addMessageSelector(new PayloadTypeSelector(Integer.class)); - executor2.addMessageSelector(new PayloadTypeSelector(String.class)); - dispatcher.addHandler(executor1); - dispatcher.addHandler(executor2); + DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(handler1); + DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint(handler2); + endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class)); + endpoint2.addMessageSelector(new PayloadTypeSelector(String.class)); + dispatcher.addHandler(endpoint1); + dispatcher.addHandler(endpoint2); dispatcher.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); - assertEquals("endpoint2 should have accepted the message", 1, counter2.get()); + assertEquals("handler1 should not have accepted the message", 0, counter1.get()); + assertEquals("handler2 should have accepted the message", 1, counter2.get()); } @Test @@ -424,40 +428,42 @@ public class DefaultMessageDispatcherTests { final AtomicInteger attemptedCounter1 = new AtomicInteger(); final AtomicInteger attemptedCounter2 = new AtomicInteger(); final CountDownLatch attemptedLatch = new CountDownLatch(2); - final CountDownLatch endpointLatch = new CountDownLatch(1); - TestEndpoint endpoint1 = new TestEndpoint(counter1, endpointLatch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, endpointLatch); + final CountDownLatch handlerLatch = new CountDownLatch(1); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, attemptedLatch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, attemptedLatch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1) { + DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(handler1); + DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint(handler2); + endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class)); + endpoint2.addMessageSelector(new PayloadTypeSelector(Integer.class)); + MessageHandler interceptor1 = new InterceptingMessageHandler(endpoint1) { @Override - public Message handle(Message message) { + public Message handle(Message message, MessageHandler target) { attemptedCounter1.incrementAndGet(); attemptedLatch.countDown(); - return super.handle(message); + return target.handle(message); } }; - PooledMessageHandler executor2 = new PooledMessageHandler(endpoint2, 1, 1) { + MessageHandler interceptor2 = new InterceptingMessageHandler(endpoint2) { @Override - public Message handle(Message message) { + public Message handle(Message message, MessageHandler target) { attemptedCounter2.incrementAndGet(); attemptedLatch.countDown(); - return super.handle(message); + return target.handle(message); } }; - executor1.addMessageSelector(new PayloadTypeSelector(Integer.class)); - executor2.addMessageSelector(new PayloadTypeSelector(Integer.class)); - dispatcher.addHandler(executor1); - dispatcher.addHandler(executor2); + dispatcher.addHandler(interceptor1); + dispatcher.addHandler(interceptor2); dispatcher.start(); attemptedLatch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, attemptedLatch.getCount()); - assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); - assertEquals("endpoint2 should not have accepted the message", 0, counter2.get()); + assertEquals("handler1 should not have accepted the message", 0, counter1.get()); + assertEquals("handler2 should not have accepted the message", 0, counter2.get()); assertEquals("executor1 should have had exactly one attempt", 1, attemptedCounter1.get()); assertEquals("executor2 should have had exactly one attempt", 1, attemptedCounter2.get()); - assertEquals("endpointLatch should not have counted down", 1, endpointLatch.getCount()); + assertEquals("handlerLatch should not have counted down", 1, handlerLatch.getCount()); assertEquals("attemptedLatch should have counted down", 0, attemptedLatch.getCount()); } @@ -466,18 +472,22 @@ public class DefaultMessageDispatcherTests { final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(1); - TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); - TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); + MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); + MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1); - PooledMessageHandler executor2 = new PooledMessageHandler(endpoint2, 1, 1); - executor1.addMessageSelector(new PayloadTypeSelector(Integer.class)); - executor2.addMessageSelector(new PayloadTypeSelector(String.class)); - dispatcher.addHandler(executor1); - dispatcher.addHandler(executor2); + DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(); + endpoint1.setHandler(handler1); + endpoint1.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); + DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint(); + endpoint2.setHandler(handler2); + endpoint2.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); + endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class)); + endpoint2.addMessageSelector(new PayloadTypeSelector(String.class)); + dispatcher.addHandler(endpoint1); + dispatcher.addHandler(endpoint2); dispatcher.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); @@ -485,23 +495,4 @@ public class DefaultMessageDispatcherTests { assertEquals("endpoint2 should have accepted the message", 1, counter2.get()); } - - private static class TestEndpoint implements MessageHandler { - - private AtomicInteger counter; - - private CountDownLatch latch; - - public TestEndpoint(AtomicInteger counter, CountDownLatch latch) { - this.counter = counter; - this.latch = latch; - } - - public Message handle(Message message) { - counter.incrementAndGet(); - latch.countDown(); - return null; - } - } - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java index 987ecd6993..71170c8539 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java @@ -48,7 +48,9 @@ public class DefaultMessageEndpointTests { endpoint.setChannelRegistry(channelRegistry); endpoint.setHandler(handler); endpoint.setDefaultOutputChannelName("replyChannel"); + endpoint.start(); endpoint.handle(new StringMessage(1, "test")); + endpoint.stop(); Message reply = replyChannel.receive(50); assertNotNull(reply); assertEquals("hello test", reply.getPayload()); @@ -67,9 +69,11 @@ public class DefaultMessageEndpointTests { DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); endpoint.setChannelRegistry(channelRegistry); endpoint.setHandler(handler); + endpoint.start(); StringMessage testMessage = new StringMessage(1, "test"); testMessage.getHeader().setReplyChannelName("replyChannel"); endpoint.handle(testMessage); + endpoint.stop(); Message reply = replyChannel.receive(50); assertNotNull(reply); assertEquals("hello test", reply.getPayload()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java index b564ca224e..718a93dd03 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java @@ -17,20 +17,22 @@ package org.springframework.integration.handler; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; import org.springframework.integration.message.Message; /** - * Factory for {@link MessageHandler} implementations that are useful for testing. + * Factory for {@link MessageHandler} implementations that are useful for + * testing. * * @author Mark Fisher */ -public class TestHandlers { +public abstract class TestHandlers { /** * Create a {@link MessageHandler} that always returns null. */ - public static MessageHandler nullHandler() { + public final static MessageHandler nullHandler() { return new MessageHandler() { public Message handle(Message message) { return null; @@ -38,10 +40,22 @@ public class TestHandlers { }; } + /** + * Create a {@link MessageHandler} that increments the provided counter. + */ + public final static MessageHandler countingHandler(final AtomicInteger counter) { + return new MessageHandler() { + public Message handle(Message message) { + counter.incrementAndGet(); + return null; + } + }; + } + /** * Create a {@link MessageHandler} that counts down on the provided latch. */ - public static MessageHandler countDownHandler(final CountDownLatch latch) { + public final static MessageHandler countDownHandler(final CountDownLatch latch) { return new MessageHandler() { public Message handle(Message message) { latch.countDown(); @@ -50,4 +64,18 @@ public class TestHandlers { }; } + /** + * Create a {@link MessageHandler} that counts down on the provided latch + * and also increments the provided counter. + */ + public final static MessageHandler countingCountDownHandler(final AtomicInteger counter, final CountDownLatch latch) { + return new MessageHandler() { + public Message handle(Message message) { + counter.incrementAndGet(); + latch.countDown(); + return null; + } + }; + } + }