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 2f4be97634..08882084e1 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 @@ -36,7 +36,6 @@ import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.dispatcher.DefaultMessageDispatcher; -import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultMessageEndpoint; @@ -179,22 +178,12 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public void registerChannel(String name, MessageChannel channel) { - this.registerChannel(name, channel, null); - } - - public void registerChannel(String name, MessageChannel channel, DispatcherPolicy dispatcherPolicy) { if (!this.initialized) { this.initialize(); } channel.setName(name); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setMessagingTaskScheduler(this.taskScheduler); - if (dispatcherPolicy != null) { - dispatcher.setMaxMessagesPerTask(dispatcherPolicy.getMaxMessagesPerTask()); - dispatcher.setReceiveTimeout(dispatcherPolicy.getReceiveTimeout()); - dispatcher.setRejectionLimit(dispatcherPolicy.getRejectionLimit()); - dispatcher.setRetryInterval(dispatcherPolicy.getRetryInterval()); - } this.dispatchers.put(channel, dispatcher); this.channelRegistry.registerChannel(name, channel); if (logger.isInfoEnabled()) { 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 ab52c2d55d..ef46200346 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 @@ -16,6 +16,7 @@ package org.springframework.integration.channel; +import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.message.Message; /** @@ -36,11 +37,9 @@ public interface MessageChannel { 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 - * message to each of the dispatcher's handlers. + * Return this channel's dispatcher policy */ - boolean isPublishSubscribe(); + DispatcherPolicy getDispatcherPolicy(); /** * Send a message, blocking indefinitely if necessary. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java index b7aec8a16b..4b8bc462ad 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java @@ -21,6 +21,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import org.springframework.beans.factory.BeanNameAware; +import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.message.Message; /** @@ -36,23 +37,38 @@ public class SimpleChannel implements MessageChannel, BeanNameAware { private String name; - private boolean publishSubscribe = false; + private final DispatcherPolicy dispatcherPolicy; private BlockingQueue> queue; + /** + * Create a channel with the specified queue capacity and dispatcher policy. + */ + public SimpleChannel(int capacity, DispatcherPolicy dispatcherPolicy) { + this.queue = new LinkedBlockingQueue>(capacity); + this.dispatcherPolicy = (dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy(); + } + /** * Create a channel with the specified queue capacity. */ public SimpleChannel(int capacity) { - queue = new LinkedBlockingQueue>(capacity); + this(capacity, null); + } + + /** + * Create a channel with the default queue capacity and the specified dispatcher policy. + */ + public SimpleChannel(DispatcherPolicy dispatcherPolicy) { + this(DEFAULT_CAPACITY, dispatcherPolicy); } /** * Create a channel with the default queue capacity. */ public SimpleChannel() { - this(DEFAULT_CAPACITY); + this(DEFAULT_CAPACITY, null); } @@ -70,12 +86,8 @@ public class SimpleChannel implements MessageChannel, BeanNameAware { return this.name; } - public boolean isPublishSubscribe() { - return this.publishSubscribe; - } - - public void setPublishSubscribe(boolean publishSubscribe) { - this.publishSubscribe = publishSubscribe; + public DispatcherPolicy getDispatcherPolicy() { + return this.dispatcherPolicy; } /** 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 c58bae544a..b479f7f877 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 @@ -24,6 +24,7 @@ import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.BeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.util.StringUtils; /** @@ -39,8 +40,6 @@ public class ChannelParser implements BeanDefinitionParser { private static final String PUBLISH_SUBSCRIBE_ATTRIBUTE = "publish-subscribe"; - private static final String PUBLISH_SUBSCRIBE_PROPERTY = "publishSubscribe"; - public BeanDefinition parse(Element element, ParserContext parserContext) { RootBeanDefinition channelDef = new RootBeanDefinition(SimpleChannel.class); @@ -51,7 +50,7 @@ public class ChannelParser implements BeanDefinitionParser { } String isPublishSubscribe = element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE); if ("true".equals(isPublishSubscribe)) { - channelDef.getPropertyValues().addPropertyValue(PUBLISH_SUBSCRIBE_PROPERTY, Boolean.TRUE); + channelDef.getConstructorArgumentValues().addGenericArgumentValue(new DispatcherPolicy(true)); } String beanName = element.getAttribute(ID_ATTRIBUTE); if (!StringUtils.hasText(beanName)) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java index 844f1c0635..0d6434970e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java @@ -27,7 +27,8 @@ import org.springframework.util.Assert; /** * Message retriever that polls a {@link MessageChannel}. The number of * messages retrieved per poll is limited by the 'maxMessagesPerTask' - * property, and the timeout for each receive call is determined by the 'receiveTimeout' + * property of the channel's dispatcher policy, and the timeout for each receive + * call is determined by the dispatcher policy's 'receiveTimeout' * property. In general, it is recommended to use a value of 1 for * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise the * retriever may be holding on to available messages while waiting for @@ -39,10 +40,6 @@ public class ChannelPollingMessageRetriever implements MessageRetriever { private MessageChannel channel; - private int maxMessagesPerTask = DispatcherPolicy.DEFAULT_MAX_MESSAGES_PER_TASK; - - private long receiveTimeout = DispatcherPolicy.DEFAULT_RECEIVE_TIMEOUT; - public ChannelPollingMessageRetriever(MessageChannel channel) { Assert.notNull(channel, "'channel' must not be null"); @@ -50,28 +47,20 @@ public class ChannelPollingMessageRetriever implements MessageRetriever { } - public void setMaxMessagesPerTask(int maxMessagesPerTask) { - Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least 1"); - this.maxMessagesPerTask = maxMessagesPerTask; - } - - public void setReceiveTimeout(long receiveTimeout) { - this.receiveTimeout = receiveTimeout; - } - public MessageChannel getChannel() { return this.channel; } public Collection> retrieveMessages() { List> messages = new LinkedList>(); - while (messages.size() < this.maxMessagesPerTask) { + while (messages.size() < this.channel.getDispatcherPolicy().getMaxMessagesPerTask()) { Message message = null; - if (this.receiveTimeout < 0) { + long timeout = this.channel.getDispatcherPolicy().getReceiveTimeout(); + if (timeout < 0) { message = this.channel.receive(); } else { - message = this.channel.receive(this.receiveTimeout); + message = this.channel.receive(timeout); } if (message == null) { return messages; 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 b31507e3f4..150db459b8 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 @@ -50,16 +50,6 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas private MessageChannel channel; - private int maxMessagesPerTask = DispatcherPolicy.DEFAULT_MAX_MESSAGES_PER_TASK; - - private long receiveTimeout = DispatcherPolicy.DEFAULT_RECEIVE_TIMEOUT; - - private int rejectionLimit = DispatcherPolicy.DEFAULT_REJECTION_LIMIT; - - private long retryInterval = DispatcherPolicy.DEFAULT_RETRY_INTERVAL; - - private boolean shouldFailOnRejectionLimit = true; - private MessagingTaskScheduler scheduler; private Schedule defaultSchedule = new PollingSchedule(5); @@ -79,29 +69,6 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas } - public void setMaxMessagesPerTask(int maxMessagesPerTask) { - Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least 1"); - this.maxMessagesPerTask = maxMessagesPerTask; - } - - public void setReceiveTimeout(long receiveTimeout) { - this.receiveTimeout = receiveTimeout; - } - - public void setRejectionLimit(int rejectionLimit) { - Assert.isTrue(rejectionLimit > 0, "'rejectionLimit' must be at least 1"); - this.rejectionLimit = rejectionLimit; - } - - public void setShouldFailOnRejectionLimit(boolean shouldFailOnRejectionLimit) { - this.shouldFailOnRejectionLimit = shouldFailOnRejectionLimit; - } - - public void setRetryInterval(long retryInterval) { - Assert.isTrue(retryInterval >= 0, "'retryInterval' must not be negative"); - this.retryInterval = retryInterval; - } - public void setMessagingTaskScheduler(MessagingTaskScheduler scheduler) { Assert.notNull(scheduler, "'scheduler' must not be null"); this.scheduler = scheduler; @@ -121,7 +88,7 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas if (schedule == null) { schedule = this.defaultSchedule; } - else if (this.channel.isPublishSubscribe()) { + else if (this.channel.getDispatcherPolicy().isPublishSubscribe()) { if (logger.isInfoEnabled()) { logger.info("This dispatcher broadcasts messages for a publish-subscribe channel. " + "Therefore all handlers are scheduled with its 'defaultSchedule', " + @@ -163,15 +130,8 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas 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); + DispatcherTask task = new DispatcherTask(channel); 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(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java index 9cf0604465..c8f78999a8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java @@ -34,6 +34,8 @@ public class DispatcherPolicy { public final static long DEFAULT_RETRY_INTERVAL = 1000; + private final boolean publishSubscribe; + private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK; private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; @@ -42,40 +44,106 @@ public class DispatcherPolicy { private long retryInterval = DEFAULT_RETRY_INTERVAL; + private boolean shouldFailOnRejectionLimit = true; + + public DispatcherPolicy() { + this.publishSubscribe = false; + } + + /** + * Create a DispatcherPolicy. + * + * @param publishSubscribe whether the dispatcher should attempt to publish + * to all of its subscribed handlers. If 'false' it will attempt + * to send to a single handler (point-to-point). + */ + public DispatcherPolicy(boolean publishSubscribe) { + this.publishSubscribe = publishSubscribe; + } + + + /** + * Return whether the dispatcher should attempt to publish to all of its handlers. + * This property is immutable. + */ + public boolean isPublishSubscribe() { + return this.publishSubscribe; + } + + /** + * Return the maximum number of messages for each retrieval attempt. + */ public int getMaxMessagesPerTask() { return this.maxMessagesPerTask; } + /** + * Set the maximum number of messages for each retrieval attempt. + */ public void setMaxMessagesPerTask(int maxMessagesPerTask) { Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagePerTask' must be at least 1"); this.maxMessagesPerTask = maxMessagesPerTask; } + /** + * Return the maximum amount of time in milliseconds to wait for a message to be available. + */ public long getReceiveTimeout() { return this.receiveTimeout; } + /** + * Set the maximum amount of time in milliseconds to wait for a message to be available. + */ public void setReceiveTimeout(long receiveTimeout) { this.receiveTimeout = receiveTimeout; } + /** + * Return the maximum number of retries upon rejection. + */ public int getRejectionLimit() { return this.rejectionLimit; } + /** + * Set the maximum number of retries upon rejection. + */ public void setRejectionLimit(int rejectionLimit) { Assert.isTrue(rejectionLimit > 0, "'rejectionLimit' must be at least 1"); this.rejectionLimit = rejectionLimit; } + /** + * Return the amount of time in milliseconds to wait between rejections. + */ public long getRetryInterval() { return this.retryInterval; } + /** + * Set the amount of time in milliseconds to wait between rejections. + */ public void setRetryInterval(long retryInterval) { Assert.isTrue(retryInterval >= 0, "'retryInterval' must not be negative"); this.retryInterval = retryInterval; } + /** + * Return whether an exception should be thrown when this dispatcher's + * {@link #rejectionLimit} is reached. + */ + public boolean getShouldFailOnRejectionLimit() { + return this.shouldFailOnRejectionLimit; + } + + /** + * Specify whether an exception should be thrown when this dispatcher's + * {@link #rejectionLimit} is reached. The default value is 'true'. + */ + public void setShouldFailOnRejectionLimit(boolean shouldFailOnRejectionLimit) { + this.shouldFailOnRejectionLimit = shouldFailOnRejectionLimit; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java index f9d92505c3..aa3797a70c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java @@ -46,15 +46,9 @@ public class DispatcherTask implements MessagingTask { private Log logger = LogFactory.getLog(this.getClass()); - private boolean publishSubscribe = false; - private Schedule schedule; - private int rejectionLimit = 5; - - private long retryInterval = 1000; - - private boolean shouldFailOnRejectionLimit = true; + private DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); private MessageRetriever retriever; @@ -64,22 +58,20 @@ public class DispatcherTask implements MessagingTask { public DispatcherTask(MessageChannel channel) { Assert.notNull(channel, "'channel' must not be null"); this.retriever = new ChannelPollingMessageRetriever(channel); - this.publishSubscribe = channel.isPublishSubscribe(); + DispatcherPolicy dispatcherPolicy = channel.getDispatcherPolicy(); + if (dispatcherPolicy != null) { + this.dispatcherPolicy = dispatcherPolicy; + } } public DispatcherTask(MessageRetriever retriever) { Assert.notNull(retriever, "'retriever' must not be null"); if (retriever instanceof ChannelPollingMessageRetriever) { - this.publishSubscribe = ((ChannelPollingMessageRetriever) retriever).getChannel().isPublishSubscribe(); + this.dispatcherPolicy = ((ChannelPollingMessageRetriever) retriever).getChannel().getDispatcherPolicy(); } this.retriever = retriever; } - - public void setPublishSubscribe(boolean publishSubscribe) { - this.publishSubscribe = publishSubscribe; - } - public void setSchedule(Schedule schedule) { Assert.notNull(schedule, "'schedule' must not be null"); this.schedule = schedule; @@ -89,24 +81,6 @@ public class DispatcherTask implements MessagingTask { return this.schedule; } - public void setRejectionLimit(int rejectionLimit) { - Assert.isTrue(rejectionLimit > 0, "'rejectionLimit' must be at least 1"); - this.rejectionLimit = rejectionLimit; - } - - public void setRetryInterval(long retryInterval) { - Assert.isTrue(retryInterval > 0, "'retryInterval' must not be negative"); - this.retryInterval = retryInterval; - } - - /** - * Specify whether an exception should be thrown when this dispatcher's - * {@link #rejectionLimit} is reached. The default value is 'true'. - */ - public void setShouldFailOnRejectionLimit(boolean shouldFailOnRejectionLimit) { - this.shouldFailOnRejectionLimit = shouldFailOnRejectionLimit; - } - public void addHandler(MessageHandler handler) { Assert.notNull(handler, "'handler' must not be null"); this.handlers.add(handler); @@ -134,15 +108,15 @@ public class DispatcherTask implements MessagingTask { protected boolean dispatchMessage(Message message) { int attempts = 0; List targets = new ArrayList(this.handlers); - while (attempts < this.rejectionLimit) { + while (attempts < this.dispatcherPolicy.getRejectionLimit()) { if (attempts > 0) { if (logger.isDebugEnabled()) { logger.debug("handler(s) rejected message after " + attempts - + " attempt(s), will try again after 'retryInterval' of " + this.retryInterval - + " milliseconds"); + + " attempt(s), will try again after 'retryInterval' of " + + this.dispatcherPolicy.getRetryInterval() + " milliseconds"); } try { - Thread.sleep(this.retryInterval); + Thread.sleep(this.dispatcherPolicy.getRetryInterval()); } catch (InterruptedException iex) { Thread.currentThread().interrupt(); @@ -161,7 +135,7 @@ public class DispatcherTask implements MessagingTask { MessageHandler handler = iter.next(); try { handler.handle(message); - if (!this.publishSubscribe) { + if (!this.dispatcherPolicy.isPublishSubscribe()) { return true; } iter.remove(); @@ -188,9 +162,9 @@ public class DispatcherTask implements MessagingTask { } attempts++; } - if (this.shouldFailOnRejectionLimit) { - throw new MessageDeliveryException("Dispatcher reached rejection limit of " + this.rejectionLimit - + ". Consider increasing the executor's concurrency and/or raising the 'rejectionLimit'."); + if (this.dispatcherPolicy.getShouldFailOnRejectionLimit()) { + throw new MessageDeliveryException("Dispatcher reached rejection limit of " + this.dispatcherPolicy.getRejectionLimit() + + ". Consider increasing the handler's concurrency and/or the dispatcherPolicy's 'rejectionLimit'."); } return false; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java index a01df0d6bd..d371036d23 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java @@ -26,6 +26,7 @@ import org.junit.Test; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.dispatcher.ChannelPollingMessageRetriever; +import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.dispatcher.DispatcherTask; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; @@ -70,10 +71,11 @@ public class ByteStreamTargetAdapterTests { @Test public void testMaxMessagesPerTaskSameAsMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setMaxMessagesPerTask(3); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setMaxMessagesPerTask(3); DispatcherTask dispatcherTask = new DispatcherTask(retriever); dispatcherTask.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3})); @@ -89,10 +91,11 @@ public class ByteStreamTargetAdapterTests { @Test public void testMaxMessagesPerTaskLessThanMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setMaxMessagesPerTask(2); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setMaxMessagesPerTask(2); DispatcherTask dispatcherTask = new DispatcherTask(retriever); dispatcherTask.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3})); @@ -107,12 +110,12 @@ public class ByteStreamTargetAdapterTests { @Test public void testMaxMessagesPerTaskExceedsMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setMaxMessagesPerTask(5); - retriever.setReceiveTimeout(0); - DispatcherTask dispatcherTask = new DispatcherTask(retriever); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setMaxMessagesPerTask(5); + dispatcherPolicy.setReceiveTimeout(0); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); + DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3})); channel.send(new GenericMessage(new byte[] {4,5,6})); @@ -126,12 +129,12 @@ public class ByteStreamTargetAdapterTests { @Test public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setMaxMessagesPerTask(2); - retriever.setReceiveTimeout(0); - DispatcherTask dispatcherTask = new DispatcherTask(retriever); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setMaxMessagesPerTask(2); + dispatcherPolicy.setReceiveTimeout(0); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); + DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3})); channel.send(new GenericMessage(new byte[] {4,5,6})); @@ -150,12 +153,12 @@ public class ByteStreamTargetAdapterTests { @Test public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setMaxMessagesPerTask(5); - retriever.setReceiveTimeout(0); - DispatcherTask dispatcherTask = new DispatcherTask(retriever); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setMaxMessagesPerTask(5); + dispatcherPolicy.setReceiveTimeout(0); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); + DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3})); channel.send(new GenericMessage(new byte[] {4,5,6})); @@ -173,11 +176,12 @@ public class ByteStreamTargetAdapterTests { @Test public void testStreamResetBetweenDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setMaxMessagesPerTask(2); + dispatcherPolicy.setReceiveTimeout(0); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setMaxMessagesPerTask(2); - retriever.setReceiveTimeout(0); DispatcherTask dispatcherTask = new DispatcherTask(retriever); dispatcherTask.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3})); @@ -196,12 +200,12 @@ public class ByteStreamTargetAdapterTests { @Test public void testStreamWriteBetweenDispatches() throws IOException { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setMaxMessagesPerTask(2); - retriever.setReceiveTimeout(0); - DispatcherTask dispatcherTask = new DispatcherTask(retriever); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setMaxMessagesPerTask(2); + dispatcherPolicy.setReceiveTimeout(0); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); + DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3})); channel.send(new GenericMessage(new byte[] {4,5,6})); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java index 7411102191..dbc798b1d9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java @@ -24,7 +24,7 @@ import org.junit.Test; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.dispatcher.ChannelPollingMessageRetriever; +import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.dispatcher.DispatcherTask; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; @@ -87,11 +87,11 @@ public class CharacterStreamTargetAdapterTests { @Test public void testMaxMessagesPerTaskSameAsMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setMaxMessagesPerTask(2); - DispatcherTask dispatcherTask = new DispatcherTask(retriever); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setMaxMessagesPerTask(2); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); + DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); channel.send(new StringMessage("foo")); channel.send(new StringMessage("bar")); @@ -103,13 +103,13 @@ public class CharacterStreamTargetAdapterTests { @Test public void testMaxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setMaxMessagesPerTask(10); + dispatcherPolicy.setReceiveTimeout(0); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); + DispatcherTask dispatcherTask = new DispatcherTask(channel); adapter.setShouldAppendNewLine(true); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setReceiveTimeout(0); - retriever.setMaxMessagesPerTask(10); - DispatcherTask dispatcherTask = new DispatcherTask(retriever); dispatcherTask.addHandler(adapter); channel.send(new StringMessage("foo")); channel.send(new StringMessage("bar")); @@ -137,12 +137,12 @@ public class CharacterStreamTargetAdapterTests { @Test public void testTwoNonStringObjectWithOutNewLines() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setReceiveTimeout(0); - retriever.setMaxMessagesPerTask(2); - DispatcherTask dispatcherTask = new DispatcherTask(retriever); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setReceiveTimeout(0); + dispatcherPolicy.setMaxMessagesPerTask(2); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); + DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); @@ -156,13 +156,13 @@ public class CharacterStreamTargetAdapterTests { @Test public void testTwoNonStringObjectWithNewLines() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setReceiveTimeout(0); + dispatcherPolicy.setMaxMessagesPerTask(2); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); + DispatcherTask dispatcherTask = new DispatcherTask(channel); adapter.setShouldAppendNewLine(true); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - retriever.setReceiveTimeout(0); - retriever.setMaxMessagesPerTask(2); - DispatcherTask dispatcherTask = new DispatcherTask(retriever); dispatcherTask.addHandler(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 5dc182fda6..26b8f57d0a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -33,6 +33,7 @@ import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.adapter.SourceAdapter; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.GenericMessage; @@ -127,8 +128,8 @@ public class MessageBusTests { @Test public void testBothHandlersReceivePublishSubscribeMessage() { - SimpleChannel inputChannel = new SimpleChannel(); - inputChannel.setPublishSubscribe(true); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true); + SimpleChannel inputChannel = new SimpleChannel(dispatcherPolicy); SimpleChannel outputChannel1 = new SimpleChannel(); SimpleChannel outputChannel2 = new SimpleChannel(); MessageHandler handler1 = new MessageHandler() { 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 475c7885eb..1cd39b1846 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 @@ -76,8 +76,7 @@ public class DefaultMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(2); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(); - channel.setPublishSubscribe(true); + 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)); @@ -123,8 +122,7 @@ public class DefaultMessageDispatcherTests { 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); + 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)); @@ -158,12 +156,11 @@ public class DefaultMessageDispatcherTests { 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); + SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); + channel.getDispatcherPolicy().setRejectionLimit(2); + channel.getDispatcherPolicy().setRetryInterval(3); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.setRejectionLimit(2); - dispatcher.setRetryInterval(3); dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1)); dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { @Override @@ -192,13 +189,12 @@ public class DefaultMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(3); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(); - channel.setPublishSubscribe(true); + SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); + channel.getDispatcherPolicy().setRejectionLimit(2); + channel.getDispatcherPolicy().setRetryInterval(3); + channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.setRejectionLimit(2); - dispatcher.setRetryInterval(3); - dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addHandler(handler1); dispatcher.addHandler(new MessageHandler() { public Message handle(Message message) { @@ -215,30 +211,16 @@ public class DefaultMessageDispatcherTests { @Test public void testNonBroadcastingDispatcherReachesRejectionLimitAndShouldFail() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(4); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); - MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(); + MessageHandler handler1 = TestHandlers.rejectingCountDownHandler(latch); + MessageHandler handler2 = TestHandlers.rejectingCountDownHandler(latch); + SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(false)); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.setRejectionLimit(2); - dispatcher.setRetryInterval(3); - dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { - @Override - public Message handle(Message message) { - latch.countDown(); - throw new MessageHandlerRejectedExecutionException(); - } - }); - dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) { - @Override - public Message handle(Message message) { - latch.countDown(); - throw new MessageHandlerRejectedExecutionException(); - } - }); + channel.getDispatcherPolicy().setRejectionLimit(2); + channel.getDispatcherPolicy().setRetryInterval(3); + dispatcher.addHandler(handler1); + dispatcher.addHandler(handler2); SimpleChannel errorChannel = new SimpleChannel(); SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); @@ -262,11 +244,11 @@ public class DefaultMessageDispatcherTests { MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); + channel.getDispatcherPolicy().setRejectionLimit(2); + channel.getDispatcherPolicy().setRetryInterval(3); + channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.setRejectionLimit(2); - dispatcher.setRetryInterval(3); - dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { @Override public Message handle(Message message) { @@ -304,12 +286,13 @@ public class DefaultMessageDispatcherTests { MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); - SimpleChannel channel = new SimpleChannel(); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setRejectionLimit(2); + dispatcherPolicy.setRetryInterval(3); + dispatcherPolicy.setShouldFailOnRejectionLimit(false); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.setRejectionLimit(2); - dispatcher.setRetryInterval(3); - dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { @Override public Message handle(Message message) { @@ -357,13 +340,13 @@ public class DefaultMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(8); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(); - channel.setPublishSubscribe(true); + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true); + dispatcherPolicy.setRejectionLimit(5); + dispatcherPolicy.setRetryInterval(3); + dispatcherPolicy.setShouldFailOnRejectionLimit(false); + SimpleChannel channel = new SimpleChannel(dispatcherPolicy); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); - dispatcher.setRejectionLimit(5); - dispatcher.setRetryInterval(3); - dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) { @Override public Message handle(Message message) { @@ -471,8 +454,7 @@ public class DefaultMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(1); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(); - channel.setPublishSubscribe(true); + SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(); 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 718a93dd03..f0d115b033 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 @@ -19,6 +19,7 @@ package org.springframework.integration.handler; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicInteger; +import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; import org.springframework.integration.message.Message; /** @@ -40,6 +41,30 @@ public abstract class TestHandlers { }; } + /** + * Create a {@link MessageHandler} that throws a {@link MessageHandlerRejectedExecutionException}. + */ + public final static MessageHandler rejectingHandler() { + return new MessageHandler() { + public Message handle(Message message) { + throw new MessageHandlerRejectedExecutionException(); + } + }; + } + + /** + * Create a {@link MessageHandler} that counts down on the provided latch and + * then throws a {@link MessageHandlerRejectedExecutionException}. + */ + public final static MessageHandler rejectingCountDownHandler(final CountDownLatch latch) { + return new MessageHandler() { + public Message handle(Message message) { + latch.countDown(); + throw new MessageHandlerRejectedExecutionException(); + } + }; + } + /** * Create a {@link MessageHandler} that increments the provided counter. */