diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java index 37016da127..f08ed70b17 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java @@ -16,59 +16,23 @@ package org.springframework.integration.dispatcher; -import java.util.ArrayList; -import java.util.Iterator; - import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.MessageRejectedException; import org.springframework.integration.message.MessageTarget; -import org.springframework.util.Assert; /** * Basic implementation of {@link MessageDispatcher} that will attempt - * to send a {@link Message} to one of its targets (the first that accepts). + * to send a {@link Message} to one of its targets. As soon as one + * of the targets accepts the Message, the dispatcher will return 'true'. + *

+ * If all targets reject the Message, the dispatcher will throw a + * MessageRejectedException. If all targets return 'false' (e.g. due + * to a timeout), the dispatcher will return 'false'. * * @author Mark Fisher */ public class SimpleDispatcher extends AbstractDispatcher { - public final static int DEFAULT_REJECTION_LIMIT = 1; - - public final static long DEFAULT_RETRY_INTERVAL = 1000; - - - private volatile int rejectionLimit = DEFAULT_REJECTION_LIMIT; - - private volatile long retryInterval = DEFAULT_RETRY_INTERVAL; - - private volatile boolean shouldFailOnRejectionLimit = true; - - - /** - * 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; - } - - /** - * 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; - } - - /** - * 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 boolean send(Message message) { if (this.targets.size() == 0) { if (logger.isWarnEnabled()) { @@ -76,44 +40,13 @@ public class SimpleDispatcher extends AbstractDispatcher { } return false; } - int attempts = 0; - MessageHandlingException lastException = null; - while (attempts < this.rejectionLimit) { - Iterator iter = new ArrayList(this.targets).iterator(); - if (!iter.hasNext()) { - return false; - } - if (attempts > 0) { - try { - this.waitBetweenAttempts(attempts); - } - catch (InterruptedException iex) { - Thread.currentThread().interrupt(); - return false; - } - } - lastException = sendMessageToFirstAcceptingTarget(message, iter); - if (lastException == null) { - return true; - } - attempts++; - } - if (this.shouldFailOnRejectionLimit) { - throw lastException; - } - return false; - } - - private MessageHandlingException sendMessageToFirstAcceptingTarget(Message message, Iterator iter) { - MessageHandlingException exception = null; int count = 0; int rejectedExceptionCount = 0; - while (iter.hasNext()) { + for (MessageTarget target : this.targets) { count++; - MessageTarget target = iter.next(); try { if (this.sendMessageToTarget(message, target)) { - return null; + return true; } if (logger.isDebugEnabled()) { logger.debug("Failed to send message to target, continuing with other targets if available."); @@ -125,28 +58,11 @@ public class SimpleDispatcher extends AbstractDispatcher { logger.debug("Target '" + target + "' rejected Message, continuing with other targets if available.", e); } } - catch (MessageHandlingException e) { - if (exception == null) { - exception = e; - } - if (logger.isDebugEnabled()) { - logger.debug("Target '" + target + "' threw an exception, continuing with other targets if available.", e); - } - } } if (rejectedExceptionCount == count) { throw new MessageRejectedException(message, "All of dispatcher's targets rejected Message."); } - return exception; - } - - private void waitBetweenAttempts(int attempts) throws InterruptedException { - if (logger.isDebugEnabled()) { - logger.debug("target(s) unable to handle message after " + attempts + - " attempt(s), will try again after 'retryInterval' of " + - this.retryInterval + " milliseconds"); - } - Thread.sleep(this.retryInterval); + return false; } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java index 31ef17be1f..cafac78cf1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java @@ -30,7 +30,6 @@ import org.springframework.integration.endpoint.DefaultEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.MessageRejectedException; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.StringMessage; @@ -69,7 +68,7 @@ public class SimpleDispatcherTests { public void noDuplicateSubscriptions() { SimpleDispatcher dispatcher = new SimpleDispatcher(); final AtomicInteger counter = new AtomicInteger(); - MessageTarget target = new CountingFalseReturningTestTarget(counter); + MessageTarget target = new CountingTestTarget(counter, false); dispatcher.subscribe(target); dispatcher.subscribe(target); dispatcher.send(new StringMessage("test")); @@ -80,9 +79,9 @@ public class SimpleDispatcherTests { public void unsubscribeBeforeSend() { SimpleDispatcher dispatcher = new SimpleDispatcher(); final AtomicInteger counter = new AtomicInteger(); - MessageTarget target1 = new CountingFalseReturningTestTarget(counter); - MessageTarget target2 = new CountingFalseReturningTestTarget(counter); - MessageTarget target3 = new CountingFalseReturningTestTarget(counter); + MessageTarget target1 = new CountingTestTarget(counter, false); + MessageTarget target2 = new CountingTestTarget(counter, false); + MessageTarget target3 = new CountingTestTarget(counter, false); dispatcher.subscribe(target1); dispatcher.subscribe(target2); dispatcher.subscribe(target3); @@ -95,9 +94,9 @@ public class SimpleDispatcherTests { public void unsubscribeBetweenSends() { SimpleDispatcher dispatcher = new SimpleDispatcher(); final AtomicInteger counter = new AtomicInteger(); - MessageTarget target1 = new CountingFalseReturningTestTarget(counter); - MessageTarget target2 = new CountingFalseReturningTestTarget(counter); - MessageTarget target3 = new CountingFalseReturningTestTarget(counter); + MessageTarget target1 = new CountingTestTarget(counter, false); + MessageTarget target2 = new CountingTestTarget(counter, false); + MessageTarget target3 = new CountingTestTarget(counter, false); dispatcher.subscribe(target1); dispatcher.subscribe(target2); dispatcher.subscribe(target3); @@ -117,8 +116,6 @@ public class SimpleDispatcherTests { @Test public void handlersWithSelectorsAndOneAccepts() throws InterruptedException { SimpleDispatcher dispatcher = new SimpleDispatcher(); - dispatcher.setRejectionLimit(5); - dispatcher.setRetryInterval(5); final CountDownLatch latch = new CountDownLatch(1); final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); @@ -144,8 +141,6 @@ public class SimpleDispatcherTests { @Test public void handlersWithSelectorsAndNoneAccept() throws InterruptedException { SimpleDispatcher dispatcher = new SimpleDispatcher(); - dispatcher.setRejectionLimit(5); - dispatcher.setRetryInterval(5); final CountDownLatch latch = new CountDownLatch(2); final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); @@ -175,25 +170,45 @@ public class SimpleDispatcherTests { } @Test - public void handlersThrowingExceptionUntilRetried() throws InterruptedException { + public void firstHandlerReturnsTrue() { SimpleDispatcher dispatcher = new SimpleDispatcher(); - dispatcher.setRejectionLimit(5); - dispatcher.setRetryInterval(5); - final AtomicInteger handlerCounter = new AtomicInteger(); - TestMessageHandler handler1 = new TestMessageHandler(handlerCounter, 4); - TestMessageHandler handler2 = new TestMessageHandler(handlerCounter, 4); - TestMessageHandler handler3 = new TestMessageHandler(handlerCounter, 2); - DefaultEndpoint endpoint1 = new DefaultEndpoint(handler1); - DefaultEndpoint endpoint2 = new DefaultEndpoint(handler2); - DefaultEndpoint endpoint3 = new DefaultEndpoint(handler3); - dispatcher.subscribe(endpoint1); - dispatcher.subscribe(endpoint2); - dispatcher.subscribe(endpoint3); - dispatcher.send(new StringMessage("test")); - assertEquals("handlers should have been invoked 9 times in total", 9, handlerCounter.get()); - assertFalse("first handler should not have handled the message", handler1.handledMessage); - assertFalse("second handler should not have handled the message", handler2.handledMessage); - assertTrue("third handler should have handled the message", handler3.handledMessage); + final AtomicInteger counter = new AtomicInteger(); + MessageTarget target1 = new CountingTestTarget(counter, true); + MessageTarget target2 = new CountingTestTarget(counter, false); + MessageTarget target3 = new CountingTestTarget(counter, false); + dispatcher.subscribe(target1); + dispatcher.subscribe(target2); + dispatcher.subscribe(target3); + assertTrue(dispatcher.send(new StringMessage("test"))); + assertEquals("only the first target should have been invoked", 1, counter.get()); + } + + @Test + public void middleHandlerReturnsTrue() { + SimpleDispatcher dispatcher = new SimpleDispatcher(); + final AtomicInteger counter = new AtomicInteger(); + MessageTarget target1 = new CountingTestTarget(counter, false); + MessageTarget target2 = new CountingTestTarget(counter, true); + MessageTarget target3 = new CountingTestTarget(counter, false); + dispatcher.subscribe(target1); + dispatcher.subscribe(target2); + dispatcher.subscribe(target3); + assertTrue(dispatcher.send(new StringMessage("test"))); + assertEquals("first two targets should have been invoked", 2, counter.get()); + } + + @Test + public void allHandlersReturnFalse() { + SimpleDispatcher dispatcher = new SimpleDispatcher(); + final AtomicInteger counter = new AtomicInteger(); + MessageTarget target1 = new CountingTestTarget(counter, false); + MessageTarget target2 = new CountingTestTarget(counter, false); + MessageTarget target3 = new CountingTestTarget(counter, false); + dispatcher.subscribe(target1); + dispatcher.subscribe(target2); + dispatcher.subscribe(target3); + assertFalse(dispatcher.send(new StringMessage("test"))); + assertEquals("each target should have been invoked", 3, counter.get()); } @@ -220,53 +235,20 @@ public class SimpleDispatcherTests { } - private static class TestMessageHandler implements MessageHandler { - - private final AtomicInteger internalCounter = new AtomicInteger(); - - private final AtomicInteger sharedCounter; - - private final int timesToFail; - - private volatile boolean handledMessage = false; - - TestMessageHandler(AtomicInteger sharedCounter, int timesToReject) { - this.sharedCounter = sharedCounter; - this.timesToFail = timesToReject; - } - - public Message handle(Message message) { - int count = internalCounter.incrementAndGet(); - this.sharedCounter.incrementAndGet(); - if (this.timesToFail == 0) { - this.handledMessage = true; - return null; - } - if (this.timesToFail < 0) { - throw new MessageHandlingException(message, "intentional test failure"); - } - if (count > timesToFail) { - this.handledMessage = true; - return null; - } - else { - throw new MessageHandlingException(message, "intentional test failure"); - } - } - } - - - private static class CountingFalseReturningTestTarget implements MessageTarget { + private static class CountingTestTarget implements MessageTarget { private final AtomicInteger counter; - CountingFalseReturningTestTarget(AtomicInteger counter) { + private final boolean returnValue; + + CountingTestTarget(AtomicInteger counter, boolean returnValue) { this.counter = counter; + this.returnValue = returnValue; } public boolean send(Message message) { this.counter.incrementAndGet(); - return false; + return this.returnValue; } }