diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java index 0fc001da89..c6a311c929 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java @@ -107,7 +107,7 @@ public abstract class AbstractTcpSendingMessageHandler extends byte[] bytes = mapper.fromMessage(message); SocketWriter writer = this.getWriter(); if (writer == null) { - throw new MessageMappingException("Failed to create SocketWriter"); + throw new MessageMappingException(message, "Failed to create SocketWriter"); } writer.write(bytes); } catch (Exception e) { @@ -115,7 +115,7 @@ public abstract class AbstractTcpSendingMessageHandler extends if (e instanceof MessageMappingException) { throw (MessageMappingException) e; } - throw new MessageMappingException("Failed to map message", e); + throw new MessageMappingException(message, "Failed to map message", e); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/core/MessagingException.java b/org.springframework.integration/src/main/java/org/springframework/integration/core/MessagingException.java index 01f973cb90..ead1bd36e5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/core/MessagingException.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/core/MessagingException.java @@ -20,11 +20,12 @@ package org.springframework.integration.core; * The base exception for any failures within the messaging system. * * @author Mark Fisher + * @author Gary Russell */ @SuppressWarnings("serial") public class MessagingException extends RuntimeException { - private final Message failedMessage; + private Message failedMessage; public MessagingException(Message message) { @@ -57,9 +58,12 @@ public class MessagingException extends RuntimeException { this.failedMessage = message; } - public Message getFailedMessage() { return this.failedMessage; } + + public void setFailedMessage(Message message) { + this.failedMessage = message; + } } 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 ca48a437f7..1b8493e735 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 @@ -22,6 +22,7 @@ import java.util.concurrent.Executor; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageHeaders; +import org.springframework.integration.core.MessagingException; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; @@ -41,6 +42,7 @@ import org.springframework.integration.message.MessageHandler; * * @author Mark Fisher * @author Iwein Fuld + * @author Gary Russell */ public class BroadcastingDispatcher extends AbstractDispatcher { @@ -121,6 +123,10 @@ public class BroadcastingDispatcher extends AbstractDispatcher { } catch (RuntimeException e) { if (!this.ignoreFailures) { + if (e instanceof MessagingException && + ((MessagingException) e).getFailedMessage() == null) { + ((MessagingException) e).setFailedMessage(message); + } throw e; } else if (this.logger.isWarnEnabled()) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java index 7720892300..ba0371f641 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java @@ -21,6 +21,7 @@ import java.util.List; import java.util.concurrent.Executor; import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessagingException; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageHandler; import org.springframework.util.Assert; @@ -41,6 +42,7 @@ import org.springframework.util.Assert; * * @author Iwein Fuld * @author Mark Fisher + * @author Gary Russell * @since 1.0.2 */ public class UnicastingDispatcher extends AbstractDispatcher { @@ -108,6 +110,10 @@ public class UnicastingDispatcher extends AbstractDispatcher { ? (RuntimeException) e : new MessageDeliveryException(message, "Dispatcher failed to deliver Message.", e); + if (e instanceof MessagingException && + ((MessagingException) e).getFailedMessage() == null) { + ((MessagingException) e).setFailedMessage(message); + } exceptions.add(runtimeException); this.handleExceptions(exceptions, message, !handlerIterator.hasNext()); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java index d628b3ac3b..9075886095 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java @@ -24,6 +24,7 @@ import static org.easymock.EasyMock.replay; import static org.easymock.EasyMock.reset; import static org.easymock.EasyMock.verify; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; import java.util.ArrayList; import java.util.Collections; @@ -35,12 +36,15 @@ import org.junit.Test; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessagingException; +import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher * @author Iwein Fuld + * @author Gary Russell */ public class BroadcastingDispatcherTests { @@ -277,6 +281,47 @@ public class BroadcastingDispatcherTests { assertEquals(originalId, messages.get(2).getHeaders().getCorrelationId()); } + /** + * Verifies that the dispatcher adds the message to the exception if it + * was not attached by the handler. + */ + @Test + public void testExceptionEnhancement() { + dispatcher = new BroadcastingDispatcher(); + dispatcher.addHandler(targetMock1); + targetMock1.handleMessage(messageMock); + expectLastCall().andThrow(new MessagingException("Mock Exception")); + replay(globalMocks); + try { + dispatcher.dispatch(messageMock); + fail("Expected Exception"); + } catch (MessagingException e) { + assertEquals(messageMock, e.getFailedMessage()); + } + verify(globalMocks); + } + + /** + * Verifies that the dispatcher does not add the message to the exception if it + * was attached by the handler. + */ + @Test + public void testNoExceptionEnhancement() { + dispatcher = new BroadcastingDispatcher(); + dispatcher.addHandler(targetMock1); + targetMock1.handleMessage(messageMock); + Message dontReplaceThisMessage = MessageBuilder.withPayload("x").build(); + expectLastCall().andThrow(new MessagingException(dontReplaceThisMessage, + "Mock Exception")); + replay(globalMocks); + try { + dispatcher.dispatch(messageMock); + fail("Expected Exception"); + } catch (MessagingException e) { + assertEquals(dontReplaceThisMessage, e.getFailedMessage()); + } + verify(globalMocks); + } private void defaultTaskExecutorMock() { taskExecutorMock.execute(isA(Runnable.class)); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java index 4f9bae9026..8787aa8037 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java @@ -19,9 +19,15 @@ import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.doThrow; import org.mockito.runners.MockitoJUnitRunner; import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessagingException; +import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; import java.util.concurrent.atomic.AtomicInteger; @@ -31,6 +37,7 @@ import static org.mockito.Mockito.*; /** * @author Iwein Fuld * @author Mark Fisher + * @author Gary Russell */ @RunWith(MockitoJUnitRunner.class) public class RoundRobinDispatcherTests { @@ -93,4 +100,38 @@ public class RoundRobinDispatcherTests { verify(differentHandler, atLeast(18)).handleMessage(message); } + /** + * Verifies that the dispatcher adds the message to the exception if it + * was not attached by the handler. + */ + @Test + public void testExceptionEnhancement() { + dispatcher.addHandler(handler); + doThrow(new MessagingException("Mock Exception")). + when(handler).handleMessage(message); + try { + dispatcher.dispatch(message); + fail("Expected Exception"); + } catch (MessagingException e) { + assertEquals(message, e.getFailedMessage()); + } + } + + /** + * Verifies that the dispatcher does not add the message to the exception if it + * was attached by the handler. + */ + @Test + public void testNoExceptionEnhancement() { + dispatcher.addHandler(handler); + Message dontReplaceThisMessage = MessageBuilder.withPayload("x").build(); + doThrow(new MessagingException(dontReplaceThisMessage, "Mock Exception")). + when(handler).handleMessage(message); + try { + dispatcher.dispatch(message); + fail("Expected Exception"); + } catch (MessagingException e) { + assertEquals(dontReplaceThisMessage, e.getFailedMessage()); + } + } }