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 7a3ac3662e..6056b2c2ad 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 @@ -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, 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 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 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); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListenerTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListenerTests.java index 4125166d7c..7b732812e8 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListenerTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListenerTests.java @@ -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) request.getPayload()).toUpperCase()); ((MessageChannel) request.getHeaders().getReplyChannel()).send(reply, 5000); } diff --git a/src/reference/docbook/jms.xml b/src/reference/docbook/jms.xml index 440d267697..87172bc189 100644 --- a/src/reference/docbook/jms.xml +++ b/src/reference/docbook/jms.xml @@ -144,6 +144,19 @@ 'error-channel' should also be one-way. For example, it could simply send to a logging handler, or it could be connected to a different JMS <outbound-channel-adapter> element. + + Starting with version 4.2 the 'error-channel' is used for the conversion errors, too. + Previously, if a JMS <message-driven-channel-adapter> or <inbound gateway> could + not deliver a message due to a conversion error, an exception would be thrown back to the + container. If the container was configured to use transactions, the message would be + rolled back and redelivered repeatedly. The conversion process occurs before and during + message construction so such errors were not sent to the 'error-channel'. Now such conversion exceptions + result in an ErrorMessage being sent to the 'error-channel', with the exception as + the payload. If you wish the transaction to be rolled back, and you have an 'error-channel' + defined, the integration flow on the 'error-channel' must re-throw the exception (or another). If the error + flow does not throw an exception, the transaction will be committed and the message removed. + If no 'error-channel' is defined, the exception is thrown back to the container, as before. +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 9e7137d489..64dd094767 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -108,5 +108,15 @@ gateway; the dependency was removed for those users who don't need it.
+
+ JMS Changes + + The error-channel now is used for the conversion errors, which have caused a transaction + rollback and message redelivery previously. + + + See for more information. + +