diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java index 82c950e0a2..81d03145ff 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java @@ -41,7 +41,7 @@ import org.springframework.util.xml.DomUtils; public class GatewayParser extends AbstractSimpleBeanDefinitionParser { private static String[] referenceAttributes = new String[] { - "default-request-channel", "default-reply-channel", "message-mapper", "exception-mapper" + "default-request-channel", "default-reply-channel", "error-channel", "message-mapper" }; private static String[] innerAttributes = new String[] { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index db8324f86f..83435a82e5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -49,7 +49,6 @@ import org.springframework.integration.annotation.Gateway; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.history.TrackableComponent; -import org.springframework.integration.mapping.InboundMessageMapper; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.integration.support.channel.ChannelResolver; import org.springframework.util.Assert; @@ -76,6 +75,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab private volatile MessageChannel defaultReplyChannel; + private volatile MessageChannel errorChannel; + private volatile long defaultRequestTimeout = -1; private volatile long defaultReplyTimeout = -1; @@ -94,8 +95,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab private volatile AsyncTaskExecutor asyncExecutor = new SimpleAsyncTaskExecutor(); - private volatile InboundMessageMapper exceptionMapper; - private volatile boolean initialized; private final Object initializationMonitor = new Object(); @@ -149,6 +148,15 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab this.defaultReplyChannel = defaultReplyChannel; } + /** + * Set the error channel. If no error channel is provided, this gateway will + * propagate Exceptions to the caller. To completely suppress Exceptions, provide + * a reference to the "nullChannel" here. + */ + public void setErrorChannel(MessageChannel errorChannel) { + this.errorChannel = errorChannel; + } + /** * Set the default timeout value for sending request messages. If not * explicitly configured with an annotation, this value will be used. @@ -193,10 +201,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab this.methodMetadataMap = methodMetadataMap; } - public void setExceptionMapper(InboundMessageMapper exceptionMapper) { - this.exceptionMapper = exceptionMapper; - } - public void setBeanClassLoader(ClassLoader beanClassLoader) { this.beanClassLoader = beanClassLoader; } @@ -363,7 +367,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab } messageMapper.setBeanFactory(this.getBeanFactory()); MethodInvocationGateway gateway = new MethodInvocationGateway(messageMapper); - gateway.setExceptionMapper(exceptionMapper); + gateway.setErrorChannel(this.errorChannel); if (this.getTaskScheduler() != null) { gateway.setTaskScheduler(this.getTaskScheduler()); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 865b85e650..87ee6cca4d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -54,21 +54,19 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement private volatile MessageChannel replyChannel; + private volatile MessageChannel errorChannel; + private volatile long replyTimeout = DEFAULT_TIMEOUT; @SuppressWarnings("rawtypes") private volatile InboundMessageMapper requestMapper = new DefaultRequestMapper(); - private volatile InboundMessageMapper exceptionMapper; - private final SimpleMessageConverter messageConverter = new SimpleMessageConverter(); private final MessagingTemplate messagingTemplate; private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor(); - private volatile boolean shouldThrowErrors = true; - private volatile boolean initialized; private volatile AbstractEndpoint replyMessageCorrelator; @@ -95,7 +93,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement } /** - * Set the reply channel. If no reply channel is provided, this template will + * Set the reply channel. If no reply channel is provided, this gateway will * always use an anonymous, temporary channel for handling replies. * * @param replyChannel the channel from which reply messages will be received @@ -104,6 +102,15 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement this.replyChannel = replyChannel; } + /** + * Set the error channel. If no error channel is provided, this gateway will + * propagate Exceptions to the caller. To completely suppress Exceptions, provide + * a reference to the "nullChannel" here. + */ + public void setErrorChannel(MessageChannel errorChannel) { + this.errorChannel = errorChannel; + } + /** * Set the timeout value for sending request messages. If not * explicitly configured, the default is one second. @@ -143,26 +150,6 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement this.messageConverter.setOutboundMessageMapper(replyMapper); } - /** - * Provide an {@link InboundMessageMapper} for creating a reply Message from - * an Exception that occurs downstream from this gateway. If no exceptionMapper - * is provided, then the {@link #shouldThrowErrors} property will dictate - * whether the error is rethrown or returned as an ErrorMessage. - */ - public void setExceptionMapper(InboundMessageMapper exceptionMapper) { - this.exceptionMapper = exceptionMapper; - } - - /** - * Specify whether the Throwable payload of a received {@link ErrorMessage} - * should be extracted and thrown from a send-and-receive operation. - * Otherwise, the ErrorMessage would be returned just like any other - * reply Message. The default is true. - */ - public void setShouldThrowErrors(boolean shouldThrowErrors) { - this.shouldThrowErrors = shouldThrowErrors; - } - /** * Specify whether this gateway should be tracked in the Message History * of Messages that originate from its send or sendAndReceive operations. @@ -193,7 +180,18 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement Assert.notNull(object, "request must not be null"); Assert.state(this.requestChannel != null, "send is not supported, because no request channel has been configured"); - this.messagingTemplate.convertAndSend(this.requestChannel, object, this.historyWritingPostProcessor); + try { + this.messagingTemplate.convertAndSend(this.requestChannel, object, this.historyWritingPostProcessor); + } + catch (Exception e) { + if (this.errorChannel != null) { + this.messagingTemplate.send(this.errorChannel, new ErrorMessage(e)); + } + else if (e instanceof RuntimeException) { + throw (RuntimeException) e; + } + throw new MessagingException("failed to send message", e); + } } protected Object receive() { @@ -239,30 +237,32 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement error = ((ErrorMessage) reply).getPayload(); } } - if (reply == null){ - - } } catch (Exception e) { logger.warn("failure occurred in gateway sendAndReceive", e); error = e; } - if (error != null && this.exceptionMapper != null) { - try { - // create a reply message from the error - Message errorMessage = this.exceptionMapper.toMessage(error); - return (shouldConvert) ? errorMessage.getPayload() : errorMessage; + if (error != null) { + if (this.errorChannel != null) { + Message errorMessage = null; + try { + Message errorFlowReply = this.messagingTemplate.sendAndReceive(this.errorChannel, new ErrorMessage(error)); + if (shouldConvert) { + return (errorFlowReply != null) ? errorFlowReply.getPayload() : null; + } + return errorFlowReply; + } + catch (Exception errorFlowFailure) { + throw new MessagingException(errorMessage, "failure occurred in error-handling flow", errorFlowFailure); + } } - catch (Exception e2) { - // ignore this, we'll handle the original error next - } - } - if (error != null && this.shouldThrowErrors) { - if (error instanceof RuntimeException) { - throw (RuntimeException) error; + else { // no errorChannel so we'll propagate + if (error instanceof RuntimeException) { + throw (RuntimeException) error; + } + throw new MessagingException("gateway received checked Exception", error); } - throw new MessagingException("gateway received checked Exception", error); } return reply; } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index 080d06f6ed..c9c58beaf6 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -515,9 +515,9 @@ - + @@ -526,16 +526,6 @@ - - - - - - - @@ -565,6 +555,16 @@ + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests-context.xml index d1a10e700b..f670df2e54 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests-context.xml @@ -16,19 +16,14 @@ default-request-channel="routingChannel" service-interface="org.springframework.integration.gateway.GatewayInvokingMessageHandlerTests$SimpleGateway"/> - - - + error-channel="errorTransformationChannel"/> @@ -67,8 +62,6 @@ - - @@ -87,7 +80,9 @@ + + - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java index 0610d5cc54..e4f799533d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java @@ -27,7 +27,6 @@ import org.springframework.integration.Message; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; -import org.springframework.integration.mapping.InboundMessageMapper; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.test.context.ContextConfiguration; @@ -53,21 +52,14 @@ public class GatewayInvokingMessageHandlerTests { @Qualifier("gatewayWithError") SimpleGateway gatewayWithError; - @Autowired - @Qualifier("gatewayWithErrorAndMapper") - SimpleGateway gatewayWithErrorAndMapper; - - @Autowired @Qualifier("gatewayWithErrorAsync") SimpleGateway gatewayWithErrorAsync; - + @Autowired - @Qualifier("gatewayWithErrorAsyncAndMapper") - SimpleGateway gatewayWithErrorAsyncAndMapper; - - - + @Qualifier("gatewayWithErrorChannelAndTransformer") + SimpleGateway gatewayWithErrorChannelAndTransformer; + @Autowired @Qualifier("inputB") SubscribableChannel output; @@ -93,38 +85,43 @@ public class GatewayInvokingMessageHandlerTests { Assert.assertEquals("oleg", message.getHeaders().get("name")); } }); - String result = gateway.sendRecieve("hello"); + String result = gateway.process("hello"); Assert.assertEquals("echo:echo:echo:hello", result); } @Test public void validateGatewayWithErrorMessageReturned() { - try { - String result = gatewayWithErrorAndMapper.sendRecieve("echoWithRuntimeExceptionChannel"); + String result = gatewayWithErrorChannelAndTransformer.process("echoWithRuntimeExceptionChannel"); Assert.assertNotNull(result); Assert.assertEquals("Error happened in message: echoWithRuntimeExceptionChannel", result); - } catch (Exception e) { + } + catch (Exception e) { Assert.fail(); } try { - gatewayWithError.sendRecieve("echoWithRuntimeExceptionChannel"); + gatewayWithError.process("echoWithRuntimeExceptionChannel"); Assert.fail(); - } catch (MessageHandlingException e) { + } + catch (MessageHandlingException e) { Assert.assertEquals("echoWithRuntimeExceptionChannel", e.getFailedMessage().getPayload()); } + try { - gatewayWithError.sendRecieve("echoWithMessagingExceptionChannel"); + gatewayWithError.process("echoWithMessagingExceptionChannel"); Assert.fail(); - } catch (MessageHandlingException e) { + } + catch (MessageHandlingException e) { Assert.assertEquals("echoWithMessagingExceptionChannel", e.getFailedMessage().getPayload()); } + try { - String result = gatewayWithErrorAndMapper.sendRecieve("echoWithMessagingExceptionChannel"); + String result = gatewayWithErrorChannelAndTransformer.process("echoWithMessagingExceptionChannel"); Assert.assertNotNull(result); Assert.assertEquals("Error happened in message: echoWithMessagingExceptionChannel", result); - } catch (Exception e) { + } + catch (Exception e) { Assert.fail(); } } @@ -132,25 +129,27 @@ public class GatewayInvokingMessageHandlerTests { @Test public void validateGatewayWithErrorAsync() { try { - gatewayWithErrorAsync.sendRecieve("echoWithErrorAsyncChannel"); + gatewayWithErrorAsync.process("echoWithErrorAsyncChannel"); Assert.fail(); - } catch (Exception e) { + } + catch (Exception e) { Assert.assertTrue(e instanceof MessageHandlingException); } } @Test - public void validateGatewayWithErrorAsyncAndMaper() { + public void validateGatewayWithErrorFlowReturningMessage() { try { - Object result = gatewayWithErrorAsyncAndMapper.sendRecieve("echoWithErrorAsyncChannel"); + Object result = gatewayWithErrorChannelAndTransformer.process("echoWithErrorAsyncChannel"); Assert.assertEquals("Error happened in message: echoWithErrorAsyncChannel", result); - } catch (Exception e) { + } + catch (Exception e) { Assert.fail(); } } - - - public static class SampleExceptionMapper implements InboundMessageMapper{ + + + public static class SampleErrorTransformer { public Message toMessage(Throwable object) throws Exception { MessageHandlingException ex = (MessageHandlingException) object; return MessageBuilder.withPayload("Error happened in message: " + ex.getFailedMessage().getPayload()).build(); @@ -158,10 +157,12 @@ public class GatewayInvokingMessageHandlerTests { } + public static interface SimpleGateway { - public String sendRecieve(String str); + public String process(String str); } + public static class SimpleService { public String echo(String value) { return "echo:" + value; @@ -176,9 +177,9 @@ public class GatewayInvokingMessageHandlerTests { public RuntimeException echoWithErrorAsync(String value) { throw new RuntimeException(value); } - } - + + @SuppressWarnings("serial") public static class SampleCheckedException extends Exception { public SampleCheckedException(String message){ diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index 1410eca60b..eebfdfa006 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -70,21 +70,24 @@ public class ChannelPublishingJmsMessageListener extends MessagingGatewaySupport private volatile DestinationResolver destinationResolver = new DynamicDestinationResolver(); private volatile JmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper(); - - public String getComponentType(){ - if (expectReply){ + + + public String getComponentType() { + if (expectReply) { return "jms:inbound-gateway"; - } else { + } + else { return "jms:message-driven-channel-adapter"; } } - + /** * Specify whether a JMS reply Message is expected. */ public void setExpectReply(boolean expectReply) { this.expectReply = expectReply; } + /** * Set the default reply destination to send reply messages to. This will * be applied in case of a request message that does not carry a diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java index 1dc54971d0..f2458e0295 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java @@ -161,7 +161,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-request-payload"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-reply-payload"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "exception-mapper"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); int defaults = 0; if (StringUtils.hasText(element.getAttribute(DEFAULT_REPLY_DESTINATION_ATTRIB))) { defaults++; diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd index 732751df32..cefd7dcd02 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd @@ -496,7 +496,6 @@ - @@ -535,6 +534,15 @@ + + + + + + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/Exception-nonSiProducer-siConsumer.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/Exception-nonSiProducer-siConsumer.xml index 10f74f74ea..0638bab753 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/Exception-nonSiProducer-siConsumer.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/Exception-nonSiProducer-siConsumer.xml @@ -23,16 +23,19 @@ + error-channel="errorTransformationChannel"/> + error-channel="errorTransformationChannel"/> - + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExceptionHandlingSiConsumerTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExceptionHandlingSiConsumerTests.java index 4eddd20e73..7f6d5f6ebb 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExceptionHandlingSiConsumerTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExceptionHandlingSiConsumerTests.java @@ -30,14 +30,12 @@ import org.junit.Test; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.mapping.InboundMessageMapper; import org.springframework.integration.support.MessageBuilder; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; /** * @author Oleg Zhurakousky - * */ public class ExceptionHandlingSiConsumerTests { @@ -120,8 +118,8 @@ public class ExceptionHandlingSiConsumerTests { } - public static class SampleErrorMessageMapper implements InboundMessageMapper { - public org.springframework.integration.Message toMessage(Throwable t) throws Exception { + public static class SampleErrorTransformer { + public org.springframework.integration.Message transform(Throwable t) throws Exception { return MessageBuilder.withPayload(t.getCause().getMessage()).build(); } }