From fd373882dede46c9d7f6153c3c49a60c7a49edf0 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 10 Jul 2009 16:23:19 +0000 Subject: [PATCH] INT-711 Added failover flag for RoundRobinDispatcher and modified Exception handling to accept all RuntimeExceptions rather than only MessageRejectedExceptions. Dispatchers with failover enabled will now throw an AggregateMesssageDeliveryException if all handlers threw Exceptions. --- .../dispatcher/AbstractDispatcher.java | 33 ++----------- .../dispatcher/AbstractUnicastDispatcher.java | 41 +++++++++++----- .../AggregateMessageDeliverException.java | 49 +++++++++++++++++++ .../dispatcher/BroadcastingDispatcher.java | 4 +- .../dispatcher/FailOverDispatcher.java | 15 ++++-- .../dispatcher/RoundRobinDispatcher.java | 28 ++++++++++- .../RoundRobinDispatcherConcurrentTests.java | 34 +++++++++++-- .../dispatcher/RoundRobinDispatcherTests.java | 17 +++---- 8 files changed, 157 insertions(+), 64 deletions(-) create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliverException.java diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index 1decc9f843..eee5322726 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -24,10 +24,7 @@ import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageHandler; -import org.springframework.integration.message.MessageRejectedException; -import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** @@ -50,9 +47,11 @@ public abstract class AbstractDispatcher implements MessageDispatcher { private final Set handlers = new OrderedAwareLinkedHashSet(); - private final Object handlerListMonitor = new Object(); - + /** + * Returns a copied, unmodifiable List of this dispatcher's handlers. + * This is provided for access by subclasses. + */ @SuppressWarnings("unchecked") protected List getHandlers() { return Collections.unmodifiableList(new ArrayList(this.handlers)); @@ -63,9 +62,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher { } public boolean removeHandler(MessageHandler handler) { - synchronized (this.handlerListMonitor) { - return this.handlers.remove(handler); - } + return this.handlers.remove(handler); } public String toString() { @@ -73,24 +70,4 @@ public abstract class AbstractDispatcher implements MessageDispatcher { return this.getClass().getSimpleName() + " with handlers: " + handlerList; } - /** - * Convenience method available for subclasses. Returns 'true' unless a - * "Selective Consumer" throws a {@link MessageRejectedException}. - */ - protected boolean sendMessageToHandler(Message message, MessageHandler handler) { - Assert.notNull(message, "'message' must not be null."); - Assert.notNull(handler, "'handler' must not be null."); - try { - handler.handleMessage(message); - return true; - } - catch (MessageRejectedException e) { - if (logger.isDebugEnabled()) { - logger.debug("Handler '" + handler + "' rejected Message, " - + "if other handlers are available this dispatcher may try to send to those.", e); - } - return false; - } - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractUnicastDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractUnicastDispatcher.java index 9cc6e14d03..e637863098 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractUnicastDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractUnicastDispatcher.java @@ -15,6 +15,7 @@ package org.springframework.integration.dispatcher; +import java.util.ArrayList; import java.util.Iterator; import java.util.List; @@ -44,29 +45,47 @@ public abstract class AbstractUnicastDispatcher extends AbstractDispatcher { public final boolean dispatch(Message message) { boolean success = false; - List handlers = this.getHandlers(); - if (handlers.isEmpty()) { + Iterator handlerIterator = this.getHandlerIterator(); + if (!handlerIterator.hasNext()) { throw new MessageDeliveryException(message, "Dispatcher has no subscribers."); } - Iterator handlerIterator = getHandlerIterator(handlers); + List exceptions = new ArrayList(); while (success == false && handlerIterator.hasNext()) { MessageHandler handler = handlerIterator.next(); - if (this.sendMessageToHandler(message, handler)) { + try { + handler.handleMessage(message); success = true; // we have a winner. } - } - if (!success) { - throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message."); + catch (Exception e) { + RuntimeException runtimeException = (e instanceof RuntimeException) + ? (RuntimeException) e + : new MessageDeliveryException(message, + "Dispatcher failed to deliver Message.", e); + exceptions.add(runtimeException); + this.handleExceptions(exceptions, message, !handlerIterator.hasNext()); + } } return success; } /** * Return the iterator that will be used to loop over the handlers. This - * allows subclasses to control the order of iteration for each - * {@link #dispatch(Message)} invocation. - * @param handlers all handlers for this dispatcher + * default simply returns the Iterator for the existing handler List. This + * method can be overridden by subclasses to control the order of iteration + * for each {@link #dispatch(Message)} invocation. */ - protected abstract Iterator getHandlerIterator(List handlers); + protected Iterator getHandlerIterator() { + return this.getHandlers().iterator(); + } + + /** + * Subclasses must implement this method to handle Exceptions that occur + * while dispatching. The list will never be null, will always have a size + * of at least one, and it will be in order with the most recent Exception + * at the end. The 'isLast' flag will be true if the Exception + * occurred during the final iteration of the MessageHandlers. + */ + protected abstract void handleExceptions( + List allExceptions, Message message, boolean isLast); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliverException.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliverException.java new file mode 100644 index 0000000000..21e5dc35d6 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliverException.java @@ -0,0 +1,49 @@ +/* + * Copyright 2002-2009 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.dispatcher; + +import java.util.List; + +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageDeliveryException; + +/** + * An Exception that encapsulates an aggregated group of Exceptions for use by + * dispatchers that may try multiple handler invocations within a single dispatch + * operation. + * + * @author Mark Fisher + * @since 1.0.3 + */ +@SuppressWarnings("serial") +public class AggregateMessageDeliverException extends MessageDeliveryException { + + private final List aggregatedExceptions; + + + public AggregateMessageDeliverException(Message undeliveredMessage, + String description, List aggregatedExceptions) { + super(undeliveredMessage, description); + this.aggregatedExceptions = aggregatedExceptions; + } + + + public List getAggregatedExceptions() { + return this.aggregatedExceptions; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java index 7c1a931ba7..c129f470cb 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java @@ -75,12 +75,12 @@ public class BroadcastingDispatcher extends AbstractDispatcher { if (this.taskExecutor != null) { this.taskExecutor.execute(new Runnable() { public void run() { - BroadcastingDispatcher.this.sendMessageToHandler(messageToSend, handler); + handler.handleMessage(messageToSend); } }); } else { - this.sendMessageToHandler(messageToSend, handler); + handler.handleMessage(messageToSend); } } return true; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java index 0c4b63dce3..ac7c4ae746 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java @@ -16,10 +16,9 @@ package org.springframework.integration.dispatcher; -import java.util.Iterator; import java.util.List; -import org.springframework.integration.message.MessageHandler; +import org.springframework.integration.core.Message; /** * {@link AbstractUnicastDispatcher} that will try its handlers in the @@ -30,10 +29,16 @@ import org.springframework.integration.message.MessageHandler; */ public class FailOverDispatcher extends AbstractUnicastDispatcher { - @Override - protected Iterator getHandlerIterator(List handlers) { - return handlers.iterator(); + protected void handleExceptions(List allExceptions, + Message message, boolean isLast) { + if (isLast) { + if (allExceptions != null && allExceptions.size() == 1) { + throw allExceptions.get(0); + } + throw new AggregateMessageDeliverException(message, + "All attempts to deliver Message to MessageHandlers failed.", allExceptions); + } } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinDispatcher.java index f376eaa3d1..f5b8f58196 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinDispatcher.java @@ -21,6 +21,7 @@ import java.util.Iterator; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; +import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageHandler; /** @@ -32,7 +33,14 @@ import org.springframework.integration.message.MessageHandler; */ public class RoundRobinDispatcher extends AbstractUnicastDispatcher { - private AtomicInteger currentHandlerIndex = new AtomicInteger(); + private volatile boolean failover = true; + + private final AtomicInteger currentHandlerIndex = new AtomicInteger(); + + + public void setFailover(boolean failover) { + this.failover = failover; + } /** * Returns an iterator that starts at a new point in the list every time the @@ -41,8 +49,12 @@ public class RoundRobinDispatcher extends AbstractUnicastDispatcher { * next() invocations. */ @Override - protected Iterator getHandlerIterator(List handlers) { + protected Iterator getHandlerIterator() { + List handlers = this.getHandlers(); int size = handlers.size(); + if (size == 0) { + return handlers.iterator(); + } int nextHandlerStartIndex = getNextHandlerStartIndex(size); List reorderedHandlers = new ArrayList( handlers.subList(nextHandlerStartIndex, size)); @@ -60,4 +72,16 @@ public class RoundRobinDispatcher extends AbstractUnicastDispatcher { return indexTail < 0 ? indexTail + size : indexTail; } + @Override + protected void handleExceptions(List allExceptions, + Message message, boolean isLast) { + if (isLast || !this.failover) { + if (allExceptions != null && allExceptions.size() == 1) { + throw allExceptions.get(0); + } + throw new AggregateMessageDeliverException(message, + "Failed to deliver Message to any MessageHandler.", allExceptions); + } + } + } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java index 3a93e2d76e..820d0380e7 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java @@ -1,9 +1,28 @@ +/* Copyright 2002-2009 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.dispatcher; -import static org.junit.Assert.*; -import static org.mockito.Mockito.*; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Before; @@ -11,18 +30,23 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.runners.MockitoJUnit44Runner; + import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageRejectedException; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +/** + * @author Iwein Fuld + */ @RunWith(MockitoJUnit44Runner.class) public class RoundRobinDispatcherConcurrentTests { private static final int TOTAL_EXECUTIONS = 40; - private AbstractUnicastDispatcher dispatcher = new RoundRobinDispatcher(); + private RoundRobinDispatcher dispatcher = new RoundRobinDispatcher(); private ThreadPoolTaskExecutor scheduler = new ThreadPoolTaskExecutor(); @@ -116,7 +140,7 @@ public class RoundRobinDispatcherConcurrentTests { } @Test - public void noHandlerSkipUnderConcurrentFailure() throws Exception { + public void noHandlerSkipUnderConcurrentFailureWithFailover() throws Exception { dispatcher.addHandler(handler1); dispatcher.addHandler(handler2); doThrow(new MessageRejectedException(message)).when(handler1).handleMessage(message); @@ -144,7 +168,7 @@ public class RoundRobinDispatcherConcurrentTests { scheduler.execute(messageSenderTask); } start.countDown(); - allDone.await(); + allDone.await(5000, TimeUnit.MILLISECONDS); assertFalse("not all messages were accepted", failed.get()); verify(handler1, times(TOTAL_EXECUTIONS/2)).handleMessage(message); verify(handler2, times(TOTAL_EXECUTIONS)).handleMessage(message); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java index 633c803e96..059878a837 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java @@ -1,4 +1,4 @@ -/* Copyright 2002-2008 the original author or authors. +/* Copyright 2002-2009 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. @@ -17,20 +17,17 @@ package org.springframework.integration.dispatcher; import static org.mockito.Mockito.*; -import java.util.concurrent.atomic.AtomicInteger; - import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.runners.MockitoJUnit44Runner; -import org.springframework.beans.DirectFieldAccessor; + import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageHandler; /** - * * @author Iwein Fuld - * + * @author Mark Fisher */ @RunWith(MockitoJUnit44Runner.class) public class RoundRobinDispatcherTests { @@ -65,12 +62,10 @@ public class RoundRobinDispatcherTests { public void overFlowCurrentHandlerIndex() throws Exception { dispatcher.addHandler(handler); dispatcher.addHandler(differentHandler); - DirectFieldAccessor accessor = new DirectFieldAccessor(dispatcher); - ((AtomicInteger)accessor.getPropertyValue("currentHandlerIndex")).set(Integer.MAX_VALUE-5); - for(long i=0; i < 40; i++){ + for (int i = 0; i < 7; i++) { dispatcher.dispatch(message); } - verify(handler, atLeast(18)).handleMessage(message); - verify(differentHandler, atLeast(18)).handleMessage(message); + verify(handler, times(4)).handleMessage(message); + verify(differentHandler, times(3)).handleMessage(message); } }