GH-3437: Add MessageRecoverer to AMQP Inbounds (#3465)

* GH-3437: Add `MessageRecoverer` to AMQP Inbounds

Fixes #3437

For better end-user experience with AMQP Inbound Endpoints, expose
a `MessageRecoverer` option
* Test the feature and document this new option

* * Fix Checkstyle & update Copyright

* Fix language in Docs

Co-authored-by: Gary Russell <[email protected]>

Co-authored-by: Gary Russell <[email protected]>

Author: artembilan

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.amqp.dsl;
 
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
@@ -102,4 +103,16 @@ public S recoveryCallback(RecoveryCallback<?> recoveryCallback) {
 		return _this();
 	}
 
+	/**
+	 * Set a {@link MessageRecoverer} when using retry within the adapter.
+	 * @param messageRecoverer the callback.
+	 * @return the spec.
+	 * @since 5.5
+	 * @see AmqpInboundChannelAdapter#setMessageRecoverer(MessageRecoverer)
+	 */
+	public S messageRecoverer(MessageRecoverer messageRecoverer) {
+		this.target.setMessageRecoverer(messageRecoverer);
+		return _this();
+	}
+
 }

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java

@@ -17,7 +17,9 @@
 package org.springframework.integration.amqp.dsl;
 
 import org.springframework.amqp.rabbit.batch.BatchingStrategy;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
 import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
@@ -176,4 +178,16 @@ public S replyHeadersMappedLast(boolean replyHeadersMappedLast) {
 		return _this();
 	}
 
