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

Fix for previous commit

Browse Source 
- failed to remove `RetryListener` from the gateway.
This commit is contained in:
Gary Russell
2018-09-19 12:11:52 -04:00
committed by Artem Bilan
parent dd6afa81d2
commit 8582d402ce
1 changed files with 5 additions and 28 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/AmqpInboundGateway.java
View File

@@ -42,9 +42,7 @@ import org.springframework.integration.gateway.MessagingGatewaySupport;
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;
import org.springframework.util.StringUtils;
@@ -190,9 +188,6 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
+ "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();
if (!this.amqpTemplateExplicitlySet) {
@@ -232,7 +227,9 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
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);
@@ -251,7 +248,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
}
}
protected class Listener implements ChannelAwareMessageListener, RetryListener {
protected class Listener implements ChannelAwareMessageListener {
@SuppressWarnings("unchecked")
@Override
@@ -362,26 +359,6 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
}
}
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
if (AmqpInboundGateway.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
}
}
}