diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java index d5145a8e01..3fc6bebb9a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java @@ -21,6 +21,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.MessagingConfigurationException; +import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageMapper; @@ -32,7 +33,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class AbstractSourceAdapter implements SourceAdapter, InitializingBean { +public abstract class AbstractSourceAdapter implements SourceAdapter, InitializingBean { protected Log logger = LogFactory.getLog(this.getClass()); @@ -40,6 +41,8 @@ public class AbstractSourceAdapter implements SourceAdapter, InitializingBean private MessageMapper mapper = new SimplePayloadMessageMapper(); + private ConsumerPolicy consumerPolicy; + private long sendTimeout = -1; @@ -61,6 +64,15 @@ public class AbstractSourceAdapter implements SourceAdapter, InitializingBean return this.mapper; } + public void setConsumerPolicy(ConsumerPolicy consumerPolicy) { + Assert.notNull(consumerPolicy, "'consumerPolicy' must not be null"); + this.consumerPolicy = consumerPolicy; + } + + public ConsumerPolicy getConsumerPolicy() { + return this.consumerPolicy; + } + public final void afterPropertiesSet() { if (this.channel == null) { throw new MessagingConfigurationException("'channel' is required"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index 6578eec675..b94358e88b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -38,31 +38,26 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements private PollableSource source; - private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(DEFAULT_PERIOD); - public PollingSourceAdapter(PollableSource source) { Assert.notNull(source, "'source' must not be null"); this.source = source; + this.setConsumerPolicy(ConsumerPolicy.newPollingPolicy(DEFAULT_PERIOD)); } public void setPeriod(int period) { Assert.isTrue(period > 0, "'period' must be a positive value"); - this.policy.setPeriod(period); + this.getConsumerPolicy().setPeriod(period); } public void setMaxMessagesPerTask(int maxMessagesPerTask) { Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be a positive value"); - this.policy.setMaxMessagesPerTask(maxMessagesPerTask); - } - - public ConsumerPolicy getConsumerPolicy() { - return this.policy; + this.getConsumerPolicy().setMaxMessagesPerTask(maxMessagesPerTask); } public int dispatch() { int messagesProcessed = 0; - int limit = this.policy.getMaxMessagesPerTask(); + int limit = this.getConsumerPolicy().getMaxMessagesPerTask(); Collection results = this.source.poll(limit); if (results != null) { if (results.size() > limit) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java index 5fe72ad111..161e6b3b5c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java @@ -16,6 +16,7 @@ package org.springframework.integration.adapter; +import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.MessageChannel; /** @@ -25,6 +26,8 @@ import org.springframework.integration.channel.MessageChannel; */ public interface SourceAdapter { + ConsumerPolicy getConsumerPolicy(); + void setChannel(MessageChannel channel); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java index 4ee647cd68..b96dc42dd3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java @@ -83,11 +83,6 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter this.taskExecutor = taskExecutor; } - public void setPolicy(ConsumerPolicy policy) { - Assert.notNull(policy, "'policy' must not be null"); - this.policy = policy; - } - @Override public void initialize() { if (this.container == null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java index 68f54b047b..24e192f87c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java @@ -24,6 +24,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.message.Message; +import org.springframework.util.Assert; /** * Abstract base class for message dispatchers. Delegates to a @@ -46,6 +47,7 @@ public abstract class AbstractMessageDispatcher implements MessageDispatcher { public void addExecutor(MessageReceivingExecutor executor) { + Assert.notNull(executor, "'executor' must not be null"); executor.start(); this.executors.add(executor); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultMessageDispatcher.java new file mode 100644 index 0000000000..2aa2c1f7a6 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultMessageDispatcher.java @@ -0,0 +1,136 @@ +/* + * 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.bus; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.RejectedExecutionException; + +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.message.Message; +import org.springframework.util.Assert; + +/** + * The default implementation of {@link MessageDispatcher}. If + * {@link #broadcast} is set to false (the default), each message + * will be sent to a single {@link MessageReceivingExecutor}. Otherwise, each + * retrieved {@link Message} will be sent to all executors. + * + * @author Mark Fisher + */ +public class DefaultMessageDispatcher extends AbstractMessageDispatcher { + + private boolean broadcast = false; + + private int rejectionLimit = 5; + + private long retryInterval = 1000; + + private boolean shouldFailOnRejectionLimit = true; + + + public DefaultMessageDispatcher(MessageRetriever retriever) { + super(retriever); + } + + + public void setBroadcast(boolean broadcast) { + this.broadcast = broadcast; + } + + 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; + } + + @Override + protected boolean dispatchMessage(Message message) { + int attempts = 0; + List targets = new ArrayList(this.getExecutors()); + while (attempts < this.rejectionLimit) { + if (attempts > 0) { + if (logger.isDebugEnabled()) { + logger.debug("executor(s) rejected message after " + attempts + + " attempt(s), will try again after 'retryInterval' of " + this.retryInterval + + " milliseconds"); + } + try { + Thread.sleep(this.retryInterval); + } + catch (InterruptedException iex) { + Thread.currentThread().interrupt(); + return false; + } + } + Iterator iter = targets.iterator(); + if (!iter.hasNext()) { + if (logger.isWarnEnabled()) { + logger.warn("dispatcher has no active executors"); + } + return false; + } + boolean encounteredRejection = false; + while (iter.hasNext()) { + MessageReceivingExecutor executor = iter.next(); + if (executor == null || !executor.isRunning()) { + if (logger.isInfoEnabled()) { + logger.info("skipping inactive executor"); + } + iter.remove(); + continue; + } + try { + executor.processMessage(message); + if (!this.broadcast) { + return true; + } + iter.remove(); + if (!iter.hasNext() && !encounteredRejection) { + return true; + } + } + catch (RejectedExecutionException rex) { + encounteredRejection = true; + if (logger.isDebugEnabled()) { + logger.debug("executor rejected task, continuing with other executors if available", rex); + } + } + } + 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'."); + } + return false; + } + +} 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 4e95990a40..86805cc4e1 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 @@ -208,9 +208,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public void registerSourceAdapter(String name, SourceAdapter adapter) { + ConsumerPolicy policy = adapter.getConsumerPolicy(); if (adapter instanceof MessageDispatcher) { MessageDispatcher dispatcher = (MessageDispatcher) adapter; - ConsumerPolicy policy = dispatcher.getConsumerPolicy(); DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy); this.addDispatcherTask(dispatcherTask); } @@ -230,7 +230,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif MessageChannel channel = adapter.getChannel(); ConsumerPolicy policy = adapter.getConsumerPolicy(); MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); - UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.setRejectionLimit(policy.getRejectionLimit()); + dispatcher.setRetryInterval(policy.getRetryInterval()); MessageReceivingExecutor executor = new MessageReceivingExecutor(adapter, policy.getConcurrency(), policy.getMaxConcurrency()); dispatcher.addExecutor(executor); this.addLifecycleComponent(name + "-executor", executor); @@ -245,6 +247,10 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif String channelName = subscription.getChannel(); String endpointName = subscription.getEndpoint(); ConsumerPolicy policy = subscription.getPolicy(); + MessageEndpoint endpoint = this.endpoints.get(endpointName); + if (endpoint == null) { + throw new MessagingException("Cannot activate subscription, unknown endpoint '" + endpointName + "'"); + } MessageChannel channel = this.lookupChannel(channelName); if (channel == null) { if (this.autoCreateChannels == false) { @@ -257,10 +263,6 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif channel = new SimpleChannel(); this.registerChannel(channelName, channel); } - MessageEndpoint endpoint = this.endpoints.get(endpointName); - if (endpoint == null) { - throw new MessagingException("Cannot activate subscription, unknown endpoint '" + endpointName + "'"); - } if (logger.isInfoEnabled()) { logger.info("activated subscription to channel '" + channelName + "' for endpoint '" + endpointName + "'"); @@ -269,7 +271,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif this.receiverExecutors.put(endpoint, executor); this.lifecycleComponents.put(endpointName + "-executor", executor); MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); - UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.setRejectionLimit(policy.getRejectionLimit()); + dispatcher.setRetryInterval(policy.getRetryInterval()); dispatcher.addExecutor(executor); DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy); if (this.isRunning()) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java index e350f83aa8..7e070d6399 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java @@ -23,8 +23,6 @@ package org.springframework.integration.bus; */ public interface MessageDispatcher { - ConsumerPolicy getConsumerPolicy(); - int dispatch(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java deleted file mode 100644 index 132e8ffc3b..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java +++ /dev/null @@ -1,95 +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.bus; - -import java.util.Iterator; -import java.util.concurrent.RejectedExecutionException; - -import org.springframework.integration.MessageDeliveryException; -import org.springframework.integration.message.Message; - -/** - * A {@link MessageDispatcher} implementation that dispatches each retrieved - * {@link Message} to a single {@link MessageReceivingExecutor}. - * - * @author Mark Fisher - */ -public class UnicastMessageDispatcher extends AbstractMessageDispatcher { - - private ConsumerPolicy policy; - - - public UnicastMessageDispatcher(MessageRetriever retriever, ConsumerPolicy policy) { - super(retriever); - this.policy = policy; - } - - - public ConsumerPolicy getConsumerPolicy() { - return this.policy; - } - - @Override - protected boolean dispatchMessage(Message message) { - int attempts = 0; - while (attempts < policy.getRejectionLimit()) { - if (attempts > 0) { - if (logger.isDebugEnabled()) { - logger.debug("executor(s) rejected message after " + attempts - + " attempt(s), will try again after 'retryInterval' of " + this.policy.getRetryInterval() - + " milliseconds"); - } - try { - Thread.sleep(policy.getRetryInterval()); - } - catch (InterruptedException iex) { - Thread.currentThread().interrupt(); - return false; - } - } - Iterator iter = this.getExecutors().iterator(); - if (!iter.hasNext()) { - if (logger.isWarnEnabled()) { - logger.warn("dispatcher has no active executors"); - } - return false; - } - while (iter.hasNext()) { - MessageReceivingExecutor executor = iter.next(); - if (executor == null || !executor.isRunning()) { - if (logger.isInfoEnabled()) { - logger.info("skipping inactive executor"); - } - continue; - } - try { - executor.processMessage(message); - return true; - } - catch (RejectedExecutionException rex) { - if (logger.isDebugEnabled()) { - logger.debug("executor rejected task, continuing with other executors if available", rex); - } - } - } - attempts++; - } - throw new MessageDeliveryException("Dispatcher reached rejection limit of " + - this.policy.getRejectionLimit() + ". Consider increasing the concurrency and/or raising the limit."); - } - -} 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 new file mode 100644 index 0000000000..0c510df0bd --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java @@ -0,0 +1,344 @@ +/* + * 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.bus; + +import static org.junit.Assert.assertEquals; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.endpoint.GenericMessageEndpoint; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class DefaultMessageDispatcherTests { + + @Test + public void testNonBroadcastingDispatcherSendsToExactlyOneEndpoint() throws InterruptedException { + 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); + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1)); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1)); + dispatcher.dispatch(); + latch.await(100, TimeUnit.MILLISECONDS); + assertEquals("exactly one endpoint should have received message", 1, counter1.get() + counter2.get()); + } + + @Test + public void testBroadcastingDispatcherSendsToAllEndpoints() throws InterruptedException { + 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); + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.setBroadcast(true); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1)); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1)); + dispatcher.dispatch(); + latch.await(100, TimeUnit.MILLISECONDS); + assertEquals("both endpoints should have received message", 2, counter1.get() + counter2.get()); + } + + @Test + public void testNonBroadcastingDispatcherSkipsInactiveExecutor() throws InterruptedException { + final AtomicInteger counter1 = new AtomicInteger(); + 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); + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1) { + @Override + public void start() { + } + }); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1)); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint3, 1, 1)); + dispatcher.dispatch(); + latch.await(100, TimeUnit.MILLISECONDS); + assertEquals("inactive endpoint should not have received message", 0, counter1.get()); + assertEquals("exactly one endpoint should have received message", 1, counter2.get() + counter3.get()); + } + + @Test + public void testBroadcastingDispatcherSkipsInactiveExecutor() throws InterruptedException { + final AtomicInteger counter1 = new AtomicInteger(); + 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); + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.setBroadcast(true); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1)); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { + @Override + public void start() { + } + }); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint3, 1, 1)); + dispatcher.dispatch(); + latch.await(100, TimeUnit.MILLISECONDS); + assertEquals("inactive endpoint should not have received message", 0, counter2.get()); + assertEquals("both active endpoints should have received message", 2, counter1.get() + counter3.get()); + } + + @Test + public void testDispatcherWithNoExecutors() { + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + assertEquals(0, dispatcher.dispatch()); + } + + @Test(expected=MessageDeliveryException.class) + public void testBroadcastingDispatcherReachesRejectionLimitAndShouldFail() { + final AtomicInteger counter1 = new AtomicInteger(); + 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); + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.setBroadcast(true); + dispatcher.setRejectionLimit(2); + dispatcher.setRetryInterval(3); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1)); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { + @Override + public void processMessage(Message message) { + throw new RejectedExecutionException(); + } + }); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint3, 1, 1)); + dispatcher.dispatch(); + } + + @Test + public void testBroadcastingDispatcherReachesRejectionLimitAndShouldNotFail() throws InterruptedException { + final AtomicInteger counter1 = new AtomicInteger(); + 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); + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.setBroadcast(true); + dispatcher.setRejectionLimit(2); + dispatcher.setRetryInterval(3); + dispatcher.setShouldFailOnRejectionLimit(false); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1)); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { + @Override + public void processMessage(Message message) { + throw new RejectedExecutionException(); + } + }); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint3, 1, 1)); + dispatcher.dispatch(); + latch.await(100, TimeUnit.MILLISECONDS); + 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()); + } + + @Test(expected=MessageDeliveryException.class) + public void testNonBroadcastingDispatcherReachesRejectionLimitAndShouldFail() { + 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); + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.setRejectionLimit(2); + dispatcher.setRetryInterval(3); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1) { + @Override + public void processMessage(Message message) { + throw new RejectedExecutionException(); + } + }); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { + @Override + public void processMessage(Message message) { + throw new RejectedExecutionException(); + } + }); + dispatcher.dispatch(); + } + + @Test + public void testNonBroadcastingDispatcherReachesRejectionLimitButShouldNotFail() { + final AtomicInteger counter1 = new AtomicInteger(); + final AtomicInteger counter2 = new AtomicInteger(); + final AtomicInteger rejectedCounter1 = new AtomicInteger(); + final AtomicInteger rejectedCounter2 = new AtomicInteger(); + final CountDownLatch latch = new CountDownLatch(1); + TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); + TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.setRejectionLimit(2); + dispatcher.setRetryInterval(3); + dispatcher.setShouldFailOnRejectionLimit(false); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1) { + @Override + public void processMessage(Message message) { + rejectedCounter1.incrementAndGet(); + throw new RejectedExecutionException(); + } + }); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { + @Override + public void processMessage(Message message) { + rejectedCounter2.incrementAndGet(); + throw new RejectedExecutionException(); + } + }); + dispatcher.dispatch(); + 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()); + } + + @Test + public void testNonBroadcastingDispatcherWithOneEndpointSucceeding() throws InterruptedException { + final AtomicInteger counter1 = new AtomicInteger(); + final AtomicInteger counter2 = new AtomicInteger(); + final AtomicInteger counter3 = new AtomicInteger(); + final AtomicInteger rejectedCounter1 = new AtomicInteger(); + final AtomicInteger rejectedCounter2 = new AtomicInteger(); + final AtomicInteger rejectedCounter3 = 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); + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + dispatcher.setRejectionLimit(2); + dispatcher.setRetryInterval(3); + dispatcher.setShouldFailOnRejectionLimit(false); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1) { + @Override + public void processMessage(Message message) { + rejectedCounter1.incrementAndGet(); + throw new RejectedExecutionException(); + } + }); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { + @Override + public void processMessage(Message message) { + if (rejectedCounter2.get() > 0) { + super.processMessage(message); + return; + } + rejectedCounter2.incrementAndGet(); + throw new RejectedExecutionException(); + } + }); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint3, 1, 1) { + @Override + public void processMessage(Message message) { + rejectedCounter3.incrementAndGet(); + throw new RejectedExecutionException(); + } + }); + dispatcher.dispatch(); + latch.await(100, TimeUnit.MILLISECONDS); + 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()); + } + + + private static class TestEndpoint extends GenericMessageEndpoint { + + private AtomicInteger counter; + + private CountDownLatch latch; + + + public TestEndpoint(AtomicInteger counter, CountDownLatch latch) { + this.counter = counter; + this.latch = latch; + } + + @Override + public void messageReceived(Message message) { + counter.incrementAndGet(); + latch.countDown(); + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java deleted file mode 100644 index c84b9df8a1..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java +++ /dev/null @@ -1,70 +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.bus; - -import static org.junit.Assert.assertTrue; - -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; - -import org.junit.Test; - -import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.endpoint.GenericMessageEndpoint; -import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.StringMessage; - -/** - * @author Mark Fisher - */ -public class UnicastMessageDispatcherTests { - - @Test - public void testDispatcherSendsToExactlyOneEndpoint() throws InterruptedException { - final AtomicBoolean endpoint1Received = new AtomicBoolean(); - final AtomicBoolean endpoint2Received = new AtomicBoolean(); - final CountDownLatch latch = new CountDownLatch(1); - MessageEndpoint endpoint1 = new GenericMessageEndpoint() { - @Override - public void messageReceived(Message message) { - endpoint1Received.set(true); - latch.countDown(); - } - }; - MessageEndpoint endpoint2 = new GenericMessageEndpoint() { - @Override - public void messageReceived(Message message) { - endpoint2Received.set(true); - latch.countDown(); - } - }; - ConsumerPolicy policy = new ConsumerPolicy(); - SimpleChannel channel = new SimpleChannel(); - channel.send(new StringMessage(1, "test")); - MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); - UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy); - dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1)); - dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1)); - dispatcher.dispatch(); - latch.await(100, TimeUnit.MILLISECONDS); - assertTrue("exactly one endpoint should have received message", - endpoint1Received.get() ^ endpoint2Received.get()); - } - -}