Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.

Browse Source
This commit is contained in:
Mark Fisher
2009-07-10 16:23:19 +00:00
parent 8bd176c58a
commit fd373882de
8 changed files with 157 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java
View File

@@ -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<MessageHandler> handlers = new OrderedAwareLinkedHashSet<MessageHandler>();
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<MessageHandler> 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;
}
}
}

41
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractUnicastDispatcher.java
View File

@@ -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<MessageHandler> handlers = this.getHandlers();
if (handlers.isEmpty()) {
Iterator<MessageHandler> handlerIterator = this.getHandlerIterator();
if (!handlerIterator.hasNext()) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
}
Iterator<MessageHandler> handlerIterator = getHandlerIterator(handlers);
List<RuntimeException> exceptions = new ArrayList<RuntimeException>();
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<MessageHandler> getHandlerIterator(List<MessageHandler> handlers);
protected Iterator<MessageHandler> 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 <code>true</code> if the Exception
* occurred during the final iteration of the MessageHandlers.
*/
protected abstract void handleExceptions(
List<RuntimeException> allExceptions, Message<?> message, boolean isLast);
}

49
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliverException.java Normal file
View File

@@ -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<? extends Exception> aggregatedExceptions;
public AggregateMessageDeliverException(Message<?> undeliveredMessage,
String description, List<? extends Exception> aggregatedExceptions) {
super(undeliveredMessage, description);
this.aggregatedExceptions = aggregatedExceptions;
}
public List<? extends Exception> getAggregatedExceptions() {
return this.aggregatedExceptions;
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java
View File

@@ -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;

15
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java
View File

@@ -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<MessageHandler> getHandlerIterator(List<MessageHandler> handlers) {
return handlers.iterator();
protected void handleExceptions(List<RuntimeException> 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);
}
}
}

28
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinDispatcher.java
View File

@@ -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 {
* <code>next()</code> invocations.
*/
@Override
protected Iterator<MessageHandler> getHandlerIterator(List<MessageHandler> handlers) {
protected Iterator<MessageHandler> getHandlerIterator() {
List<MessageHandler> handlers = this.getHandlers();
int size = handlers.size();
if (size == 0) {
return handlers.iterator();
}
int nextHandlerStartIndex = getNextHandlerStartIndex(size);
List<MessageHandler> reorderedHandlers = new ArrayList<MessageHandler>(
handlers.subList(nextHandlerStartIndex, size));
@@ -60,4 +72,16 @@ public class RoundRobinDispatcher extends AbstractUnicastDispatcher {
return indexTail < 0 ? indexTail + size : indexTail;
}
@Override
protected void handleExceptions(List<RuntimeException> 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);
}
}
}

34
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java
View File

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

17
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java
View File

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