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

Refactored SimpleDispatcher and added test cases in SimpleDispatcherTests. Also removed MessageHandlerNotRunningException.

Browse Source
This commit is contained in:
Mark Fisher
2008-07-18 23:00:03 +00:00
parent ce88e75725
commit 0aa8aeb129
8 changed files with 251 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java
View File

@@ -18,17 +18,17 @@ package org.springframework.integration.dispatcher;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.Assert;
/**
* Basic implementation of {@link MessageDispatcher}.
* Basic implementation of {@link MessageDispatcher} that will attempt
* to send a {@link Message} to one of its targets (the first that accepts).
*
* @author Mark Fisher
*/
@@ -71,66 +71,81 @@ public class SimpleDispatcher extends AbstractDispatcher {
}
public boolean send(Message<?> message) {
if (this.targets.size() == 0) {
if (logger.isWarnEnabled()) {
logger.warn("Dispatcher has no targets.");
}
return false;
}
int attempts = 0;
List<MessageTarget> targetList = new ArrayList<MessageTarget>(this.targets);
MessageHandlingException lastException = null;
while (attempts < this.rejectionLimit) {
Iterator<MessageTarget> iter = new ArrayList<MessageTarget>(this.targets).iterator();
if (!iter.hasNext()) {
return false;
}
if (attempts > 0) {
if (logger.isDebugEnabled()) {
logger.debug("target(s) rejected message after " + attempts +
" attempt(s), will try again after 'retryInterval' of " +
this.retryInterval + " milliseconds");
}
try {
Thread.sleep(this.retryInterval);
this.waitBetweenAttempts(attempts);
}
catch (InterruptedException iex) {
Thread.currentThread().interrupt();
return false;
}
}
Iterator<MessageTarget> iter = targetList.iterator();
if (!iter.hasNext()) {
if (logger.isWarnEnabled()) {
logger.warn("no active targets");
}
return false;
}
boolean rejected = false;
while (iter.hasNext()) {
MessageTarget target = iter.next();
try {
if (this.sendMessageToTarget(message, target)) {
return true;
}
if (logger.isDebugEnabled()) {
logger.debug("target rejected message, continuing with other targets if available");
}
iter.remove();
}
catch (MessageHandlerNotRunningException e) {
if (logger.isDebugEnabled()) {
logger.debug("target is not running, continuing with other targets if available", e);
}
}
catch (MessageHandlerRejectedExecutionException e) {
rejected = true;
if (logger.isDebugEnabled()) {
logger.debug("target '" + target + "' is busy, continuing with other targets if available", e);
}
}
}
if (!rejected) {
lastException = sendMessageToFirstAcceptingTarget(message, iter);
if (lastException == null) {
return true;
}
attempts++;
}
if (this.shouldFailOnRejectionLimit) {
throw new MessageDeliveryException(message, "Dispatcher reached rejection limit of "
+ this.rejectionLimit
+ ". Consider increasing the target's concurrency and/or "
+ "the dispatcherPolicy's 'rejectionLimit'.");
throw new MessageDeliveryException(message, "Dispatcher reached rejection limit of " + this.rejectionLimit, lastException);
}
return false;
}
private MessageHandlingException sendMessageToFirstAcceptingTarget(Message<?> message, Iterator<MessageTarget> iter) {
MessageHandlingException lastException = null;
int count = 0;
int rejectedExceptionCount = 0;
while (iter.hasNext()) {
count++;
MessageTarget target = iter.next();
try {
if (this.sendMessageToTarget(message, target)) {
return null;
}
if (logger.isDebugEnabled()) {
logger.debug("Failed to send message to target, continuing with other targets if available.");
}
}
catch (MessageRejectedException e) {
rejectedExceptionCount++;
if (logger.isDebugEnabled()) {
logger.debug("Target '" + target + "' rejected Message, continuing with other targets if available.", e);
}
}
catch (MessageHandlingException e) {
lastException = e;
if (logger.isDebugEnabled()) {
logger.debug("Target '" + target + "' threw an exception, continuing with other targets if available.", e);
}
}
}
if (rejectedExceptionCount == count) {
throw new MessageRejectedException(message, "All of dispatcher's targets rejected Message.");
}
return lastException;
}
private void waitBetweenAttempts(int attempts) throws InterruptedException {
if (logger.isDebugEnabled()) {
logger.debug("target(s) unable to handle message after " + attempts +
" attempt(s), will try again after 'retryInterval' of " +
this.retryInterval + " milliseconds");
}
Thread.sleep(this.retryInterval);
}
}

