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;
}
}
-
}
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 4804499a69..0183526fca 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
@@ -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;
}
-
}
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
new file mode 100644
index 0000000000..f3b980100b
--- /dev/null
+++ b/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 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.
+ *
+ * @author Iwein Fuld
+ */
+public class LoadBalancingDispatcher extends AbstractDispatcher {
+
+ private final Queue handlerQueue = new ConcurrentLinkedQueue();
+
+ public boolean dispatch(Message message) {
+ Set handlers = new HashSet(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;
+ }
+
+}
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java
index 3b7e4691d5..57ee896762 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java
+++ b/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;
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
new file mode 100644
index 0000000000..525bbae198
--- /dev/null
+++ b/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);
+ }
+}