From 0ec6cd61d9521456cf696127f306813bbe874f00 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 18 Aug 2008 19:36:04 +0000 Subject: [PATCH] SimpleDispatcher no longer continues to retry sending any time it catches an Exception. Instead, it will try to send to each of its targets until one of them accepts the message (returns true). If a target returns false (e.g. for a timeout) or throws a MessageRejectedException (e.g. a MessageSelector did not accept the Message), then it will continue trying its other targets. However, any other Exceptions will be re-thrown immediately. Also, it no longer attempts to send to its targets more than once. Thus, the 'rejectionLimit' and 'retryInterval' properties have been removed. The retry behavior actually belongs in an interceptor/template on a per-MessageTarget basis. Some targets are retry-able (e.g. might throw a RemoteAccessException), but others are not. Therefore, the SimpleDispatcher should not have this awareness; such configuration options belong with the individual target instead (part of INT-337). --- .../dispatcher/SimpleDispatcher.java | 102 ++------------- .../dispatcher/SimpleDispatcherTests.java | 120 ++++++++---------- 2 files changed, 60 insertions(+), 162 deletions(-) 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; } }