5
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java
View File

@@ -30,14 +30,12 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.EndpointInterceptor;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
@@ -121,9 +119,6 @@ public class ConcurrencyInterceptor extends EndpointInterceptorAdapter
@Override
public boolean aroundSend(final Message<?> message, final MessageTarget endpoint) {
if (endpoint instanceof Lifecycle && !((Lifecycle) endpoint).isRunning()) {
throw new MessageHandlerNotRunningException(message);
}
try {
this.executor.execute(new Runnable() {
public void run() {

34
org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageHandlerNotRunningException.java
View File

@@ -1,34 +0,0 @@
/*
* 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.handler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
/**
* An exception indicating that a handler is not currently running.
*
* @author Mark Fisher
*/
@SuppressWarnings("serial")
public class MessageHandlerNotRunningException extends MessageHandlingException {
public MessageHandlerNotRunningException(Message<?> message) {
super(message, "handler is not running");
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryException.java
View File

@@ -32,4 +32,8 @@ public class MessageDeliveryException extends MessagingException {
super(undeliveredMessage, description);
}
public MessageDeliveryException(Message<?> undeliveredMessage, String description, Throwable cause) {
super(undeliveredMessage, description, cause);
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java
View File

@@ -109,8 +109,8 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
return null;
}
if (!type.isAssignableFrom(value.getClass())) {
throw new MessagingException("Type mismatch for header '" + key + "'. Expected ["
+ type + "] but actual type is [" + value.getClass() + "]");
throw new IllegalArgumentException("Incorrect type specified for header '" + key
+ "'. Expected [" + type + "] but actual type is [" + value.getClass() + "]");
}
return (T) value;
}

76
org.springframework.integration/src/main/java/org/springframework/integration/transformer/AnnotationMethodTransformerAdapter.java
View File

@@ -31,6 +31,7 @@ import org.springframework.integration.util.AbstractMethodInvokingAdapter;
/**
* @author Mark Fisher
*/
@SuppressWarnings("unchecked")
public class AnnotationMethodTransformerAdapter extends AbstractMethodInvokingAdapter implements MessageHandler {
private volatile MessageMapper mapper;
@@ -48,8 +49,7 @@ public class AnnotationMethodTransformerAdapter extends AbstractMethodInvokingAd
: new DefaultMessageMapper();
}
@SuppressWarnings("unchecked")
public Message<?> transform(Message<?> message) {
public Message<?> handle(Message<?> message) {
if (!this.isInitialized()) {
this.afterPropertiesSet();
}
@@ -66,48 +66,48 @@ public class AnnotationMethodTransformerAdapter extends AbstractMethodInvokingAd
else {
args = new Object[] { param };
}
Object result = null;
try {
result = this.invokeMethod(args);
}
catch (NoSuchMethodException e) {
result = this.invokeMethod(message);
this.methodExpectsMessage = true;
}
if (result == null) {
if (logger.isDebugEnabled()) {
logger.debug("MessageTransformer returned a null result");
}
return null;
}
if (result instanceof Properties && !(message.getPayload() instanceof Properties)) {
Properties propertiesToSet = (Properties) result;
MessageBuilder builder = MessageBuilder.fromMessage(message);
for (Object keyObject : propertiesToSet.keySet()) {
String key = (String) keyObject;
builder.setHeader(key, propertiesToSet.getProperty(key));
}
return builder.build();
}
else if (result instanceof Map && !(message.getPayload() instanceof Map)) {
Map<String, ?> attributesToSet = (Map) result;
MessageBuilder builder = MessageBuilder.fromMessage(message);
for (String key : attributesToSet.keySet()) {
builder.setHeader(key, attributesToSet.get(key));
}
return builder.build();
}
else {
return MessageBuilder.fromPayload(result).copyHeaders(message.getHeaders()).build();
}
return this.invokeMethodAndReturnMessage(message, args);
}
catch (Exception e) {
throw new MessagingException(message, "failed to transform message payload", e);
}
}
public Message<?> handle(Message<?> message) {
return this.transform(message);
private Message<?> invokeMethodAndReturnMessage(Message<?> message, Object[] args) throws Exception {
Object result = null;
try {
result = this.invokeMethod(args);
}
catch (NoSuchMethodException e) {
result = this.invokeMethod(message);
this.methodExpectsMessage = true;
}
if (result == null) {
if (logger.isDebugEnabled()) {
logger.debug("handler invocation returned a null result");
}
return null;
}
if (result instanceof Properties && !(message.getPayload() instanceof Properties)) {
Properties propertiesToSet = (Properties) result;
MessageBuilder builder = MessageBuilder.fromMessage(message);
for (Object keyObject : propertiesToSet.keySet()) {
String key = (String) keyObject;
builder.setHeader(key, propertiesToSet.getProperty(key));
}
return builder.build();
}
else if (result instanceof Map && !(message.getPayload() instanceof Map)) {
Map<String, ?> attributesToSet = (Map) result;
MessageBuilder builder = MessageBuilder.fromMessage(message);
for (String key : attributesToSet.keySet()) {
builder.setHeader(key, attributesToSet.get(key));
}
return builder.build();
}
else {
return MessageBuilder.fromPayload(result).copyHeaders(message.getHeaders()).build();
}
}
}

146
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java
View File

@@ -17,18 +17,22 @@
package org.springframework.integration.dispatcher;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.TestHandlers;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.selector.MessageSelector;
/**
* @author Mark Fisher
@@ -59,9 +63,145 @@ public class SimpleDispatcherTests {
assertEquals("only 1 handler should have received the message", 1, counter1.get() + counter2.get());
}
@Test
public void testHandlersWithSelectorsAndOneAccepts() throws InterruptedException {
SimpleDispatcher dispatcher = new SimpleDispatcher();
dispatcher.setRejectionLimit(5);
dispatcher.setRetryInterval(5);
final CountDownLatch latch = new CountDownLatch(1);
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final AtomicInteger selectorCounter = new AtomicInteger();
HandlerEndpoint endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch));
HandlerEndpoint endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch));
HandlerEndpoint endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch));
endpoint1.setMessageSelector(new TestMessageSelector(selectorCounter, false));
endpoint2.setMessageSelector(new TestMessageSelector(selectorCounter, false));
endpoint3.setMessageSelector(new TestMessageSelector(selectorCounter, true));
dispatcher.addTarget(endpoint1);
dispatcher.addTarget(endpoint2);
dispatcher.addTarget(endpoint3);
dispatcher.send(new StringMessage("test"));
assertEquals(0, latch.getCount());
assertEquals("selectors should have been invoked one time each", 3, selectorCounter.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter1.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter2.get());
assertEquals("handler with accepting selector should have received the message", 1, counter3.get());
}
private static MessageTarget createEndpoint(MessageHandler handler) {
@Test()
public void testHandlersWithSelectorsAndNoneAccept() throws InterruptedException {
SimpleDispatcher dispatcher = new SimpleDispatcher();
dispatcher.setRejectionLimit(5);
dispatcher.setRetryInterval(5);
final CountDownLatch latch = new CountDownLatch(2);
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final AtomicInteger selectorCounter = new AtomicInteger();
HandlerEndpoint endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch));
HandlerEndpoint endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch));
HandlerEndpoint endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch));
endpoint1.setMessageSelector(new TestMessageSelector(selectorCounter, false));
endpoint2.setMessageSelector(new TestMessageSelector(selectorCounter, false));
endpoint3.setMessageSelector(new TestMessageSelector(selectorCounter, false));
dispatcher.addTarget(endpoint1);
dispatcher.addTarget(endpoint2);
dispatcher.addTarget(endpoint3);
boolean exceptionThrown = false;
try {
dispatcher.send(new StringMessage("test"));
}
catch (MessageRejectedException e) {
exceptionThrown = true;
}
assertTrue(exceptionThrown);
assertEquals("selectors should have been invoked one time each", 3, selectorCounter.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter1.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter2.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter3.get());
}
@Test
public void testHandlersThrowingExceptionUntilRetried() throws InterruptedException {
SimpleDispatcher dispatcher = new SimpleDispatcher();
dispatcher.setRejectionLimit(5);
dispatcher.setRetryInterval(5);
final AtomicInteger handlerCounter = new AtomicInteger();
TestMessageHandler handler1 = new TestMessageHandler(handlerCounter, 4);
TestMessageHandler handler2 = new TestMessageHandler(handlerCounter, 4);
TestMessageHandler handler3 = new TestMessageHandler(handlerCounter, 2);
HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1);
HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2);
HandlerEndpoint endpoint3 = new HandlerEndpoint(handler3);
dispatcher.addTarget(endpoint1);
dispatcher.addTarget(endpoint2);
dispatcher.addTarget(endpoint3);
dispatcher.send(new StringMessage("test"));
assertEquals("handlers should have been invoked 9 times in total", 9, handlerCounter.get());
assertFalse("first handler should not have handled the message", handler1.handledMessage);
assertFalse("second handler should not have handled the message", handler2.handledMessage);
assertTrue("third handler should have handled the message", handler3.handledMessage);
}
private static HandlerEndpoint createEndpoint(MessageHandler handler) {
return new HandlerEndpoint(handler);
}
private static class TestMessageSelector implements MessageSelector {
private final AtomicInteger counter;
private final boolean accept;
TestMessageSelector(AtomicInteger counter, boolean accept) {
this.counter = counter;
this.accept = accept;
}
public boolean accept(Message<?> message) {
this.counter.incrementAndGet();
return this.accept;
}
}
private static class TestMessageHandler implements MessageHandler {
private final AtomicInteger internalCounter = new AtomicInteger();
private final AtomicInteger sharedCounter;
private final int timesToFail;
private volatile boolean handledMessage = false;
TestMessageHandler(AtomicInteger sharedCounter, int timesToReject) {
this.sharedCounter = sharedCounter;
this.timesToFail = timesToReject;
}
public Message<?> handle(Message<?> message) {
int count = internalCounter.incrementAndGet();
this.sharedCounter.incrementAndGet();
if (this.timesToFail == 0) {
this.handledMessage = true;
return null;
}
if (this.timesToFail < 0) {
throw new MessageHandlingException(message, "intentional test failure");
}
if (count > timesToFail) {
this.handledMessage = true;
return null;
}
else {
throw new MessageHandlingException(message, "intentional test failure");
}
}
}
}

4
org.springframework.integration/src/test/java/org/springframework/integration/message/MessageHeadersTests.java
View File

@@ -56,8 +56,8 @@ public class MessageHeadersTests {
assertEquals(value, headers.get("test", Integer.class));
}
@Test(expected = MessagingException.class)
public void testTypeMismatchWhenAccessingHeaderValue() {
@Test(expected = IllegalArgumentException.class)
public void testHeaderValueAccessWithIncorrectType() {
Integer value = new Integer(123);
Map<String, Object> map = new HashMap<String, Object>();
map.put("test", value);