diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java index 2574bb938d..659aaff0ea 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java @@ -16,7 +16,7 @@ package org.springframework.integration.channel; -import org.springframework.integration.dispatcher.AbstractSendOnceDispatcher; +import org.springframework.integration.dispatcher.AbstractWinningHandlerDispatcher; import org.springframework.integration.dispatcher.LoadBalancingDispatcher; /** @@ -27,12 +27,12 @@ import org.springframework.integration.dispatcher.LoadBalancingDispatcher; * @author Mark Fisher * @author Iwein Fuld */ -public class DirectChannel extends AbstractSubscribableChannel { +public class DirectChannel extends AbstractSubscribableChannel { public DirectChannel() { super(new LoadBalancingDispatcher()); } - public DirectChannel(AbstractSendOnceDispatcher dispatcher){ + public DirectChannel(AbstractWinningHandlerDispatcher dispatcher){ super(dispatcher); } 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 3facdfaeee..e990f4236a 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 @@ -16,8 +16,9 @@ package org.springframework.integration.dispatcher; -import java.util.Queue; -import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -37,7 +38,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher { protected final Log logger = LogFactory.getLog(this.getClass()); - private final Queue handlers = new ConcurrentLinkedQueue(); + private final List handlers = new ArrayList(); private volatile TaskExecutor taskExecutor; @@ -54,15 +55,15 @@ public abstract class AbstractDispatcher implements MessageDispatcher { return this.taskExecutor; } - protected Queue getHandlers() { - return handlers; + protected List getHandlers() { + return Collections.unmodifiableList(handlers); } public boolean addHandler(MessageHandler handler) { if (this.handlers.contains(handler)) { return false; } - return this.handlers.offer(handler); + return this.handlers.add(handler); } public boolean removeHandler(MessageHandler handler) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractSendOnceDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractSendOnceDispatcher.java deleted file mode 100644 index 5e12533e77..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractSendOnceDispatcher.java +++ /dev/null @@ -1,9 +0,0 @@ -package org.springframework.integration.dispatcher; - -import org.springframework.integration.core.Message; - -public abstract class AbstractSendOnceDispatcher extends AbstractDispatcher { - - public abstract boolean dispatch(Message message); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractWinningHandlerDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractWinningHandlerDispatcher.java new file mode 100644 index 0000000000..f49b1cc9e6 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractWinningHandlerDispatcher.java @@ -0,0 +1,61 @@ +package org.springframework.integration.dispatcher; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.integration.message.MessageHandler; +import org.springframework.integration.message.MessageRejectedException; + +/** + * Implementation of {@link MessageDispatcher} that will attempt to send a + * {@link Message} to at most one of its handlers. The handlers will be tried + * one by one. As soon as one of the handlers accepts the Message, the + * dispatcher will return true and ignore the rest of it's + * handlers. + *

+ * If the dispatcher has no handlers, a {@link MessageDeliveryException} will be + * thrown. If all handlers reject the Message, the dispatcher will throw a + * MessageRejectedException. + *

+ * The implementations of this class control the order in which the handlers + * will be tried through the getNextHandler method. + * + * @author Iwein Fuld + * + */ +public abstract class AbstractWinningHandlerDispatcher extends AbstractDispatcher { + + public final boolean dispatch(Message message) { + boolean success = false; + List triedHandlers = new ArrayList(); + int size; + List handlers = this.getHandlers(); + if (handlers.isEmpty()) { + throw new MessageDeliveryException(message, "Dispatcher has no subscribers."); + } + size = handlers.size(); + for (int i = 0; triedHandlers.size() < size && success == false; i++) { + MessageHandler handler = handlers.get(getNextHandlerIndex(size, i)); + if (this.sendMessageToHandler(message, handler)) { + success = true; //we have a winner. + } + triedHandlers.add(handler); + } + if (!success) { + throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message."); + } + return success; + } + + /** + * Return the next handler index. Subclasses have to implement this method + * to determine the order in which handlers will be invoked. + * + * @param size the total number of handlers + * @param loopIndex the current index of the loop + */ + protected abstract int getNextHandlerIndex(int size, int loopIndex); + +} 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 d248b8500a..9800fdd030 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,42 +16,22 @@ package org.springframework.integration.dispatcher; -import java.util.Iterator; - -import org.springframework.integration.core.Message; -import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.MessageHandler; -import org.springframework.integration.message.MessageRejectedException; - /** - * Basic implementation of {@link MessageDispatcher} that will attempt to send a - * {@link Message} to one of its handlers. As soon as one of the - * handlers accepts the Message, the dispatcher will return 'true'. - *

- * If the dispatcher has no handlers, a {@link MessageDeliveryException} will be - * thrown. If all handlers reject the Message, the dispatcher will throw a - * MessageRejectedException. + * {@link AbstractWinningHandlerDispatcher} that will try it's handlers in the + * same order every dispatch. * * @author Mark Fisher * @author Iwein Fuld */ -public class FailOverDispatcher extends AbstractSendOnceDispatcher { +public class FailOverDispatcher extends AbstractWinningHandlerDispatcher { - public boolean dispatch(Message message) { - if (this.getHandlers().size() == 0) { - throw new MessageDeliveryException(message, "Dispatcher has no subscribers."); - } - Iterator handlerIterator = this.getHandlers().iterator(); - boolean sent = false; - while (sent == false && handlerIterator.hasNext()) { - if (this.sendMessageToHandler(message, handlerIterator.next())) { - sent = true; - } - } - if (!sent) { - throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message."); - } - return sent; + /** + * Returns the current loop index, so each dispatch will try + * the handlers in the same order. + */ + @Override + protected int getNextHandlerIndex(int size, int loopIndex) { + return loopIndex; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/LoadBalancingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/LoadBalancingDispatcher.java index bf4918eec7..f232615a70 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/LoadBalancingDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/LoadBalancingDispatcher.java @@ -15,52 +15,26 @@ */ package org.springframework.integration.dispatcher; -import java.util.Queue; -import java.util.concurrent.locks.ReentrantLock; - -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 java.util.concurrent.atomic.AtomicInteger; /** - * Round-robin implementation of {@link MessageDispatcher} that will attempt to - * send a {@link Message} to one of its handlers. As soon as one of the - * handlers accepts the Message, the dispatcher will return 'true'. This - * implementation will load balance over its handlers using a round-robin - * strategy. - *

- * If the dispatcher has no handlers, a {@link MessageDeliveryException} will be - * thrown. If all handlers reject the Message, the dispatcher will throw a - * MessageRejectedException. + * Round-robin implementation of {@link AbstractWinningHandlerDispatcher}. This + * implementation will keep track of the index in which its handlers have been + * tried and return a different handler every dispatch. * * @author Iwein Fuld */ -public class LoadBalancingDispatcher extends AbstractSendOnceDispatcher { +public class LoadBalancingDispatcher extends AbstractWinningHandlerDispatcher { - private ReentrantLock queueLock = new ReentrantLock(); - - public boolean dispatch(Message message) { - queueLock.lock(); - Queue handlers = this.getHandlers(); - if (handlers.isEmpty()) { - throw new MessageDeliveryException(message, "Dispatcher has no subscribers."); - } - boolean success = false; - int size = handlers.size(); - queueLock.unlock(); - for (int i = 0; i < size && success == false; i++) { - queueLock.lock(); - MessageHandler handler = handlers.poll(); - handlers.offer(handler); - queueLock.unlock(); - if (this.sendMessageToHandler(message, handler)) { - success = true; - } - } - if (!success) { - throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message."); - } - return success; + private AtomicInteger currentHandlerIndex = new AtomicInteger(); + + /** + * Keeps track of the last index over multiple dispatch + * invocations. Each invocation of this method will increment the index by + * one, overflowing at size. loopIndex is ignored. + */ + protected int getNextHandlerIndex(int size, int loopIndex) { + int indexTail = currentHandlerIndex.getAndIncrement() % size; + return indexTail < 0 ? indexTail + size : indexTail; } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/LoadBalancingDispatcherConcurrentTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/LoadBalancingDispatcherConcurrentTests.java new file mode 100644 index 0000000000..92f8b53386 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/LoadBalancingDispatcherConcurrentTests.java @@ -0,0 +1,111 @@ +package org.springframework.integration.dispatcher; + +import static org.junit.Assert.*; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Before; +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.scheduling.concurrent.ThreadPoolTaskExecutor; + +@RunWith(MockitoJUnit44Runner.class) +public class LoadBalancingDispatcherConcurrentTests { + + private static final int TOTAL_EXECUTIONS = 40; + + private AbstractWinningHandlerDispatcher dispatcher = new LoadBalancingDispatcher(); + + private ThreadPoolTaskExecutor scheduler = new ThreadPoolTaskExecutor(); + + @Mock + private MessageHandler handler1; + + @Mock + private MessageHandler handler2; + + @Mock + private MessageHandler handler3; + + @Mock + private MessageHandler handler4; + + @Mock + private Message message; + + @Before + public void initialize() throws Exception { + scheduler.setCorePoolSize(10); + scheduler.setMaxPoolSize(10); + scheduler.initialize(); + } + + @Test(timeout = 1000) + public void noHandlerExhaustion() throws Exception { + dispatcher.addHandler(handler1); + dispatcher.addHandler(handler2); + dispatcher.addHandler(handler3); + dispatcher.addHandler(handler4); + final CountDownLatch start = new CountDownLatch(1); + final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS); + final Message message = this.message; + final AtomicBoolean failed = new AtomicBoolean(false); + Runnable messageSenderTask = new Runnable() { + public void run() { + try { + start.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + if (!dispatcher.dispatch(message)) { + failed.set(true); + } + allDone.countDown(); + } + }; + for (int i = 0; i < TOTAL_EXECUTIONS; i++) { + scheduler.execute(messageSenderTask); + } + start.countDown(); + allDone.await(); + assertFalse("not all messages were accepted", failed.get()); + } + + @Test(timeout=2000) + public void unlockOnFailure() throws Exception { + // dispatcher has no subscribers (shouldn't lead to deadlock) + final CountDownLatch start = new CountDownLatch(1); + final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS); + final Message message = this.message; + Runnable messageSenderTask = new Runnable() { + public void run() { + try { + start.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + try { + dispatcher.dispatch(message); + fail("this shouldn't happen"); + } + catch (MessageDeliveryException e) { + //expected + } + allDone.countDown(); + } + }; + for (int i = 0; i < TOTAL_EXECUTIONS; i++) { + scheduler.execute(messageSenderTask); + } + start.countDown(); + allDone.await(); + } +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/LoadBalancingDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/LoadBalancingDispatcherTests.java index 525bbae198..4d50f1d411 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/LoadBalancingDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/LoadBalancingDispatcherTests.java @@ -17,10 +17,13 @@ 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; @@ -32,7 +35,7 @@ import org.springframework.integration.message.MessageHandler; @RunWith(MockitoJUnit44Runner.class) public class LoadBalancingDispatcherTests { - private LoadBalancingDispatcher dispatcher = new LoadBalancingDispatcher(); + private AbstractWinningHandlerDispatcher dispatcher = new LoadBalancingDispatcher(); @Mock private MessageHandler handler; @@ -58,4 +61,16 @@ public class LoadBalancingDispatcherTests { verify(handler).handleMessage(message); verify(differentHandler).handleMessage(message); } + @Test + 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++){ + dispatcher.dispatch(message); + } + verify(handler, atLeast(18)).handleMessage(message); + verify(differentHandler, atLeast(18)).handleMessage(message); + } }