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 f603105df0..3579444930 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 @@ -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 targetList = new ArrayList(this.targets); + MessageHandlingException lastException = null; while (attempts < this.rejectionLimit) { + Iterator iter = new ArrayList(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 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 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); + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java index 691cca53c5..9fd63503ab 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java @@ -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() { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageHandlerNotRunningException.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageHandlerNotRunningException.java deleted file mode 100644 index e841304ba6..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageHandlerNotRunningException.java +++ /dev/null @@ -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"); - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryException.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryException.java index f92d2bef34..07954ddf7e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryException.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryException.java @@ -32,4 +32,8 @@ public class MessageDeliveryException extends MessagingException { super(undeliveredMessage, description); } + public MessageDeliveryException(Message undeliveredMessage, String description, Throwable cause) { + super(undeliveredMessage, description, cause); + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java index 2ad01a9cb7..8b5d4817c3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java @@ -109,8 +109,8 @@ public final class MessageHeaders implements Map, 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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/transformer/AnnotationMethodTransformerAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/transformer/AnnotationMethodTransformerAdapter.java index 5685f4edcc..bafc6b5b69 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/transformer/AnnotationMethodTransformerAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/transformer/AnnotationMethodTransformerAdapter.java @@ -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 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 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(); + } } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java index 1d43cd7956..4a1a1cb84b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java @@ -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"); + } + } + } + } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageHeadersTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageHeadersTests.java index 4f2fb6b4e3..1e3d4647cd 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageHeadersTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageHeadersTests.java @@ -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 map = new HashMap(); map.put("test", value);