GH-1473: Move RabbitFutures to Top Level Classes

- to aid migration from 2.4.x to 3.0.x so that the return types will not change

* Increase test coverage.
* Fix since.
* Add author to new files.

Author: garyrussell

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

@@ -0,0 +1,54 @@
+/*
+ * 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.rabbit;
+
+import java.util.concurrent.ScheduledFuture;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
+import org.springframework.core.ParameterizedTypeReference;
+
+/**
+ * A {@link RabbitFuture} with a return type of the template's
+ * generic parameter.
+ * @param <C> the type.
+ *
+ * @author Gary Russell
+ * @since 2.4.7
+ */
+public class RabbitConverterFuture<C> extends RabbitFuture<C> {
+
+	private volatile ParameterizedTypeReference<C> returnType;
+
+	RabbitConverterFuture(String correlationId, Message requestMessage,
+			BiConsumer<String, ChannelHolder> canceler,
+			Function<RabbitFuture<?>, ScheduledFuture<?>> timeoutTaskFunction) {
+
+		super(correlationId, requestMessage, canceler, timeoutTaskFunction);
+	}
+
+	public ParameterizedTypeReference<C> getReturnType() {
+		return this.returnType;
+	}
+
+	public void setReturnType(ParameterizedTypeReference<C> returnType) {
+		this.returnType = returnType;
+	}
+
+}

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

@@ -0,0 +1,116 @@
+/*
+ * 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.rabbit;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledFuture;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
+
+/**
+ * Base class for {@link CompletableFuture}s returned by {@link AsyncRabbitTemplate2}.
+ * @param <T> the type.
+ *
+ * @author Gary Russell
+ * @since 2.4.7
+ */
+public abstract class RabbitFuture<T> extends CompletableFuture<T> {
+
+	private final String correlationId;
+
+	private final Message requestMessage;
+
+	private final BiConsumer<String, ChannelHolder> canceler;
+
+	private final Function<RabbitFuture<?>, ScheduledFuture<?>> timeoutTaskFunction;
+
+	private ScheduledFuture<?> timeoutTask;
+
+	private volatile CompletableFuture<Boolean> confirm;
+
+	private String nackCause;
+
+	private ChannelHolder channelHolder;
+
+	protected RabbitFuture(String correlationId, Message requestMessage, BiConsumer<String, ChannelHolder> canceler,
+			Function<RabbitFuture<?>, ScheduledFuture<?>> timeoutTaskFunction) {
+
+		this.correlationId = correlationId;
+		this.requestMessage = requestMessage;
+		this.canceler = canceler;
+		this.timeoutTaskFunction = timeoutTaskFunction;
+	}
+
+	void setChannelHolder(ChannelHolder channel) {
+		this.channelHolder = channel;
+	}
+
+	String getCorrelationId() {
+		return this.correlationId;
+	}
+
+	ChannelHolder getChannelHolder() {
+		return this.channelHolder;
+	}
+
+	Message getRequestMessage() {
+		return this.requestMessage;
+	}
+
+	@Override
+	public boolean cancel(boolean mayInterruptIfRunning) {
+		if (this.timeoutTask != null) {
+			this.timeoutTask.cancel(true);
+		}
+		this.canceler.accept(this.correlationId, this.channelHolder);
+		return super.cancel(mayInterruptIfRunning);
+	}
+
+	/**
+	 * When confirms are enabled contains a {@link CompletableFuture}
+	 * for the confirmation.
+	 * @return the future.
+	 */
+	public CompletableFuture<Boolean> getConfirm() {
+		return this.confirm;
+	}
+
+	void setConfirm(CompletableFuture<Boolean> confirm) {
+		this.confirm = confirm;
+	}
+
+	/**
+	 * When confirms are enabled and a nack is received, contains
+	 * the cause for the nack, if any.
+	 * @return the cause.
+	 */
+	public String getNackCause() {
+		return this.nackCause;
+	}
+
+	void setNackCause(String nackCause) {
+		this.nackCause = nackCause;
+	}
+
+	void startTimer() {
+		this.timeoutTask = this.timeoutTaskFunction.apply(this);
+	}
+
+}

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

@@ -0,0 +1,40 @@
+/*
+ * 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.rabbit;
+
+import java.util.concurrent.ScheduledFuture;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
+
+/**
+ * A {@link RabbitFuture} with a return type of {@link Message}.
+ *
+ * @author Gary Russell
+ * @since 2.4.7
+ */
+public class RabbitMessageFuture extends RabbitFuture<Message> {
+
+	RabbitMessageFuture(String correlationId, Message requestMessage, BiConsumer<String, ChannelHolder> canceler,
+			Function<RabbitFuture<?>, ScheduledFuture<?>> timeoutTaskFunction) {
+
+		super(correlationId, requestMessage, canceler, timeoutTaskFunction);
+	}
+
+}

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

@@ -0,0 +1,59 @@
+/*
+ * 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.rabbit;
+
+import java.util.concurrent.ConcurrentMap;
+
+import org.springframework.amqp.core.AmqpReplyTimeoutException;
+import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
+import org.springframework.lang.Nullable;
+
+/**
+ * A {@link Runnable} used to time out a {@link RabbitFuture}.
+ *
+ * @author Gary Russell
+ * @since 2.4.7
+ */
+public class TimeoutTask implements Runnable {
+
+	private final RabbitFuture<?> future;
+
+	private final ConcurrentMap<String, RabbitFuture<?>> pending;
+
+	private final DirectReplyToMessageListenerContainer container;
+
+	TimeoutTask(RabbitFuture<?> future, ConcurrentMap<String, RabbitFuture<?>> pending,
+			@Nullable DirectReplyToMessageListenerContainer container) {
+
+		this.future = future;
+		this.pending = pending;
+		this.container = container;
+	}
+
+	@Override
+	public void run() {
+		this.pending.remove(this.future.getCorrelationId());
+		ChannelHolder holder = this.future.getChannelHolder();
+		if (holder != null && this.container != null) {
+			this.container.releaseConsumerFor(holder, false, null); // NOSONAR
+		}
+		this.future.completeExceptionally(
+				new AmqpReplyTimeoutException("Reply timed out", this.future.getRequestMessage()));
+	}
+
+}