From accedde988de33d75831c1d946e79d17388c1cd3 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 25 Jun 2010 03:47:04 +0000 Subject: [PATCH] INT-904, INT-907 updated error handling in sendAndReceive so that exceptionMapper is consulted for ErrorMessage payloads as well - in case an error occurred downstream in an asynchronous pipeline --- .../gateway/AbstractMessagingGateway.java | 80 ++++++++++--------- .../ChannelPublishingJmsMessageListener.java | 23 ++---- 2 files changed, 47 insertions(+), 56 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java index 66bec0b28a..46e3099694 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java @@ -31,7 +31,6 @@ import org.springframework.integration.message.InboundMessageMapper; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.MessageMappingException; import org.springframework.scheduling.support.PeriodicTrigger; import org.springframework.util.Assert; @@ -114,6 +113,16 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { this.shouldThrowErrors = shouldThrowErrors; } + /** + * 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; + } + @Override protected void onInit() throws Exception { this.initialized = true; @@ -180,26 +189,27 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { this.registerReplyMessageCorrelator(); } Message reply = null; + Throwable error = null; try { reply = this.channelTemplate.sendAndReceive(message, this.requestChannel); - } catch (Exception e) { - logger.warn("Execution of endpoint by the MessageListener resulted in : " + e); - if (this.exceptionMapper != null){ - try { - reply = exceptionMapper.toMessage(e); - } catch (Exception e2) { - logger.warn("Problem mapping " + e + " to message with: " + exceptionMapper); - } - } - if (reply == null){ // if reply wasn't mapped re-throw - if (e instanceof RuntimeException){ - throw (RuntimeException)e; - } - } } - - if (reply != null && this.shouldThrowErrors && reply instanceof ErrorMessage) { - Throwable error = ((ErrorMessage) reply).getPayload(); + catch (Exception e) { + logger.warn("failure occurred in gateway sendAndReceive.", e); + error = e; + } + if (reply instanceof ErrorMessage) { + error = ((ErrorMessage) reply).getPayload(); + } + if (error != null && this.exceptionMapper != null) { + try { + // create a reply message from the error + return this.exceptionMapper.toMessage(error); + } + catch (Exception e2) { + // ignore this, we'll handle the original error next + } + } + if (error != null && this.shouldThrowErrors) { if (error instanceof RuntimeException) { throw (RuntimeException) error; } @@ -235,6 +245,19 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { } } + protected Object fromMessage(Message message) { + return (message != null ? message.getPayload() : null); + } + + protected Message toMessage(Object object) { + if (object instanceof Message) { + return (Message) object; + } + else { + return MessageBuilder.withPayload(object).build(); + } + } + @Override // guarded by super#lifecycleLock protected void doStart() { if (this.replyMessageCorrelator != null) { @@ -249,25 +272,4 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { } } - public InboundMessageMapper getExceptionMapper() { - return exceptionMapper; - } - - public void setExceptionMapper(InboundMessageMapper exceptionMapper) { - this.exceptionMapper = exceptionMapper; - } - - protected Object fromMessage(Message message) { - throw new MessageMappingException("Can not map " + message + " to a object. No Mappers defined"); - } - - - protected Message toMessage(Object object) { - if (object instanceof Message){ - return (Message) object; - } else { - throw new MessageMappingException("Can not map " + object + " to a message. No Mappers defined"); - } - } - } 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 9f779a0093..2e3fcb431c 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 @@ -23,8 +23,6 @@ import javax.jms.JMSException; import javax.jms.MessageProducer; import javax.jms.Session; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.core.Message; import org.springframework.integration.gateway.AbstractMessagingGateway; @@ -46,10 +44,8 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky */ public class ChannelPublishingJmsMessageListener extends AbstractMessagingGateway - implements SessionAwareMessageListener, InitializingBean { + implements SessionAwareMessageListener, InitializingBean { - private final Log logger = LogFactory.getLog(this.getClass()); - private volatile boolean expectReply; private volatile MessageConverter messageConverter; @@ -201,21 +197,17 @@ public class ChannelPublishingJmsMessageListener extends AbstractMessagingGatewa public void setExtractReplyPayload(boolean extractReplyPayload) { this.extractReplyPayload = extractReplyPayload; } - /* - * (non-Javadoc) - * @see org.springframework.integration.gateway.AbstractMessagingGateway#onInit() - */ - public final void onInit() { + + public final void onInit() throws Exception { if (!(this.messageConverter instanceof HeaderMappingMessageConverter)) { HeaderMappingMessageConverter hmmc = new HeaderMappingMessageConverter(this.messageConverter, this.headerMapper); hmmc.setExtractJmsMessageBody(this.extractRequestPayload); hmmc.setExtractIntegrationMessagePayload(this.extractReplyPayload); this.messageConverter = hmmc; } + super.onInit(); } - /** - * - */ + public void onMessage(javax.jms.Message jmsMessage, Session session) throws JMSException { Object object = this.messageConverter.fromMessage(jmsMessage); Message requestMessage = (object instanceof Message) ? @@ -225,8 +217,7 @@ public class ChannelPublishingJmsMessageListener extends AbstractMessagingGatewa } else { Message replyMessage = this.sendAndReceiveMessage(requestMessage); - - if (replyMessage != null){ + if (replyMessage != null) { Destination destination = this.getReplyDestination(jmsMessage, session); if (destination != null){ javax.jms.Message jmsReply = this.messageConverter.toMessage(replyMessage, session); @@ -270,7 +261,6 @@ public class ChannelPublishingJmsMessageListener extends AbstractMessagingGatewa * @see javax.jms.Message#getJMSReplyTo() */ private Destination getReplyDestination(javax.jms.Message request, Session session) throws JMSException { - Destination replyTo = request.getJMSReplyTo(); if (replyTo == null) { replyTo = resolveDefaultReplyDestination(session); @@ -279,7 +269,6 @@ public class ChannelPublishingJmsMessageListener extends AbstractMessagingGatewa "Request message does not contain reply-to destination, and no default reply destination set."); } } - return replyTo; }