Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Mark Fisher
2010-06-25 03:47:04 +00:00
parent fef67c7326
commit accedde988
2 changed files with 47 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java
View File

@@ -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<Throwable> 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<Throwable> getExceptionMapper() {
return exceptionMapper;
}
public void setExceptionMapper(InboundMessageMapper<Throwable> 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");
}
}
}

23
spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java
View File

@@ -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<javax.jms.Message>, InitializingBean {
implements SessionAwareMessageListener<javax.jms.Message>, 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;
}