diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java index cd8473f25c..611bfe457a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/MessageProducerSupportTests.java @@ -16,70 +16,111 @@ package org.springframework.integration.endpoint; -import org.junit.Ignore; +import static org.junit.Assert.assertEquals; + import org.junit.Test; import org.springframework.integration.Message; import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessagingException; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.handler.ServiceActivatingHandler; +import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.test.util.TestUtils; /** * @author Oleg Zhurakousky - * + * @author Mark Fisher + * @since 2.0.1 */ public class MessageProducerSupportTests { - + @Test(expected=MessageDeliveryException.class) - public void validateExceptionIfNoErrorChannel(){ + public void validateExceptionIfNoErrorChannel() { DirectChannel outChannel = new DirectChannel(); - outChannel.subscribe(new MessageHandler() { + outChannel.subscribe(new MessageHandler() { public void handleMessage(Message message) throws MessagingException { throw new RuntimeException("problems"); } }); MessageProducerSupport mps = new MessageProducerSupport() {}; - mps.setOutputChannel(outChannel); mps.setBeanFactory(TestUtils.createTestApplicationContext()); mps.afterPropertiesSet(); mps.start(); - - mps.sendMessage(new GenericMessage("hello")); + mps.sendMessage(new GenericMessage("hello")); } - - @Test(expected=MessageDeliveryException.class) - @Ignore - public void validateExceptionIfSendToErrorChannelFails(){ + + @Test(expected=MessageHandlingException.class) + public void validateExceptionIfSendToErrorChannelFails() { DirectChannel outChannel = new DirectChannel(); - outChannel.subscribe(new MessageHandler() { + outChannel.subscribe(new MessageHandler() { public void handleMessage(Message message) throws MessagingException { throw new RuntimeException("problems"); } }); PublishSubscribeChannel errorChannel = new PublishSubscribeChannel(); - ServiceActivatingHandler handler = new ServiceActivatingHandler(new MyOneWayErrorService()); + ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingErrorService()); handler.afterPropertiesSet(); errorChannel.subscribe(handler); MessageProducerSupport mps = new MessageProducerSupport() {}; - mps.setOutputChannel(outChannel); mps.setErrorChannel(errorChannel); mps.setBeanFactory(TestUtils.createTestApplicationContext()); mps.afterPropertiesSet(); mps.start(); - - mps.sendMessage(new GenericMessage("hello")); + mps.sendMessage(new GenericMessage("hello")); } - - public static class MyOneWayErrorService { - public void handleErrorMessage(Message errorMessage){ + + @Test + public void validateSuccessfulErrorFlowDoesNotThrowErrors() { + DirectChannel outChannel = new DirectChannel(); + outChannel.subscribe(new MessageHandler() { + public void handleMessage(Message message) throws MessagingException { + throw new RuntimeException("problems"); + } + }); + PublishSubscribeChannel errorChannel = new PublishSubscribeChannel(); + SuccessfulErrorService errorService = new SuccessfulErrorService(); + ServiceActivatingHandler handler = new ServiceActivatingHandler(errorService); + handler.afterPropertiesSet(); + errorChannel.subscribe(handler); + MessageProducerSupport mps = new MessageProducerSupport() {}; + mps.setOutputChannel(outChannel); + mps.setErrorChannel(errorChannel); + mps.setBeanFactory(TestUtils.createTestApplicationContext()); + mps.afterPropertiesSet(); + mps.start(); + Message message = new GenericMessage("hello"); + mps.sendMessage(message); + assertEquals(ErrorMessage.class, errorService.lastMessage.getClass()); + ErrorMessage errorMessage = (ErrorMessage) errorService.lastMessage; + assertEquals(MessageDeliveryException.class, errorMessage.getPayload().getClass()); + MessageDeliveryException exception = (MessageDeliveryException) errorMessage.getPayload(); + assertEquals(message, exception.getFailedMessage()); + } + + + private static class SuccessfulErrorService { + + private volatile Message lastMessage; + + @SuppressWarnings("unused") + public void handleErrorMessage(Message errorMessage) { + this.lastMessage = errorMessage; + } + } + + + private static class FailingErrorService { + @SuppressWarnings("unused") + public void handleErrorMessage(Message errorMessage) { throw new RuntimeException("ooops"); } } + }