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

INT-3670: JMS Conversion Error Handling

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3670

Exceptions during inbound message conversion are now routed to the error-channel (if present).

When there is no error-channel, the exception is thrown back to the container as before.

Add a note to the `jms.xml` about the `error-channel` usage changes
This commit is contained in:
Gary Russell
2015-03-03 17:34:04 -05:00
committed by Artem Bilan
parent 75ff56962a
commit e7534aa4b8
4 changed files with 125 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -45,6 +45,8 @@ import org.springframework.jms.support.destination.DestinationResolver;
import org.springframework.jms.support.destination.DynamicDestinationResolver;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.Assert;
/**
@@ -57,6 +59,7 @@ import org.springframework.util.Assert;
* @author Juergen Hoeller
* @author Oleg Zhurakousky
* @author Artem Bilan
* @author Gary Russell
*/
public class ChannelPublishingJmsMessageListener
implements SessionAwareMessageListener<javax.jms.Message>, InitializingBean,
@@ -306,46 +309,60 @@ public class ChannelPublishingJmsMessageListener
@Override
public void onMessage(javax.jms.Message jmsMessage, Session session) throws JMSException {
Object result = jmsMessage;
if (this.extractRequestPayload) {
result = this.messageConverter.fromMessage(jmsMessage);
if (logger.isDebugEnabled()) {
logger.debug("converted JMS Message [" + jmsMessage + "] to integration Message payload ["
+ result + "]");
}
}
Map<String, Object> headers = headerMapper.toHeaders(jmsMessage);
Message<?> requestMessage = (result instanceof Message<?>) ?
this.messageBuilderFactory.fromMessage((Message<?>) result).copyHeaders(headers).build() :
this.messageBuilderFactory.withPayload(result).copyHeaders(headers).build();
if (!this.expectReply) {
this.gatewayDelegate.send(requestMessage);
}
else {
Message<?> replyMessage = this.gatewayDelegate.sendAndReceiveMessage(requestMessage);
if (replyMessage != null) {
Destination destination = this.getReplyDestination(jmsMessage, session);
if (destination != null) {
// convert SI Message to JMS Message
Object replyResult = replyMessage;
if (this.extractReplyPayload) {
replyResult = replyMessage.getPayload();
}
try {
javax.jms.Message jmsReply = this.messageConverter.toMessage(replyResult, session);
// map SI Message Headers to JMS Message Properties/Headers
headerMapper.fromHeaders(replyMessage.getHeaders(), jmsReply);
this.copyCorrelationIdFromRequestToReply(jmsMessage, jmsReply);
this.sendReply(jmsReply, destination, session);
}
catch (RuntimeException e) {
logger.error("Failed to generate JMS Reply Message from: " + replyResult, e);
throw e;
}
Message<?> requestMessage = null;
boolean errors = false;
try {
if (this.extractRequestPayload) {
result = this.messageConverter.fromMessage(jmsMessage);
if (logger.isDebugEnabled()) {
logger.debug("converted JMS Message [" + jmsMessage + "] to integration Message payload ["
+ result + "]");
}
}
else if (logger.isDebugEnabled()) {
logger.debug("expected a reply but none was received");
Map<String, Object> headers = headerMapper.toHeaders(jmsMessage);
requestMessage = (result instanceof Message<?>) ?
this.messageBuilderFactory.fromMessage((Message<?>) result).copyHeaders(headers).build() :
this.messageBuilderFactory.withPayload(result).copyHeaders(headers).build();
}
catch (RuntimeException e) {
MessageChannel errorChannel = this.gatewayDelegate.getErrorChannel();
if (errorChannel == null) {
throw e;
}
errorChannel.send(new ErrorMessage(new MessagingException("Inbound conversion failed for: " + jmsMessage, e)));
errors = true;
}
if (!errors) {
if (!this.expectReply) {
this.gatewayDelegate.send(requestMessage);
}
else {
Message<?> replyMessage = this.gatewayDelegate.sendAndReceiveMessage(requestMessage);
if (replyMessage != null) {
Destination destination = this.getReplyDestination(jmsMessage, session);
if (destination != null) {
// convert SI Message to JMS Message
Object replyResult = replyMessage;
if (this.extractReplyPayload) {
replyResult = replyMessage.getPayload();
}
try {
javax.jms.Message jmsReply = this.messageConverter.toMessage(replyResult, session);
// map SI Message Headers to JMS Message Properties/Headers
headerMapper.fromHeaders(replyMessage.getHeaders(), jmsReply);
this.copyCorrelationIdFromRequestToReply(jmsMessage, jmsReply);
this.sendReply(jmsReply, destination, session);
}
catch (RuntimeException e) {
logger.error("Failed to generate JMS Reply Message from: " + replyResult, e);
throw e;
}
}
}
else if (logger.isDebugEnabled()) {
logger.debug("expected a reply but none was received");
}
}
}
}
@@ -474,6 +491,11 @@ public class ChannelPublishingJmsMessageListener
private class GatewayDelegate extends MessagingGatewaySupport {
@Override
public MessageChannel getErrorChannel() {
return super.getErrorChannel();
}
@Override
protected void send(Object request) {
super.send(request);

44
spring-integration-jms/src/test/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListenerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,22 +16,33 @@
package org.springframework.integration.jms;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Session;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.mockito.Matchers;
import org.mockito.internal.stubbing.answers.DoesNothing;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
/**
@@ -46,7 +57,7 @@ public class ChannelPublishingJmsMessageListenerTests {
@Test(expected = InvalidDestinationException.class)
public void noReplyToAndNoDefault() throws JMSException {
final QueueChannel requestChannel = new QueueChannel();
this.startBackgroundReplier(requestChannel);
startBackgroundReplier(requestChannel);
ChannelPublishingJmsMessageListener listener = new ChannelPublishingJmsMessageListener();
listener.setExpectReply(true);
listener.setRequestChannel(requestChannel);
@@ -57,11 +68,38 @@ public class ChannelPublishingJmsMessageListenerTests {
listener.onMessage(jmsMessage, session);
}
@Test
public void testBadConversion() throws Exception {
final QueueChannel requestChannel = new QueueChannel();
ChannelPublishingJmsMessageListener listener = new ChannelPublishingJmsMessageListener();
Log logger = spy(TestUtils.getPropertyValue(listener, "logger", Log.class));
doAnswer(new DoesNothing()).when(logger).error(Matchers.anyString(), Matchers.any(Throwable.class));
new DirectFieldAccessor(listener).setPropertyValue("logger", logger);
listener.setRequestChannel(requestChannel);
QueueChannel errorChannel = new QueueChannel();
listener.setErrorChannel(errorChannel);
listener.setBeanFactory(mock(BeanFactory.class));
listener.setMessageConverter(new TestMessageConverter() {
@Override
public Object fromMessage(javax.jms.Message message) throws JMSException, MessageConversionException {
return null;
}
});
listener.afterPropertiesSet();
javax.jms.Message jmsMessage = session.createTextMessage("test");
listener.onMessage(jmsMessage, mock(Session.class));
ErrorMessage received = (ErrorMessage) errorChannel.receive(0);
assertNotNull(received);
assertThat(received.getPayload().getMessage(), startsWith("Inbound conversion failed"));
}
private void startBackgroundReplier(final PollableChannel channel) {
new SimpleAsyncTaskExecutor().execute(new Runnable() {
@Override
public void run() {
Message<?> request = channel.receive(5000);
Message<?> request = channel.receive(50000);
Message<?> reply = new GenericMessage<String>(((String) request.getPayload()).toUpperCase());
((MessageChannel) request.getHeaders().getReplyChannel()).send(reply, 5000);
}