+	/**
+	 * Set a {@link MessageRecoverer} when using retry within the adapter.
+	 * @param messageRecoverer the callback.
+	 * @return the spec.
+	 * @since 5.5
+	 * @see AmqpInboundChannelAdapter#setMessageRecoverer(MessageRecoverer)
+	 */
+	public S messageRecoverer(MessageRecoverer messageRecoverer) {
+		this.target.setMessageRecoverer(messageRecoverer);
+		return _this();
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -29,6 +29,8 @@
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareBatchMessageListener;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+import org.springframework.amqp.rabbit.retry.MessageBatchRecoverer;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.support.converter.MessageConverter;
@@ -110,6 +112,8 @@ public enum BatchMode {
 
 	private RecoveryCallback<?> recoveryCallback;
 
+	private MessageRecoverer messageRecoverer;
+
 	private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
 
 	private boolean bindSourceMessage;
@@ -154,6 +158,7 @@ public void setRetryTemplate(RetryTemplate retryTemplate) {
 
 	/**
 	 * Set a {@link RecoveryCallback} when using retry within the adapter.
+	 * Mutually exclusive with {@link #setMessageRecoverer(MessageRecoverer)}.
 	 * @param recoveryCallback the callback.
 	 * @since 4.3.10
 	 * @see #setRetryTemplate(RetryTemplate)
@@ -162,6 +167,16 @@ public void setRecoveryCallback(RecoveryCallback<?> recoveryCallback) {
 		this.recoveryCallback = recoveryCallback;
 	}
 
+	/**
+	 * Configure a {@link MessageRecoverer} for retry operations.
+	 * A more AMQP-specific convenience instead of {@link #setRecoveryCallback(RecoveryCallback)}.
+	 * @param messageRecoverer the {@link MessageRecoverer} to use.
+	 * @since 5.5
+	 */
+	public void setMessageRecoverer(MessageRecoverer messageRecoverer) {
+		this.messageRecoverer = messageRecoverer;
+	}
+
 	/**
 	 * Set a batching strategy to use when de-batching messages created by a batching
 	 * producer (such as the BatchingRabbitTemplate).
@@ -206,6 +221,7 @@ protected void onInit() {
 			Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
 					+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
 					+ "send an error message when retries are exhausted");
+			setupRecoveryCallbackIfAny();
 		}
 		Listener messageListener;
 		if (this.messageListenerContainer.isConsumerBatchEnabled()) {
@@ -219,6 +235,38 @@ protected void onInit() {
 		super.onInit();
 	}
 
+	private void setupRecoveryCallbackIfAny() {
+		Assert.state(this.recoveryCallback == null || this.messageRecoverer == null,
+				"Only one of 'recoveryCallback' or 'messageRecoverer' may be provided, but not both");
+		if (this.messageRecoverer != null) {
+			if (this.messageListenerContainer.isConsumerBatchEnabled()) {
+				Assert.isInstanceOf(MessageBatchRecoverer.class, this.messageRecoverer,
+						"The 'messageRecoverer' must be an instance of MessageBatchRecoverer " +
+								"when consumer configured for batch mode");
+				this.recoveryCallback =
+						context -> {
+							@SuppressWarnings("unchecked")
+							List<Message> messagesToRecover =
+									(List<Message>) RetrySynchronizationManager.getContext()
+											.getAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
+							((MessageBatchRecoverer) this.messageRecoverer).recover(messagesToRecover,
+									context.getLastThrowable());
+							return null;
+						};
+			}
+			else {
+				this.recoveryCallback =
+						context -> {
+							Message messageToRecover =
+									(Message) RetrySynchronizationManager.getContext()
+											.getAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
+							this.messageRecoverer.recover(messageToRecover, context.getLastThrowable());
+							return null;
+						};
+			}
+		}
+	}
+
 	@Override
 	protected void doStart() {
 		this.messageListenerContainer.start();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -30,6 +30,7 @@
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
@@ -83,7 +84,9 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
 
 	private RetryTemplate retryTemplate;
 
-	private RecoveryCallback<? extends Object> recoveryCallback;
+	private RecoveryCallback<?> recoveryCallback;
+
+	private MessageRecoverer messageRecoverer;
 
 	private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
 
@@ -181,6 +184,7 @@ public void setRetryTemplate(RetryTemplate retryTemplate) {
 
 	/**
 	 * Set a {@link RecoveryCallback} when using retry within the gateway.
+	 * Mutually exclusive with {@link #setMessageRecoverer(MessageRecoverer)}.
 	 * @param recoveryCallback the callback.
 	 * @since 4.3.10
 	 * @see #setRetryTemplate(RetryTemplate)
@@ -189,6 +193,16 @@ public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallb
 		this.recoveryCallback = recoveryCallback;
 	}
 
+	/**
+	 * Configure a {@link MessageRecoverer} for retry operations.
+	 * A more AMQP-specific convenience instead of {@link #setRecoveryCallback(RecoveryCallback)}.
+	 * @param messageRecoverer the {@link MessageRecoverer} to use.
+	 * @since 5.5
+	 */
+	public void setMessageRecoverer(MessageRecoverer messageRecoverer) {
+		this.messageRecoverer = messageRecoverer;
+	}
+
 	/**
 	 * Set a batching strategy to use when de-batching messages.
 	 * Default is {@link SimpleBatchingStrategy}.
@@ -239,6 +253,7 @@ protected void onInit() {
 			Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
 					+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
 					+ "send an error message when retries are exhausted");
+			setupRecoveryCallbackIfAny();
 		}
 		Listener messageListener = new Listener();
 		this.messageListenerContainer.setMessageListener(messageListener);
@@ -254,6 +269,21 @@ protected void onInit() {
 		}
 	}
 
+	private void setupRecoveryCallbackIfAny() {
+		Assert.state(this.recoveryCallback == null || this.messageRecoverer == null,
+				"Only one of 'recoveryCallback' or 'messageRecoverer' may be provided, but not both");
+		if (this.messageRecoverer != null) {
+				this.recoveryCallback =
+						context -> {
+							Message messageToRecover =
+									(Message) RetrySynchronizationManager.getContext()
+											.getAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
+							this.messageRecoverer.recover(messageToRecover, context.getLastThrowable());
+							return null;
+						};
+		}
+	}
+
 	@Override
 	protected void doStart() {
 		super.doStart();