From 35b098b16468847e4ccf8ce3f3e5bf0c338d1227 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sun, 20 Jan 2008 20:03:54 +0000 Subject: [PATCH] Added 'queueCapacity' and 'keepAliveSeconds' properties to ConcurrencyPolicy. Refactored DefaultMessageEndpoint and ConcurrentHandler to delegate to ConcurrencyPolicy and avoid duplication of concurrency properties. Added namespace support for 'dispatcher-policy' as a sub-element of 'channel', and also added namespace support for the new concurrency properties. --- .../integration/config/ChannelParser.java | 42 ++++++- .../integration/config/EndpointParser.java | 14 ++- .../config/spring-integration-1.0.xsd | 20 ++++ .../endpoint/ConcurrencyPolicy.java | 36 +++++- .../endpoint/DefaultMessageEndpoint.java | 4 +- .../handler/ConcurrentHandler.java | 103 +++++++++--------- .../DefaultMessageDispatcherTests.java | 40 +++---- 7 files changed, 176 insertions(+), 83 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java index b479f7f877..7c84487bcc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java @@ -17,6 +17,8 @@ package org.springframework.integration.config; import org.w3c.dom.Element; +import org.w3c.dom.Node; +import org.w3c.dom.NodeList; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.parsing.BeanComponentDefinition; @@ -40,6 +42,8 @@ public class ChannelParser implements BeanDefinitionParser { private static final String PUBLISH_SUBSCRIBE_ATTRIBUTE = "publish-subscribe"; + private static final String DISPATCHER_POLICY_ELEMENT = "dispatcher-policy"; + public BeanDefinition parse(Element element, ParserContext parserContext) { RootBeanDefinition channelDef = new RootBeanDefinition(SimpleChannel.class); @@ -48,10 +52,19 @@ public class ChannelParser implements BeanDefinitionParser { if (StringUtils.hasText(capacity)) { channelDef.getConstructorArgumentValues().addGenericArgumentValue(Integer.parseInt(capacity)); } - String isPublishSubscribe = element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE); - if ("true".equals(isPublishSubscribe)) { - channelDef.getConstructorArgumentValues().addGenericArgumentValue(new DispatcherPolicy(true)); + boolean isPublishSubscribe = "true".equals(element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE)); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(isPublishSubscribe); + NodeList childNodes = element.getChildNodes(); + for (int i = 0; i < childNodes.getLength(); i++) { + Node child = childNodes.item(i); + if (child.getNodeType() == Node.ELEMENT_NODE) { + String localName = child.getLocalName(); + if (DISPATCHER_POLICY_ELEMENT.equals(localName)) { + configureDispatcherPolicy((Element) child, dispatcherPolicy); + } + } } + channelDef.getConstructorArgumentValues().addGenericArgumentValue(dispatcherPolicy); String beanName = element.getAttribute(ID_ATTRIBUTE); if (!StringUtils.hasText(beanName)) { beanName = parserContext.getReaderContext().generateBeanName(channelDef); @@ -60,4 +73,27 @@ public class ChannelParser implements BeanDefinitionParser { return channelDef; } + private void configureDispatcherPolicy(Element element, DispatcherPolicy dispatcherPolicy) { + String maxMessagesPerTask = element.getAttribute("max-messages-per-task"); + if (StringUtils.hasText(maxMessagesPerTask)) { + dispatcherPolicy.setMaxMessagesPerTask(Integer.parseInt(maxMessagesPerTask)); + } + String receiveTimeout = element.getAttribute("receive-timeout"); + if (StringUtils.hasText(receiveTimeout)) { + dispatcherPolicy.setReceiveTimeout(Long.parseLong(receiveTimeout)); + } + String rejectionLimit = element.getAttribute("rejection-limit"); + if (StringUtils.hasText(rejectionLimit)) { + dispatcherPolicy.setRejectionLimit(Integer.parseInt(rejectionLimit)); + } + String retryInterval = element.getAttribute("retry-interval"); + if (StringUtils.hasText(retryInterval)) { + dispatcherPolicy.setRetryInterval(Long.parseLong(retryInterval)); + } + String shouldFailOnRejectionLimit = element.getAttribute("should-fail-on-rejection-limit"); + if (StringUtils.hasText(shouldFailOnRejectionLimit)) { + dispatcherPolicy.setShouldFailOnRejectionLimit("true".equals(shouldFailOnRejectionLimit)); + } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java index a04ad4bc49..c37d62487f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java @@ -88,6 +88,10 @@ public class EndpointParser implements BeanDefinitionParser { private static final String MAX_SIZE_ATTRIBUTE = "max"; + private static final String QUEUE_CAPACITY_ATTRIBUTE = "queue-capacity"; + + private static final String KEEP_ALIVE_ATTRIBUTE = "keep-alive"; + private static final String CONCURRENCY_POLICY_PROPERTY = "concurrencyPolicy"; @@ -166,15 +170,23 @@ public class EndpointParser implements BeanDefinitionParser { } private void parseConcurrencyPolicy(Element concurrencyElement, RootBeanDefinition endpointDef) { - ConcurrencyPolicy policy = new ConcurrencyPolicy(); String coreSize = concurrencyElement.getAttribute(CORE_SIZE_ATTRIBUTE); String maxSize = concurrencyElement.getAttribute(MAX_SIZE_ATTRIBUTE); + String queueCapacity = concurrencyElement.getAttribute(QUEUE_CAPACITY_ATTRIBUTE); + String keepAlive = concurrencyElement.getAttribute(KEEP_ALIVE_ATTRIBUTE); + ConcurrencyPolicy policy = new ConcurrencyPolicy(); if (StringUtils.hasText(coreSize)) { policy.setCoreSize(Integer.parseInt(coreSize)); } if (StringUtils.hasText(maxSize)) { policy.setMaxSize(Integer.parseInt(maxSize)); } + if (StringUtils.hasText(queueCapacity)) { + policy.setQueueCapacity(Integer.parseInt(queueCapacity)); + } + if (StringUtils.hasText(keepAlive)) { + policy.setKeepAliveSeconds(Integer.parseInt(keepAlive)); + } endpointDef.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd index 73996e8a65..e6979063b1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd @@ -49,6 +49,9 @@ + + + @@ -134,6 +137,8 @@ + + @@ -149,6 +154,21 @@ + + + + + Defines a dispatcher policy. + + + + + + + + + + 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 a59323e50c..e4deeed585 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 @@ -25,15 +25,29 @@ import org.springframework.util.Assert; */ public class ConcurrencyPolicy implements EndpointPolicy { - private int coreSize; + public static final int DEFAULT_CORE_SIZE = 1; - private int maxSize; + public static final int DEFAULT_MAX_SIZE = 10; + + public static final int DEFAULT_QUEUE_CAPACITY = 0; + + public static final int DEFAULT_KEEP_ALIVE_SECONDS = 60; + + + private int coreSize = DEFAULT_CORE_SIZE; + + private int maxSize = DEFAULT_MAX_SIZE; + + private int queueCapacity = DEFAULT_QUEUE_CAPACITY; + + private int keepAliveSeconds = DEFAULT_KEEP_ALIVE_SECONDS; public ConcurrencyPolicy() { } public ConcurrencyPolicy(int coreSize, int maxSize) { + Assert.isTrue(maxSize >= coreSize, "'coreSize' must not exceed 'maxSize'"); this.setCoreSize(coreSize); this.setMaxSize(maxSize); } @@ -57,4 +71,22 @@ public class ConcurrencyPolicy implements EndpointPolicy { this.maxSize = maxSize; } + public int getQueueCapacity() { + return this.queueCapacity; + } + + public void setQueueCapacity(int queueCapacity) { + Assert.isTrue(queueCapacity >= 0, "'queueCapacity' must not be negative"); + this.queueCapacity = queueCapacity; + } + + public int getKeepAliveSeconds() { + return this.keepAliveSeconds; + } + + public void setKeepAliveSeconds(int keepAliveSeconds) { + Assert.isTrue(keepAliveSeconds >= 0, "'keepAliveSeconds' must not be negative"); + this.keepAliveSeconds = keepAliveSeconds; + } + } 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 a59772beec..b8a0e05c2d 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 @@ -147,11 +147,9 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA public void afterPropertiesSet() { if (this.concurrencyPolicy != null) { if (!(this.handler instanceof ConcurrentHandler)) { - this.handler = new ConcurrentHandler(this.handler); + this.handler = new ConcurrentHandler(this.handler, this.concurrencyPolicy); } ConcurrentHandler concurrentHandler = (ConcurrentHandler) this.handler; - concurrentHandler.setCorePoolSize(this.concurrencyPolicy.getCoreSize()); - concurrentHandler.setMaxPoolSize(this.concurrencyPolicy.getMaxSize()); if (this.errorHandler != null) { concurrentHandler.setErrorHandler(this.errorHandler); } 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 index 223d9102fe..3e41d021c6 100644 --- 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 @@ -23,6 +23,7 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.context.Lifecycle; import org.springframework.integration.dispatcher.MessageHandlerNotRunningException; import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; +import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.message.Message; import org.springframework.integration.util.ErrorHandler; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; @@ -42,15 +43,11 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin private MessageHandler handler; - private ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + private ThreadPoolTaskExecutor executor; - private int corePoolSize = 1; + private volatile int currentQueueCapacity; - private int maxPoolSize = 5; - - private int queueCapacity = 0; - - private int keepAliveSeconds = 60; + private final ConcurrencyPolicy concurrencyPolicy; private ErrorHandler errorHandler; @@ -60,18 +57,20 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin public ConcurrentHandler(MessageHandler handler) { - Assert.notNull(handler, "'handler' must not be null"); - this.handler = handler; + this(handler, null); } - public ConcurrentHandler(MessageHandler handler, int corePoolSize, int maxPoolSize) { + public ConcurrentHandler(MessageHandler handler, ConcurrencyPolicy concurrencyPolicy) { Assert.notNull(handler, "'handler' must not be null"); - Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1"); - Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1"); - Assert.isTrue(maxPoolSize >= corePoolSize, "'corePoolSize' cannot exceed 'maxPoolSize'"); + if (concurrencyPolicy != null) { + Assert.isTrue(concurrencyPolicy.getMaxSize() >= concurrencyPolicy.getCoreSize(), + "'coreSize' must not exceed 'maxSize'"); + this.concurrencyPolicy = concurrencyPolicy; + } + else { + this.concurrencyPolicy = new ConcurrencyPolicy(); + } this.handler = handler; - this.corePoolSize = corePoolSize; - this.maxPoolSize = maxPoolSize; } @@ -80,58 +79,54 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin this.executor = executor; } - public void setCorePoolSize(int corePoolSize) { - Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1"); - this.corePoolSize = corePoolSize; - if (this.executor != null) { - this.executor.setCorePoolSize(corePoolSize); - } - } - - public void setMaxPoolSize(int maxPoolSize) { - Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1"); - this.maxPoolSize = maxPoolSize; - if (this.executor != null) { - this.executor.setMaxPoolSize(maxPoolSize); - } - } - - public void setQueueCapacity(int queueCapacity) { - this.queueCapacity = queueCapacity; - if (this.executor != null) { - this.executor.setQueueCapacity(queueCapacity); - } - } - - public void setKeepAliveSeconds(int keepAliveSeconds) { - this.keepAliveSeconds = keepAliveSeconds; - if (this.executor != null) { - this.executor.setKeepAliveSeconds(keepAliveSeconds); - } - } - public void setErrorHandler(ErrorHandler errorHandler) { this.errorHandler = errorHandler; } public void afterPropertiesSet() { - initializeExecutor(); + refreshExecutor(); + } + + public void refreshExecutor() { + if (this.executor == null || currentQueueCapacity != this.concurrencyPolicy.getQueueCapacity()) { + this.initializeExecutor(); + } + else { + this.refreshRuntimeModifiableExecutorProperties(); + } } private void initializeExecutor() { - if (this.executor == null) { + synchronized (this.lifecycleMonitor) { + if (this.executor != null) { + this.executor.shutdown(); + } this.executor = new ThreadPoolTaskExecutor(); + this.currentQueueCapacity = this.concurrencyPolicy.getQueueCapacity(); + this.executor.setQueueCapacity(this.currentQueueCapacity); + CustomizableThreadFactory threadFactory = new CustomizableThreadFactory(); + threadFactory.setThreadNamePrefix("handler-"); + this.executor.setThreadFactory(threadFactory); + this.refreshRuntimeModifiableExecutorProperties(); } - this.executor.setCorePoolSize(this.corePoolSize); - this.executor.setMaxPoolSize(this.maxPoolSize); - this.executor.setQueueCapacity(this.queueCapacity); - this.executor.setKeepAliveSeconds(this.keepAliveSeconds); - CustomizableThreadFactory threadFactory = new CustomizableThreadFactory(); - threadFactory.setThreadNamePrefix("handler-"); - this.executor.setThreadFactory(threadFactory); this.executor.afterPropertiesSet(); } + private void refreshRuntimeModifiableExecutorProperties() { + int coreSize = this.concurrencyPolicy.getCoreSize(); + int maxSize = this.concurrencyPolicy.getMaxSize(); + int keepAlive = this.concurrencyPolicy.getKeepAliveSeconds(); + if (this.executor.getCorePoolSize() != coreSize) { + this.executor.setCorePoolSize(coreSize); + } + if (this.executor.getMaxPoolSize() != maxSize) { + this.executor.setMaxPoolSize(maxSize); + } + if (this.executor.getKeepAliveSeconds() != keepAlive) { + this.executor.setKeepAliveSeconds(keepAlive); + } + } + public boolean isRunning() { return this.running; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java index 1cd39b1846..f0b6872449 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java @@ -58,8 +58,8 @@ public class DefaultMessageDispatcherTests { SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1)); - dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1))); + dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1))); SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); scheduler.start(); dispatcher.setMessagingTaskScheduler(scheduler); @@ -79,8 +79,8 @@ public class DefaultMessageDispatcherTests { SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1)); - dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1))); + dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1))); dispatcher.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); @@ -99,13 +99,13 @@ public class DefaultMessageDispatcherTests { SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)) { @Override public void start() { } }); - dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1)); - dispatcher.addHandler(new ConcurrentHandler(handler3, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1))); + dispatcher.addHandler(new ConcurrentHandler(handler3, new ConcurrencyPolicy(1, 1))); dispatcher.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); @@ -125,13 +125,13 @@ public class DefaultMessageDispatcherTests { SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1)); - dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1))); + dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)) { @Override public void start() { } }); - dispatcher.addHandler(new ConcurrentHandler(handler3, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler3, new ConcurrencyPolicy(1, 1))); dispatcher.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); @@ -161,14 +161,14 @@ public class DefaultMessageDispatcherTests { channel.getDispatcherPolicy().setRetryInterval(3); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1)); - dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1))); + dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)) { @Override public Message handle(Message message) { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new ConcurrentHandler(handler3, 1, 1)); + dispatcher.addHandler(new ConcurrentHandler(handler3, new ConcurrencyPolicy(1, 1))); SimpleChannel errorChannel = new SimpleChannel(); SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); @@ -249,7 +249,7 @@ public class DefaultMessageDispatcherTests { channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)) { @Override public Message handle(Message message) { rejectedCounter1.incrementAndGet(); @@ -257,7 +257,7 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)) { @Override public Message handle(Message message) { rejectedCounter2.incrementAndGet(); @@ -293,7 +293,7 @@ public class DefaultMessageDispatcherTests { SimpleChannel channel = new SimpleChannel(dispatcherPolicy); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)) { @Override public Message handle(Message message) { rejectedCounter1.incrementAndGet(); @@ -301,7 +301,7 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)) { @Override public Message handle(Message message) { if (rejectedCounter2.get() == 1) { @@ -312,7 +312,7 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new ConcurrentHandler(handler3, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler3, new ConcurrencyPolicy(1, 1)) { @Override public Message handle(Message message) { rejectedCounter3.incrementAndGet(); @@ -347,7 +347,7 @@ public class DefaultMessageDispatcherTests { SimpleChannel channel = new SimpleChannel(dispatcherPolicy); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)) { @Override public Message handle(Message message) { if (rejectedCounter1.get() == 2) { @@ -358,7 +358,7 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(); } }); - dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { + dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)) { @Override public Message handle(Message message) { if (rejectedCounter2.get() == 4) {