GH-1480: Switch to CompletableFuture in s-r-stream

Resolves #1480

Also reinstate deprecated `AsyncAmqpTemplate2`.

Author: garyrussell

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.core;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * This interface was added in 2.4.7 to aid migration from methods returning
+ * {@code ListenableFuture}s to {@link CompletableFuture}s.
+ *
+ * @author Gary Russell
+ * @since 2.4.7
+ * @deprecated in favor of {@link AsyncAmqpTemplate}.
+ */
+@Deprecated
+public interface AsyncAmqpTemplate2 extends AsyncAmqpTemplate {
+
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamOperations.java

@@ -16,13 +16,14 @@
 
 package org.springframework.rabbit.stream.producer;
 
+import java.util.concurrent.CompletableFuture;
+
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.lang.Nullable;
 import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
-import org.springframework.util.concurrent.ListenableFuture;
 
 import com.rabbitmq.stream.MessageBuilder;
 
@@ -40,14 +41,14 @@ public interface RabbitStreamOperations extends AutoCloseable {
 	 * @param message the message.
 	 * @return a future to indicate success/failure.
 	 */
-	ListenableFuture<Boolean> send(Message message);
+	CompletableFuture<Boolean> send(Message message);
 
 	/**
 	 * Convert to and send a Spring AMQP message.
 	 * @param message the payload.
 	 * @return a future to indicate success/failure.
 	 */
-	ListenableFuture<Boolean> convertAndSend(Object message);
+	CompletableFuture<Boolean> convertAndSend(Object message);
 
 	/**
 	 * Convert to and send a Spring AMQP message. If a {@link MessagePostProcessor} is
@@ -57,15 +58,15 @@ public interface RabbitStreamOperations extends AutoCloseable {
 	 * @param mpp a message post processor.
 	 * @return a future to indicate success/failure.
 	 */
-	ListenableFuture<Boolean> convertAndSend(Object message, @Nullable MessagePostProcessor mpp);
+	CompletableFuture<Boolean> convertAndSend(Object message, @Nullable MessagePostProcessor mpp);
 
 	/**
 	 * Send a native stream message.
 	 * @param message the message.
 	 * @return a future to indicate success/failure.
 	 * @see #messageBuilder()
 	 */
-	ListenableFuture<Boolean> send(com.rabbitmq.stream.Message message);
+	CompletableFuture<Boolean> send(com.rabbitmq.stream.Message message);
 
 	/**
 	 * Return the producer's {@link MessageBuilder} to create native stream messages.

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamOperations2.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.rabbit.stream.producer;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Provides methods for sending messages using a RabbitMQ Stream producer,
+ * returning {@link CompletableFuture}.
+ * This interface was added in 2.4.7 to aid migration from methods returning
+ * {@code ListenableFuture}s to {@link CompletableFuture}s.
+ *
+ * @author Gary Russell
+ * @since 2.4.7
+ * @deprecated in favor of {@link RabbitStreamOperations}.
+ */
+@Deprecated
+public interface RabbitStreamOperations2 extends RabbitStreamOperations {
+
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-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,6 +16,8 @@
 
 package org.springframework.rabbit.stream.producer;
 
+import java.util.concurrent.CompletableFuture;
+
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.support.converter.MessageConverter;
@@ -27,8 +29,6 @@
 import org.springframework.rabbit.stream.support.converter.DefaultStreamMessageConverter;
 import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
 import org.springframework.util.Assert;
-import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.util.concurrent.SettableListenableFuture;
 
 import com.rabbitmq.stream.ConfirmationHandler;
 import com.rabbitmq.stream.Constants;
@@ -138,27 +138,27 @@ public StreamMessageConverter streamMessageConverter() {
 
 
 	@Override
-	public ListenableFuture<Boolean> send(Message message) {
-		SettableListenableFuture<Boolean> future = new SettableListenableFuture<>();
+	public CompletableFuture<Boolean> send(Message message) {
+		CompletableFuture<Boolean> future = new CompletableFuture<>();
 		createOrGetProducer().send(this.streamConverter.fromMessage(message), handleConfirm(future));
 		return future;
 	}
 
 	@Override
-	public ListenableFuture<Boolean> convertAndSend(Object message) {
+	public CompletableFuture<Boolean> convertAndSend(Object message) {
 		return convertAndSend(message, null);
 	}
 
 	@Override
-	public ListenableFuture<Boolean> convertAndSend(Object message, @Nullable MessagePostProcessor mpp) {
+	public CompletableFuture<Boolean> convertAndSend(Object message, @Nullable MessagePostProcessor mpp) {
 		Message message2 = this.messageConverter.toMessage(message, new StreamMessageProperties());
 		Assert.notNull(message2, "The message converter returned null");
 		if (mpp != null) {
 			message2 = mpp.postProcessMessage(message2);
 			if (message2 == null) {
 				this.logger.debug("Message Post Processor returned null, message not sent");
-				SettableListenableFuture<Boolean> future = new SettableListenableFuture<>();
-				future.set(false);
+				CompletableFuture<Boolean> future = new CompletableFuture<>();
+				future.complete(false);
 				return future;
 			}
 		}
@@ -167,8 +167,8 @@ public ListenableFuture<Boolean> convertAndSend(Object message, @Nullable Messag
 
 
 	@Override
-	public ListenableFuture<Boolean> send(com.rabbitmq.stream.Message message) {
-		SettableListenableFuture<Boolean> future = new SettableListenableFuture<>();
+	public CompletableFuture<Boolean> send(com.rabbitmq.stream.Message message) {
+		CompletableFuture<Boolean> future = new CompletableFuture<>();
 		createOrGetProducer().send(message, handleConfirm(future));
 		return future;
 	}
@@ -178,10 +178,10 @@ public MessageBuilder messageBuilder() {
 		return createOrGetProducer().messageBuilder();
 	}
 
-	private ConfirmationHandler handleConfirm(SettableListenableFuture<Boolean> future) {
+	private ConfirmationHandler handleConfirm(CompletableFuture<Boolean> future) {
 		return confStatus -> {
 			if (confStatus.isConfirmed()) {
-				future.set(true);
+				future.complete(true);
 			}
 			else {
 				int code = confStatus.getCode();
@@ -203,7 +203,7 @@ private ConfirmationHandler handleConfirm(SettableListenableFuture<Boolean> futu
 					errorMessage = "Unknown code: " + code;
 					break;
 				}
-				future.setException(new StreamSendException(errorMessage, code));
+				future.completeExceptionally(new StreamSendException(errorMessage, code));
 			}
 		};
 	}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate2.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.rabbit.stream.producer;
+
+import java.util.concurrent.CompletableFuture;
+
+import com.rabbitmq.stream.Environment;
+
+/**
+ * This interface was added in 2.4.7 to aid migration from methods returning
+ * {@code ListenableFuture}s to {@link CompletableFuture}s.
+ *
+ * @author Gary Russell
+ * @since 2.8
+ * @deprecated in favor of {@link RabbitStreamTemplate}.
+ */
+@Deprecated
+public class RabbitStreamTemplate2 extends RabbitStreamTemplate implements RabbitStreamOperations2 {
+
+	public RabbitStreamTemplate2(Environment environment, String streamName) {
+		super(environment, streamName);
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2.java

@@ -18,6 +18,7 @@
 
 import java.util.concurrent.CompletableFuture;
 
+import org.springframework.amqp.core.AsyncAmqpTemplate2;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
@@ -32,7 +33,7 @@
  *
  */
 @Deprecated
-public class AsyncRabbitTemplate2 extends AsyncRabbitTemplate {
+public class AsyncRabbitTemplate2 extends AsyncRabbitTemplate implements AsyncAmqpTemplate2 {
 
 	public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey,
 			String replyQueue, String replyAddress) {

src/reference/asciidoc/appendix.adoc

@@ -44,6 +44,11 @@ See <<async-template>> for more information.
 The `Jackson2JsonMessageConverter` can now determine the charset from the `contentEncoding` header.
 See <<json-message-converter>> for more information.
 
+==== Stream Support Changes
+
+`RabbitStreamOperations` and `RabbitStreamTemplate` have been deprecated in favor of `RabbitStreamOperations2` and `RabbitStreamTemplate2` respectively; they return `CompletableFuture` instead of `ListenableFuture`.
+See <<stream-support>> for more information.
+
 ==== Changes in 2.3 Since 2.2
 
 This section describes the changes between version 2.2 and version 2.3.

src/reference/asciidoc/stream.adoc

@@ -67,6 +67,8 @@ The `ProducerCustomizer` provides a mechanism to customize the producer before i
 
 Refer to the https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/[Java Client Documentation] about customizing the `Environment` and `Producer`.
 
+IMPORTANT: In version 2.4.7 `RabbitStreamOperations2` and `RabbitStreamTemplate2` were added to assist migration to this version; `RabbitStreamOperations2` and `RabbitStreamTemplate2` are now deprecated in favor of `RabbitStreamOperations` and `RabbitStreamTemplate` respectively.
+
 ==== Receiving Messages
 
 Asynchronous message reception is provided by the `StreamListenerContainer` (and the `StreamRabbitListenerContainerFactory` when using `@RabbitListener`).

src/reference/asciidoc/whats-new.adoc

@@ -16,3 +16,8 @@ The remoting feature (using RMI) is no longer supported.
 The `AsyncRabbitTemplate2`, which was added in 2.4.7 to aid migration to this release, is deprecated in favor of `AsyncRabbitTemplate`.
 The `AsyncRabbitTemplate` now returns `CompletableFuture` s instead of `ListenableFuture` s.
 See <<async-template>> for more information.
+
+==== Stream Support Changes
+
+`RabbitStreamOperations2` and `RabbitStreamTemplate2` have been deprecated in favor of `RabbitStreamOperations` and `RabbitStreamTemplate` respectively.
+See <<stream-support>> for more information.