OPEN - issue INT-567: Add round-robin dispatching strategy

http://jira.springframework.org/browse/INT-567

tidying in AbstractDispatcher, added LoadBalancingDispatcher
first Mockito testcase

org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.dispatcher;
 
+import java.util.Collection;
+import java.util.Collections;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 
@@ -26,21 +28,22 @@ import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.core.Message;
 import org.springframework.integration.message.MessageHandler;
 import org.springframework.integration.message.MessageRejectedException;
+import org.springframework.util.Assert;
 
 /**
  * Base class for {@link MessageDispatcher} implementations.
  * 
  * @author Mark Fisher
+ * @author Iwein Fuld
  */
 public abstract class AbstractDispatcher implements MessageDispatcher {
 
 	protected final Log logger = LogFactory.getLog(this.getClass());
 
-	protected final Set<MessageHandler> handlers = new CopyOnWriteArraySet<MessageHandler>();
+	private final Set<MessageHandler> handlers = new CopyOnWriteArraySet<MessageHandler>();
 
 	private volatile TaskExecutor taskExecutor;
 
-
 	public boolean addHandler(MessageHandler handler) {
 		return this.handlers.add(handler);
 	}
@@ -62,6 +65,10 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
 		return this.taskExecutor;
 	}
 
+	protected Set<MessageHandler> getHandlers() {
+		return Collections.unmodifiableSet(handlers);
+	}
+	
 	public String toString() {
 		return this.getClass().getSimpleName() + " with handlers: " + this.handlers;
 	}
@@ -71,16 +78,17 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
 	 * "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, continuing with other handlers if available.", e);
+				logger.debug("Handler '" + handler + "' rejected Message, if other handlers are available this dispatcher may try to send to those.", e);
 			}
 			return false;
 		}
 	}
-
 }

org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java

@@ -31,6 +31,7 @@ import org.springframework.integration.message.MessageHandler;
  * to the other handlers.
  * 
  * @author Mark Fisher
+ * @author Iwein Fuld
  */
 public class BroadcastingDispatcher extends AbstractDispatcher {
 
@@ -48,8 +49,8 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
 
 	public boolean dispatch(Message<?> message) {
 		int sequenceNumber = 1;
-		int sequenceSize = this.handlers.size();
-		for (final MessageHandler handler : this.handlers) {
+		int sequenceSize = getHandlers().size();
+		for (final MessageHandler handler : getHandlers()) {
 			final Message<?> messageToSend = (!this.applySequence) ? message
 				: MessageBuilder.fromMessage(message)
 						.setSequenceNumber(sequenceNumber++)
@@ -70,5 +71,4 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
 		}
 		return true;
 	}
-
 }

org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/LoadBalancingDispatcher.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2002-2008 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.HashSet;
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.springframework.integration.core.Message;
+import org.springframework.integration.message.MessageDeliveryException;
+import org.springframework.integration.message.MessageHandler;
+import org.springframework.integration.message.MessageRejectedException;
+
+/**
+ * Round-robin implementation of {@link MessageDispatcher} that will attempt to
+ * send a {@link Message} to one of its handlers. As soon as <em>one</em> of the
+ * handlers accepts the Message, the dispatcher will return 'true'. This
+ * implementation will load balance over its handlers using a round-robin
+ * strategy.
+ * <p>
+ * If the dispatcher has no handlers, a {@link MessageDeliveryException} will be
+ * thrown. If all handlers reject the Message, the dispatcher will throw a
+ * MessageRejectedException.
+ * 
+ * @author Iwein Fuld
+ */
+public class LoadBalancingDispatcher extends AbstractDispatcher {
+
+	private final Queue<MessageHandler> handlerQueue = new ConcurrentLinkedQueue<MessageHandler>();
+	
+	public boolean dispatch(Message<?> message) {
+		Set<MessageHandler> handlers = new HashSet<MessageHandler>(this.getHandlers());
+		if (handlers.isEmpty()) {
+			throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
+		}
+		if (this.handlerQueue.isEmpty()){
+			handlerQueue.addAll(handlers);
+		}
+		boolean success = false;
+		while (!handlerQueue.isEmpty() && success == false) {
+			MessageHandler handler = handlerQueue.poll();
+			if (this.sendMessageToHandler(message, handler)) {
+				success = true;
+			}
+		}
+		if (!success) {
+			throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message.");
+		}
+		return success;
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java

@@ -31,16 +31,17 @@ import org.springframework.integration.message.MessageRejectedException;
  * throw a MessageRejectedException.
  * 
  * @author Mark Fisher
+ * @author Iwein Fuld
  */
 public class SimpleDispatcher extends AbstractDispatcher {
 
 	public boolean dispatch(Message<?> message) {
-		if (this.handlers.size() == 0) {
+		if (this.getHandlers().size() == 0) {
 			throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
 		}
 		int count = 0;
 		int rejectedExceptionCount = 0;
-		for (MessageHandler handler : this.handlers) {
+		for (MessageHandler handler : this.getHandlers()) {
 			count++;
 			if (this.sendMessageToHandler(message, handler)) {
 				return true;

org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/LoadBalancingDispatcherTests.java

@@ -0,0 +1,61 @@
+/* Copyright 2002-2008 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.mockito.Mockito.*;
+
+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.MessageHandler;
+
+/**
+ * 
+ * @author Iwein Fuld
+ * 
+ */
+@RunWith(MockitoJUnit44Runner.class)
+public class LoadBalancingDispatcherTests {
+
+	private LoadBalancingDispatcher dispatcher = new LoadBalancingDispatcher();
+
+	@Mock
+	private MessageHandler handler;
+
+	@Mock
+	private Message<?> message;
+
+	@Mock
+	private MessageHandler differentHandler;
+
+	@Test
+	public void dispatchMessageWithSingleHandler() throws Exception {
+		dispatcher.addHandler(handler);
+		dispatcher.dispatch(message);
+	}
+
+	@Test
+	public void differentHandlerInvokedOnSecondMessage() throws Exception {
+		dispatcher.addHandler(handler);
+		dispatcher.addHandler(differentHandler);
+		dispatcher.dispatch(message);
+		dispatcher.dispatch(message);
+		verify(handler).handleMessage(message);
+		verify(differentHandler).handleMessage(message);
+	}
+}