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

Use RetrySynchronizationManager

Browse Source 
Use the `RetrySynchronizationManager` instead of a `RetryListener`.

Fix `setAttributesIfNecessary` for gateway conversion errors (this, at least,
should be cherry-picked).
This commit is contained in:
Gary Russell
2018-09-18 16:05:01 -04:00
committed by Artem Bilan
parent 585258373e
commit dd6afa81d2
3 changed files with 17 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -39,9 +39,7 @@ import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.support.RetrySynchronizationManager;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
@@ -132,9 +130,6 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
+ "send an error message when retries are exhausted");
}
Listener messageListener = new Listener();
if (this.retryTemplate != null) {
this.retryTemplate.registerListener(messageListener);
}
this.messageListenerContainer.setMessageListener(messageListener);
this.messageListenerContainer.afterPropertiesSet();
super.onInit();
@@ -177,7 +172,9 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
}
if (needAttributes) {
AttributeAccessor attributes = attributesHolder.get();
AttributeAccessor attributes = this.retryTemplate != null
? RetrySynchronizationManager.getContext()
: attributesHolder.get();
if (attributes != null) {
attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
attributes.setAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, amqpMessage);
@@ -196,7 +193,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
}
}
protected class Listener implements ChannelAwareMessageListener, RetryListener {
protected class Listener implements ChannelAwareMessageListener {
@SuppressWarnings("unchecked")
@Override
@@ -259,26 +256,6 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
return messagingMessage;
}
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
if (AmqpInboundChannelAdapter.this.recoveryCallback != null) {
attributesHolder.set(context);
}
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
attributesHolder.remove();
}
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
}
}

1
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
View File

@@ -296,6 +296,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
}
catch (RuntimeException e) {
if (getErrorChannel() != null) {
setAttributesIfNecessary(message, null);
AmqpInboundGateway.this.messagingTemplate.send(getErrorChannel(), buildErrorMessage(null,
new ListenerExecutionFailedException("Message conversion failed", e, message)));
}

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

@@ -248,9 +248,12 @@ public class InboundEndpointTests {
});
adapter.afterPropertiesSet();
((ChannelAwareMessageListener) container.getMessageListener()).onMessage(null, null);
((ChannelAwareMessageListener) container.getMessageListener())
.onMessage(mock(org.springframework.amqp.core.Message.class), null);
assertNull(outputChannel.receive(0));
assertNotNull(errorChannel.receive(0));
Message<?> received = errorChannel.receive(0);
assertNotNull(received);
assertNotNull(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE));
}
@Test
@@ -276,14 +279,17 @@ public class InboundEndpointTests {
@Override
public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException {
return null;
throw new MessageConversionException("intended");
}
});
adapter.afterPropertiesSet();
((ChannelAwareMessageListener) container.getMessageListener()).onMessage(null, null);
((ChannelAwareMessageListener) container.getMessageListener())
.onMessage(mock(org.springframework.amqp.core.Message.class), null);
assertNull(outputChannel.receive(0));
assertNotNull(errorChannel.receive(0));
Message<?> received = errorChannel.receive(0);
assertNotNull(received);
assertNotNull(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE));
}
@Test