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

INT-4482: AMQP: Fix Double ErrorMessage

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

The outer try/catch sends an `ErrorMessage` for all exceptions; it should
only do so for `MessageConversionException`. Integration flow exceptions
will have been already handled by `MessageProducerSupport`.

Also, populate the raw message header consistently - previously it only
was populated for flow exceptions. Although the LEFE contains the raw
message, it should be in the `ErrorMessage` header for consistency.

**cherry-pick to 5.0.x, 4.3.x**

* Polishing - PR Comments

# Conflicts:
#	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
#	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java
This commit is contained in:
Gary Russell
2018-06-05 16:34:58 -04:00
committed by Artem Bilan
parent 30c83b0ec7
commit 37c92a128b
2 changed files with 14 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
View File

@@ -24,6 +24,7 @@ import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.core.AttributeAccessor;
@@ -197,14 +198,11 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
@SuppressWarnings("unchecked")
@Override
public void onMessage(final Message message, final Channel channel) throws Exception {
boolean retryDisabled = AmqpInboundChannelAdapter.this.retryTemplate == null;
try {
if (AmqpInboundChannelAdapter.this.retryTemplate == null) {
try {
processMessage(message, channel);
}
finally {
attributesHolder.remove();
}
if (retryDisabled) {
processMessage(message, channel);
}
else {
AmqpInboundChannelAdapter.this.retryTemplate.execute(new RetryCallback<Object, RuntimeException>() {
@@ -218,8 +216,9 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
}, (RecoveryCallback<Object>) AmqpInboundChannelAdapter.this.recoveryCallback);
}
}
catch (RuntimeException e) {
catch (MessageConversionException e) {
if (getErrorChannel() != null) {
setAttributesIfNecessary(message, null);
getMessagingTemplate().send(getErrorChannel(), buildErrorMessage(null,
new ListenerExecutionFailedException("Message conversion failed", e, message)));
}
@@ -227,6 +226,11 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
throw e;
}
}
finally {
if (retryDisabled) {
attributesHolder.remove();
}
}
}
private void processMessage(Message message, Channel channel) {

10
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
View File

@@ -251,17 +251,11 @@ public class InboundEndpointTests {
adapter.setOutputChannel(outputChannel);
QueueChannel errorChannel = new QueueChannel();
adapter.setErrorChannel(errorChannel);
adapter.setMessageConverter(new MessageConverter() {
@Override
public org.springframework.amqp.core.Message toMessage(Object object, MessageProperties messageProperties)
throws MessageConversionException {
throw new MessageConversionException("intended");
}
adapter.setMessageConverter(new SimpleMessageConverter() {
@Override
public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException {
return null;
throw new MessageConversionException("intended");
}
});