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

INT-4148: Don't Wrap AMQP Listener Exceptions

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

The subscribable channel incorrectly wrapped exceptions such as `MessageConversionException`
preventing the error handler from properly handling them.
This commit is contained in:
Gary Russell
2016-10-27 15:10:24 -04:00
committed by Artem Bilan
parent c5fbd93787
commit 49638fdea2
2 changed files with 31 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
View File

@@ -42,7 +42,6 @@ import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.util.Assert;
@@ -186,7 +185,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS,
Integer.class);
}
this.setMaxSubscribers(this.maxSubscribers);
setMaxSubscribers(this.maxSubscribers);
String queue = this.obtainQueueName(this.admin, this.channelName);
this.container.setQueueNames(queue);
MessageConverter converter = (this.getAmqpTemplate() instanceof RabbitTemplate)
@@ -307,14 +306,9 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
}
}
else {
throw new MessageDeliveryException(
messageToSend, exceptionMessage, e);
throw new MessageDeliveryException(messageToSend, exceptionMessage, e);
}
}
catch (Exception e) {
throw new MessagingException("Failure occurred in AMQP listener " +
"while attempting to convert and dispatch Message.", e);
}
}
protected Message<Object> buildMessage(org.springframework.amqp.core.Message message, Object converted) {

30
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java
View File

@@ -23,6 +23,8 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.BDDMockito.willThrow;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import java.util.Collection;
@@ -32,15 +34,21 @@ import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
@@ -70,7 +78,10 @@ public class ChannelTests extends LogAdjustingTestSupport {
@ClassRule
public static final BrokerRunning brokerIsRunning =
BrokerRunning.isRunningWithEmptyQueues("pollableWithEP", "withEP");
BrokerRunning.isRunningWithEmptyQueues("pollableWithEP", "withEP", "testConvertFail");
@Rule
public ExpectedException exception = ExpectedException.none();
@Autowired
private PublishSubscribeAmqpChannel channel;
@@ -231,6 +242,23 @@ public class ChannelTests extends LogAdjustingTestSupport {
assertSame(this.mapperOut, TestUtils.getPropertyValue(this.pollableWithEP, "outboundHeaderMapper"));
}
@Test
public void messageConversionTests() throws Exception {
RabbitTemplate amqpTemplate = new RabbitTemplate(this.factory);
MessageConverter messageConverter = mock(MessageConverter.class);
amqpTemplate.setMessageConverter(messageConverter);
PointToPointSubscribableAmqpChannel channel = new PointToPointSubscribableAmqpChannel("testConvertFail",
new SimpleMessageListenerContainer(this.factory), amqpTemplate);
channel.afterPropertiesSet();
MessageListener listener = TestUtils.getPropertyValue(channel, "container.messageListener",
MessageListener.class);
willThrow(new MessageConversionException("foo", new IllegalStateException("bar")))
.given(messageConverter).fromMessage(any(org.springframework.amqp.core.Message.class));
this.exception.expect(MessageConversionException.class);
this.exception.expectCause(instanceOf(IllegalStateException.class));
listener.onMessage(mock(org.springframework.amqp.core.Message.class));
}
public static class Foo {
private String bar;