Fix compatibility with latest Spring AMQP (#3352)

* Fix compatibility with latest Spring AMQP

* * Fix typo in the `amqp.adoc`

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java

@@ -24,6 +24,7 @@ import java.util.UUID;
 import java.util.concurrent.ScheduledFuture;
 
 import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.core.ReturnedMessage;
 import org.springframework.amqp.rabbit.connection.Connection;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
@@ -602,25 +603,44 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
 				: getMessageBuilderFactory().withPayload(replyObject);
 	}
 
+	/**
+	 * Build Spring message object based on the provided returned AMQP message info.
+	 * @param message the returned AMQP message
+	 * @param replyCode the returned message reason code
+	 * @param replyText the returned message reason text
+	 * @param exchange the exchange the message returned from
+	 * @param returnedRoutingKey the routing key for returned message
+	 * @param converter the converter to deserialize body of the returned AMQP message
+	 * @return the Spring message which represents a returned AMQP message
+	 * @deprecated since 5.4 in favor of {@link #buildReturnedMessage(ReturnedMessage, MessageConverter)}
+	 */
+	@Deprecated
 	protected Message<?> buildReturnedMessage(org.springframework.amqp.core.Message message,
 			int replyCode, String replyText, String exchange, String returnedRoutingKey, MessageConverter converter) {
 
-		Object returnedObject = converter.fromMessage(message);
+		return buildReturnedMessage(new ReturnedMessage(message, replyCode, replyText, exchange, returnedRoutingKey),
+				converter);
+	}
+
+	protected Message<?> buildReturnedMessage(ReturnedMessage returnedMessage, MessageConverter converter) {
+		org.springframework.amqp.core.Message amqpMessage = returnedMessage.getMessage();
+		Object returnedObject = converter.fromMessage(amqpMessage);
 		AbstractIntegrationMessageBuilder<?> builder = prepareMessageBuilder(returnedObject);
-		Map<String, ?> headers = getHeaderMapper().toHeadersFromReply(message.getMessageProperties());
+		Map<String, ?> headers = getHeaderMapper().toHeadersFromReply(amqpMessage.getMessageProperties());
 		if (this.errorMessageStrategy == null) {
 			builder.copyHeadersIfAbsent(headers)
-					.setHeader(AmqpHeaders.RETURN_REPLY_CODE, replyCode)
-					.setHeader(AmqpHeaders.RETURN_REPLY_TEXT, replyText)
-					.setHeader(AmqpHeaders.RETURN_EXCHANGE, exchange)
-					.setHeader(AmqpHeaders.RETURN_ROUTING_KEY, returnedRoutingKey);
+					.setHeader(AmqpHeaders.RETURN_REPLY_CODE, returnedMessage.getReplyCode())
+					.setHeader(AmqpHeaders.RETURN_REPLY_TEXT, returnedMessage.getReplyText())
+					.setHeader(AmqpHeaders.RETURN_EXCHANGE, returnedMessage.getExchange())
+					.setHeader(AmqpHeaders.RETURN_ROUTING_KEY, returnedMessage.getRoutingKey());
 		}
-		Message<?> returnedMessage = builder.build();
+		Message<?> message = builder.build();
 		if (this.errorMessageStrategy != null) {
-			returnedMessage = this.errorMessageStrategy.buildErrorMessage(new ReturnedAmqpMessageException(
-					returnedMessage, message, replyCode, replyText, exchange, returnedRoutingKey), null);
+			message = this.errorMessageStrategy.buildErrorMessage(new ReturnedAmqpMessageException(
+					message, amqpMessage, returnedMessage.getReplyCode(), returnedMessage.getReplyText(),
+					returnedMessage.getExchange(), returnedMessage.getRoutingKey()), null);
 		}
-		return returnedMessage;
+		return message;
 	}
 
 	protected void handleConfirm(CorrelationData correlationData, boolean ack, String cause) {

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java

@@ -23,11 +23,11 @@ import java.util.concurrent.TimeoutException;
 
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.ReturnedMessage;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
 import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
-import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.IntegrationPatternType;
@@ -48,7 +48,7 @@ import org.springframework.util.Assert;
  * @since 2.1
  */
 public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
-		implements ConfirmCallback, ReturnCallback {
+		implements ConfirmCallback, RabbitTemplate.ReturnsCallback {
 
 	private static final Duration DEFAULT_CONFIRM_TIMEOUT = Duration.ofSeconds(5);
 
@@ -141,7 +141,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
 		if (getReturnChannel() != null) {
 			Assert.notNull(this.rabbitTemplate,
 					"RabbitTemplate implementation is required for publisher confirms");
-			this.rabbitTemplate.setReturnCallback(this);
+			this.rabbitTemplate.setReturnsCallback(this);
 		}
 		Duration confirmTimeout = getConfirmTimeout();
 		if (confirmTimeout != null) {
@@ -267,13 +267,10 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
 	}
 
 	@Override
-	public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
-			String exchange, String routingKey) {
-
+	public void returnedMessage(ReturnedMessage returnedMessage) {
 		// no need for null check; we asserted we have a RabbitTemplate in doInit()
 		MessageConverter converter = this.rabbitTemplate.getMessageConverter();
-		Message<?> returned = buildReturnedMessage(message, replyCode, replyText, exchange,
-				routingKey, converter);
+		Message<?> returned = buildReturnedMessage(returnedMessage, converter);
 		getReturnChannel().send(returned);
 	}
 

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java

@@ -18,6 +18,7 @@ package org.springframework.integration.amqp.outbound;
 
 import org.springframework.amqp.core.AmqpMessageReturnedException;
 import org.springframework.amqp.core.AmqpReplyTimeoutException;
+import org.springframework.amqp.core.ReturnedMessage;
 import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
 import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
@@ -147,8 +148,9 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint {
 				else {
 					AmqpMessageReturnedException amre = (AmqpMessageReturnedException) ex;
 					Message<?> returnedMessage = buildReturnedMessage(
-							amre.getReturnedMessage(), amre.getReplyCode(), amre.getReplyText(), amre.getExchange(),
-							amre.getRoutingKey(), AsyncAmqpOutboundGateway.this.messageConverter);
+							new ReturnedMessage(amre.getReturnedMessage(), amre.getReplyCode(), amre.getReplyText(),
+									amre.getExchange(), amre.getRoutingKey()),
+							AsyncAmqpOutboundGateway.this.messageConverter);
 					sendOutput(returnedMessage, getReturnChannel(), true);
 				}
 			}

src/reference/asciidoc/amqp.adoc

@@ -309,7 +309,7 @@ The following example shows how to configure an `AmqpMessageSource` with Java co
 ----
 @Bean
 public AmqpMessageSource source(ConnectionFactory connectionFactory) {
-    return new AmpqpMessageSource(connectionFactory, "someQueue");
+    return new AmqpMessageSource(connectionFactory, "someQueue");
 }
 ----
 ====