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); } }