GH-1473: Switch to CompletableFuture

Resolves #1473

* Deprecate `AsyncRabbitTemplate2`.
* Copy implementation from  `AsyncRabbitTemplate2` to `AsyncRabbitTemplate`
* Fix other `ListenableFuture` usages.

Author: garyrussell

spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2020-2022 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.
@@ -16,12 +16,13 @@
 
 package org.springframework.amqp.core;
 
+import java.util.concurrent.CompletableFuture;
+
 import org.springframework.core.ParameterizedTypeReference;
-import org.springframework.util.concurrent.ListenableFuture;
 
 /**
  * Classes implementing this interface can perform asynchronous send and
- * receive operations.
+ * receive operations using {@link CompletableFuture}s.
  *
  * @author Gary Russell
  * @since 2.0
@@ -33,47 +34,47 @@ public interface AsyncAmqpTemplate {
 	 * Send a message to the default exchange with the default routing key. If the message
 	 * contains a correlationId property, it must be unique.
 	 * @param message the message.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	ListenableFuture<Message> sendAndReceive(Message message);
+	CompletableFuture<Message> sendAndReceive(Message message);
 
 	/**
 	 * Send a message to the default exchange with the supplied routing key. If the message
 	 * contains a correlationId property, it must be unique.
 	 * @param routingKey the routing key.
 	 * @param message the message.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	ListenableFuture<Message> sendAndReceive(String routingKey, Message message);
+	CompletableFuture<Message> sendAndReceive(String routingKey, Message message);
 
 	/**
 	 * Send a message to the supplied exchange and routing key. If the message
 	 * contains a correlationId property, it must be unique.
 	 * @param exchange the exchange.
 	 * @param routingKey the routing key.
 	 * @param message the message.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	ListenableFuture<Message> sendAndReceive(String exchange, String routingKey, Message message);
+	CompletableFuture<Message> sendAndReceive(String exchange, String routingKey, Message message);
 
 	/**
 	 * Convert the object to a message and send it to the default exchange with the
 	 * default routing key.
 	 * @param object the object to convert.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceive(Object object);
+	<C> CompletableFuture<C> convertSendAndReceive(Object object);
 
 	/**
 	 * Convert the object to a message and send it to the default exchange with the
 	 * provided routing key.
 	 * @param routingKey the routing key.
 	 * @param object the object to convert.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceive(String routingKey, Object object);
+	<C> CompletableFuture<C> convertSendAndReceive(String routingKey, Object object);
 
 	/**
 	 * Convert the object to a message and send it to the provided exchange and
@@ -82,9 +83,9 @@ public interface AsyncAmqpTemplate {
 	 * @param routingKey the routing key.
 	 * @param object the object to convert.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object);
+	<C> CompletableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object);
 
 	/**
 	 * Convert the object to a message and send it to the default exchange with the
@@ -93,9 +94,9 @@ public interface AsyncAmqpTemplate {
 	 * @param object the object to convert.
 	 * @param messagePostProcessor the post processor.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor);
+	<C> CompletableFuture<C> convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor);
 
 	/**
 	 * Convert the object to a message and send it to the default exchange with the
@@ -105,9 +106,9 @@ public interface AsyncAmqpTemplate {
 	 * @param object the object to convert.
 	 * @param messagePostProcessor the post processor.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceive(String routingKey, Object object,
+	<C> CompletableFuture<C> convertSendAndReceive(String routingKey, Object object,
 			MessagePostProcessor messagePostProcessor);
 
 	/**
@@ -119,9 +120,9 @@ <C> ListenableFuture<C> convertSendAndReceive(String routingKey, Object object,
 	 * @param object the object to convert.
 	 * @param messagePostProcessor the post processor.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object,
+	<C> CompletableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object,
 			MessagePostProcessor messagePostProcessor);
 
 	/**
@@ -130,9 +131,9 @@ <C> ListenableFuture<C> convertSendAndReceive(String exchange, String routingKey
 	 * @param object the object to convert.
 	 * @param responseType the response type.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceiveAsType(Object object, ParameterizedTypeReference<C> responseType);
+	<C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, ParameterizedTypeReference<C> responseType);
 
 	/**
 	 * Convert the object to a message and send it to the default exchange with the
@@ -141,9 +142,9 @@ <C> ListenableFuture<C> convertSendAndReceive(String exchange, String routingKey
 	 * @param object the object to convert.
 	 * @param responseType the response type.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
+	<C> CompletableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
 			ParameterizedTypeReference<C> responseType);
 
 	/**
@@ -154,9 +155,9 @@ <C> ListenableFuture<C> convertSendAndReceiveAsType(String routingKey, Object ob
 	 * @param object the object to convert.
 	 * @param responseType the response type.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
+	<C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
 			ParameterizedTypeReference<C> responseType);
 
 	/**
@@ -167,9 +168,9 @@ <C> ListenableFuture<C> convertSendAndReceiveAsType(String exchange, String rout
 	 * @param messagePostProcessor the post processor.
 	 * @param responseType the response type.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor,
+	<C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor,
 			ParameterizedTypeReference<C> responseType);
 
 	/**
@@ -181,9 +182,9 @@ <C> ListenableFuture<C> convertSendAndReceiveAsType(Object object, MessagePostPr
 	 * @param messagePostProcessor the post processor.
 	 * @param responseType the response type.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
+	<C> CompletableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
 			MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType);
 
 	/**
@@ -196,9 +197,9 @@ <C> ListenableFuture<C> convertSendAndReceiveAsType(String routingKey, Object ob
 	 * @param messagePostProcessor the post processor.
 	 * @param responseType the response type.
 	 * @param <C> the expected result type.
-	 * @return the {@link ListenableFuture}.
+	 * @return the {@link CompletableFuture}.
 	 */
-	<C> ListenableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
+	<C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
 			MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType);
 
 }