From 439ccd174c2da332628780c0f30502004d06f9f8 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 27 Jul 2022 16:14:28 -0400 Subject: [PATCH] GH-1473: Switch to CompletableFuture Resolves https://github.com/spring-projects/spring-amqp/issues/1473 Given the stability of the project, it was simplest to copy the `AsyncRabbitTemplate` rather than adding a lot of conditional code. **2.4.x only; I will submit a separate PR for main** --- .../amqp/core/AsyncAmqpTemplate2.java | 205 +++++ .../amqp/rabbit/AsyncRabbitTemplate.java | 4 +- .../amqp/rabbit/AsyncRabbitTemplate2.java | 857 ++++++++++++++++++ .../rabbit/connection/CorrelationData.java | 16 +- .../PublisherCallbackChannelImpl.java | 4 +- .../AbstractAdaptableMessageListener.java | 18 +- .../adapter/DelegatingInvocableHandler.java | 6 +- .../listener/adapter/HandlerAdapter.java | 6 +- ...RepublishMessageRecovererWithConfirms.java | 3 +- .../rabbit/AsyncRabbitTemplate2Tests.java | 536 +++++++++++ .../amqp/rabbit/AsyncRabbitTemplateTests.java | 17 +- .../rabbit/annotation/AsyncListenerTests.java | 51 +- .../ComplexTypeJsonIntegrationTests.java | 14 +- .../core/MessagingTemplateConfirmsTests.java | 6 +- ...atePublisherCallbacksIntegrationTests.java | 10 +- ...tePublisherCallbacksIntegrationTests2.java | 6 +- ...tingConnectionFactoryIntegrationTests.java | 2 +- .../rabbit/listener/AsyncReplyToTests.java | 11 +- .../adapter/MessageListenerAdapterTests.java | 13 +- src/reference/asciidoc/amqp.adoc | 4 + src/reference/asciidoc/whats-new.adoc | 8 + 21 files changed, 1723 insertions(+), 74 deletions(-) create mode 100644 spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate2.java create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2.java create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2Tests.java diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate2.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate2.java new file mode 100644 index 00000000..3d755305 --- /dev/null +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate2.java @@ -0,0 +1,205 @@ +/* + * Copyright 2020 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; + +import org.springframework.core.ParameterizedTypeReference; + +/** + * Classes implementing this interface can perform asynchronous send and + * receive operations using {@link CompletableFuture}s. + * + * @author Gary Russell + * @since 2.4.7 + * + */ +public interface AsyncAmqpTemplate2 { + + /** + * 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 CompletableFuture}. + */ + CompletableFuture 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 CompletableFuture}. + */ + CompletableFuture 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 CompletableFuture}. + */ + CompletableFuture 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 the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture 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 the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceive(String routingKey, Object object); + + /** + * Convert the object to a message and send it to the provided exchange and + * routing key. + * @param exchange the exchange. + * @param routingKey the routing key. + * @param object the object to convert. + * @param the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceive(String exchange, String routingKey, Object object); + + /** + * Convert the object to a message and send it to the default exchange with the + * default routing key after invoking the {@link MessagePostProcessor}. + * If the post processor adds a correlationId property, it must be unique. + * @param object the object to convert. + * @param messagePostProcessor the post processor. + * @param the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor); + + /** + * Convert the object to a message and send it to the default exchange with the + * provided routing key after invoking the {@link MessagePostProcessor}. + * If the post processor adds a correlationId property, it must be unique. + * @param routingKey the routing key. + * @param object the object to convert. + * @param messagePostProcessor the post processor. + * @param the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceive(String routingKey, Object object, + MessagePostProcessor messagePostProcessor); + + /** + * Convert the object to a message and send it to the provided exchange and + * routing key after invoking the {@link MessagePostProcessor}. + * If the post processor adds a correlationId property, it must be unique. + * @param exchange the exchange + * @param routingKey the routing key. + * @param object the object to convert. + * @param messagePostProcessor the post processor. + * @param the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceive(String exchange, String routingKey, Object object, + MessagePostProcessor messagePostProcessor); + + /** + * 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 responseType the response type. + * @param the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceiveAsType(Object object, ParameterizedTypeReference responseType); + + /** + * 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 responseType the response type. + * @param the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceiveAsType(String routingKey, Object object, + ParameterizedTypeReference responseType); + + /** + * Convert the object to a message and send it to the provided exchange and + * routing key. + * @param exchange the exchange. + * @param routingKey the routing key. + * @param object the object to convert. + * @param responseType the response type. + * @param the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, + ParameterizedTypeReference responseType); + + /** + * Convert the object to a message and send it to the default exchange with the + * default routing key after invoking the {@link MessagePostProcessor}. + * If the post processor adds a correlationId property, it must be unique. + * @param object the object to convert. + * @param messagePostProcessor the post processor. + * @param responseType the response type. + * @param the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor, + ParameterizedTypeReference responseType); + + /** + * Convert the object to a message and send it to the default exchange with the + * provided routing key after invoking the {@link MessagePostProcessor}. + * If the post processor adds a correlationId property, it must be unique. + * @param routingKey the routing key. + * @param object the object to convert. + * @param messagePostProcessor the post processor. + * @param responseType the response type. + * @param the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceiveAsType(String routingKey, Object object, + MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType); + + /** + * Convert the object to a message and send it to the provided exchange and + * routing key after invoking the {@link MessagePostProcessor}. + * If the post processor adds a correlationId property, it must be unique. + * @param exchange the exchange + * @param routingKey the routing key. + * @param object the object to convert. + * @param messagePostProcessor the post processor. + * @param responseType the response type. + * @param the expected result type. + * @return the {@link CompletableFuture}. + */ + CompletableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, + MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType); + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java index 21b08d00..a45ff16f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-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. @@ -89,7 +89,9 @@ import com.rabbitmq.client.Channel; * @author Artem Bilan * * @since 1.6 + * @deprecated in favor of {@link AsyncRabbitTemplate2}. */ +@Deprecated public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessageListener, ReturnsCallback, ConfirmCallback, BeanNameAware, SmartLifecycle { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2.java new file mode 100644 index 00000000..8d4fb2c0 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2.java @@ -0,0 +1,857 @@ +/* + * 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.Date; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ScheduledFuture; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.AmqpIllegalStateException; +import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.AmqpMessageReturnedException; +import org.springframework.amqp.core.AmqpReplyTimeoutException; +import org.springframework.amqp.core.AsyncAmqpTemplate2; +import org.springframework.amqp.core.Correlation; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessagePostProcessor; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.core.ReturnedMessage; +import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitFuture; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.springframework.amqp.rabbit.connection.PublisherCallbackChannel; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback; +import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnsCallback; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.amqp.support.converter.SmartMessageConverter; +import org.springframework.amqp.utils.JavaUtils; +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.context.SmartLifecycle; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.expression.Expression; +import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +import com.rabbitmq.client.Channel; + +/** + * Provides asynchronous send and receive operations returning a {@link CompletableFuture} + * allowing the caller to obtain the reply later, using {@code get()} or a callback. + *

+ * When confirms are enabled, the future has a confirm property which is itself a + * {@link CompletableFuture}. If the reply is received before the publisher confirm, + * the confirm is discarded since the reply implicitly indicates the message was + * published. + *

+ * Returned (undeliverable) request messages are presented as a + * {@link AmqpMessageReturnedException} cause of an + * {@link java.util.concurrent.ExecutionException}. + *

+ * Internally, the template uses a {@link RabbitTemplate} and an + * {@link AbstractMessageListenerContainer} either provided or constructed internally + * (a {@link SimpleMessageListenerContainer}). + * If an external {@link RabbitTemplate} is provided and confirms/returns are enabled, + * it must not previously have had callbacks registered because this object needs to + * be the callback. + * + * @author Gary Russell + * @author Artem Bilan + * + * @since 1.6 + */ +public class AsyncRabbitTemplate2 implements AsyncAmqpTemplate2, ChannelAwareMessageListener, ReturnsCallback, + ConfirmCallback, BeanNameAware, SmartLifecycle { + + public static final int DEFAULT_RECEIVE_TIMEOUT = 30000; + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final RabbitTemplate template; + + private final AbstractMessageListenerContainer container; + + private final DirectReplyToMessageListenerContainer directReplyToContainer; + + private final String replyAddress; + + private final ConcurrentMap> pending = new ConcurrentHashMap<>(); + + private final CorrelationMessagePostProcessor messagePostProcessor = new CorrelationMessagePostProcessor<>(); + + private volatile boolean running; + + private volatile boolean enableConfirms; + + private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; + + private int phase; + + private boolean autoStartup = true; + + private String beanName; + + private TaskScheduler taskScheduler; + + private boolean internalTaskScheduler = true; + + /** + * Construct an instance using the provided arguments. Replies will be + * routed to the default exchange using the reply queue name as the routing + * key. + * @param connectionFactory the connection factory. + * @param exchange the default exchange to which requests will be sent. + * @param routingKey the default routing key. + * @param replyQueue the name of the reply queue to listen for replies. + */ + public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey, + String replyQueue) { + this(connectionFactory, exchange, routingKey, replyQueue, null); + } + + /** + * Construct an instance using the provided arguments. If 'replyAddress' is null, + * replies will be routed to the default exchange using the reply queue name as the + * routing key. Otherwise it should have the form exchange/routingKey and must + * cause messages to be routed to the reply queue. + * @param connectionFactory the connection factory. + * @param exchange the default exchange to which requests will be sent. + * @param routingKey the default routing key. + * @param replyQueue the name of the reply queue to listen for replies. + * @param replyAddress the reply address (exchange/routingKey). + */ + public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey, + String replyQueue, String replyAddress) { + Assert.notNull(connectionFactory, "'connectionFactory' cannot be null"); + Assert.notNull(routingKey, "'routingKey' cannot be null"); + Assert.notNull(replyQueue, "'replyQueue' cannot be null"); + this.template = new RabbitTemplate(connectionFactory); + this.template.setExchange(exchange == null ? "" : exchange); + this.template.setRoutingKey(routingKey); + this.container = new SimpleMessageListenerContainer(connectionFactory); + JavaUtils.INSTANCE + .acceptIfNotNull(this.template.getAfterReceivePostProcessors(), + (value) -> this.container.setAfterReceivePostProcessors( + value.toArray(new MessagePostProcessor[0]))); + this.container.setQueueNames(replyQueue); + this.container.setMessageListener(this); + this.container.afterPropertiesSet(); + this.directReplyToContainer = null; + if (replyAddress == null) { + this.replyAddress = replyQueue; + } + else { + this.replyAddress = replyAddress; + } + + } + + /** + * Construct an instance using the provided arguments. The first queue the container + * is configured to listen to will be used as the reply queue. Replies will be + * routed using the default exchange with that queue name as the routing key. + * @param template a {@link RabbitTemplate} + * @param container a {@link AbstractMessageListenerContainer}. + */ + public AsyncRabbitTemplate2(RabbitTemplate template, AbstractMessageListenerContainer container) { + this(template, container, null); + } + + /** + * Construct an instance using the provided arguments. The first queue the container + * is configured to listen to will be used as the reply queue. If 'replyAddress' is + * null, replies will be routed using the default exchange with that queue name as the + * routing key. Otherwise it should have the form exchange/routingKey and must + * cause messages to be routed to the reply queue. + * @param template a {@link RabbitTemplate}. + * @param container a {@link AbstractMessageListenerContainer}. + * @param replyAddress the reply address. + */ + public AsyncRabbitTemplate2(RabbitTemplate template, AbstractMessageListenerContainer container, + String replyAddress) { + Assert.notNull(template, "'template' cannot be null"); + Assert.notNull(container, "'container' cannot be null"); + this.template = template; + this.container = container; + this.container.setMessageListener(this); + this.directReplyToContainer = null; + if (replyAddress == null) { + this.replyAddress = container.getQueueNames()[0]; + } + else { + this.replyAddress = replyAddress; + } + } + + /** + * Construct an instance using the provided arguments. "Direct replyTo" is used for + * replies. + * @param connectionFactory the connection factory. + * @param exchange the default exchange to which requests will be sent. + * @param routingKey the default routing key. + * @since 2.0 + */ + public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey) { + this(new RabbitTemplate(connectionFactory)); + Assert.notNull(routingKey, "'routingKey' cannot be null"); + this.template.setExchange(exchange == null ? "" : exchange); + this.template.setRoutingKey(routingKey); + } + + /** + * Construct an instance using the provided arguments. "Direct replyTo" is used for + * replies. + * @param template a {@link RabbitTemplate} + * @since 2.0 + */ + public AsyncRabbitTemplate2(RabbitTemplate template) { + Assert.notNull(template, "'template' cannot be null"); + this.template = template; + this.container = null; + this.replyAddress = null; + this.directReplyToContainer = new DirectReplyToMessageListenerContainer(this.template.getConnectionFactory()); + JavaUtils.INSTANCE + .acceptIfNotNull(template.getAfterReceivePostProcessors(), + (value) -> this.directReplyToContainer.setAfterReceivePostProcessors( + value.toArray(new MessagePostProcessor[0]))); + this.directReplyToContainer.setMessageListener(this); + } + + /** + * @param autoStartup true for auto start. + * @see #isAutoStartup() + */ + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + /** + * @param phase the phase. + * @see #getPhase() + */ + public void setPhase(int phase) { + this.phase = phase; + } + + /** + * Set to true to enable the receipt of returned messages that cannot be delivered + * in the form of a {@link AmqpMessageReturnedException}. + * @param mandatory true to enable returns. + */ + public void setMandatory(boolean mandatory) { + this.template.setReturnsCallback(this); + this.template.setMandatory(mandatory); + } + + /** + * @param mandatoryExpression a SpEL {@link Expression} to evaluate against each request + * message. The result of the evaluation must be a {@code boolean} value. + * @since 2.0 + */ + public void setMandatoryExpression(Expression mandatoryExpression) { + this.template.setReturnsCallback(this); + this.template.setMandatoryExpression(mandatoryExpression); + } + + /** + * @param mandatoryExpression a SpEL {@link Expression} to evaluate against each request + * message. The result of the evaluation must be a {@code boolean} value. + * @since 2.0 + */ + public void setMandatoryExpressionString(String mandatoryExpression) { + this.template.setReturnsCallback(this); + this.template.setMandatoryExpressionString(mandatoryExpression); + } + + /** + * Set to true to enable publisher confirms. When enabled, the {@link RabbitFuture} + * returned by the send and receive operation will have a + * {@code CompletableFuture} in its {@code confirm} property. + * @param enableConfirms true to enable publisher confirms. + */ + public void setEnableConfirms(boolean enableConfirms) { + this.enableConfirms = enableConfirms; + if (enableConfirms) { + this.template.setConfirmCallback(this); + } + } + + public String getBeanName() { + return this.beanName; + } + + @Override + public void setBeanName(String beanName) { + this.beanName = beanName; + } + + /** + * @return a reference to the underlying connection factory in the + * {@link RabbitTemplate}. + */ + public ConnectionFactory getConnectionFactory() { + return this.template.getConnectionFactory(); + } + + /** + * Set the receive timeout - the future returned by the send and receive + * methods will be canceled when this timeout expires. {@code <= 0} means + * futures never expire. Beware that this will cause a memory leak if a + * reply is not received. Default: 30000 (30 seconds). + * @param receiveTimeout the timeout in milliseconds. + */ + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + /** + * Set the task scheduler to expire timed out futures. + * @param taskScheduler the task scheduler + * @see #setReceiveTimeout(long) + */ + public synchronized void setTaskScheduler(TaskScheduler taskScheduler) { + Assert.notNull(taskScheduler, "'taskScheduler' cannot be null"); + this.internalTaskScheduler = false; + this.taskScheduler = taskScheduler; + } + + /** + * @return a reference to the underlying {@link RabbitTemplate}'s + * {@link MessageConverter}. + */ + public MessageConverter getMessageConverter() { + return this.template.getMessageConverter(); + } + + /** + * Return the underlying {@link RabbitTemplate} used for sending. + * @return the template. + * @since 2.2 + */ + public RabbitTemplate getRabbitTemplate() { + return this.template; + } + + @Override + public RabbitMessageFuture2 sendAndReceive(Message message) { + return sendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), message); + } + + @Override + public RabbitMessageFuture2 sendAndReceive(String routingKey, Message message) { + return sendAndReceive(this.template.getExchange(), routingKey, message); + } + + @Override + public RabbitMessageFuture2 sendAndReceive(String exchange, String routingKey, Message message) { + String correlationId = getOrSetCorrelationIdAndSetReplyTo(message, null); + RabbitMessageFuture2 future = new RabbitMessageFuture2(correlationId, message); + CorrelationData correlationData = null; + if (this.enableConfirms) { + correlationData = new CorrelationData(correlationId); + future.setConfirm(new CompletableFuture<>()); + } + this.pending.put(correlationId, future); + if (this.container != null) { + this.template.send(exchange, routingKey, message, correlationData); + } + else { + ChannelHolder channelHolder = this.directReplyToContainer.getChannelHolder(); + future.setChannelHolder(channelHolder); + sendDirect(channelHolder.getChannel(), exchange, routingKey, message, correlationData); + } + future.startTimer(); + return future; + } + + @Override + public RabbitConverterFuture2 convertSendAndReceive(Object object) { + return convertSendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), object, null); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceive(String routingKey, Object object) { + return convertSendAndReceive(this.template.getExchange(), routingKey, object, null); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceive(String exchange, String routingKey, Object object) { + return convertSendAndReceive(exchange, routingKey, object, null); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceive(Object object, + MessagePostProcessor messagePostProcessor) { + return convertSendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), object, + messagePostProcessor); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceive(String routingKey, Object object, + MessagePostProcessor messagePostProcessor) { + return convertSendAndReceive(this.template.getExchange(), routingKey, object, messagePostProcessor); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceive(String exchange, String routingKey, Object object, + MessagePostProcessor messagePostProcessor) { + return convertSendAndReceive(exchange, routingKey, object, messagePostProcessor, null); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceiveAsType(Object object, + ParameterizedTypeReference responseType) { + return convertSendAndReceiveAsType(this.template.getExchange(), this.template.getRoutingKey(), object, + null, responseType); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceiveAsType(String routingKey, Object object, + ParameterizedTypeReference responseType) { + return convertSendAndReceiveAsType(this.template.getExchange(), routingKey, object, null, responseType); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceiveAsType(String exchange, String routingKey, Object object, + ParameterizedTypeReference responseType) { + return convertSendAndReceiveAsType(exchange, routingKey, object, null, responseType); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceiveAsType(Object object, + MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { + return convertSendAndReceiveAsType(this.template.getExchange(), this.template.getRoutingKey(), object, + messagePostProcessor, responseType); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceiveAsType(String routingKey, Object object, + MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { + return convertSendAndReceiveAsType(this.template.getExchange(), routingKey, object, messagePostProcessor, + responseType); + } + + @Override + public RabbitConverterFuture2 convertSendAndReceiveAsType(String exchange, String routingKey, Object object, + MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { + Assert.state(this.template.getMessageConverter() instanceof SmartMessageConverter, + "template's message converter must be a SmartMessageConverter"); + return convertSendAndReceive(exchange, routingKey, object, messagePostProcessor, responseType); + } + + private RabbitConverterFuture2 convertSendAndReceive(String exchange, String routingKey, Object object, + MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { + + AsyncCorrelationData correlationData = new AsyncCorrelationData(messagePostProcessor, responseType, + this.enableConfirms); + if (this.container != null) { + this.template.convertAndSend(exchange, routingKey, object, this.messagePostProcessor, correlationData); + } + else { + MessageConverter converter = this.template.getMessageConverter(); + if (converter == null) { + throw new AmqpIllegalStateException( + "No 'messageConverter' specified. Check configuration of RabbitTemplate."); + } + Message message = converter.toMessage(object, new MessageProperties()); + this.messagePostProcessor.postProcessMessage(message, correlationData, + this.template.nullSafeExchange(exchange), this.template.nullSafeRoutingKey(routingKey)); + ChannelHolder channelHolder = this.directReplyToContainer.getChannelHolder(); + correlationData.future.setChannelHolder(channelHolder); + sendDirect(channelHolder.getChannel(), exchange, routingKey, message, correlationData); + } + RabbitConverterFuture2 future = correlationData.future; + future.startTimer(); + return future; + } + + private void sendDirect(Channel channel, String exchange, String routingKey, Message message, + CorrelationData correlationData) { + message.getMessageProperties().setReplyTo(Address.AMQ_RABBITMQ_REPLY_TO); + try { + if (channel instanceof PublisherCallbackChannel) { + this.template.addListener(channel); + } + this.template.doSend(channel, exchange, routingKey, message, this.template.isMandatoryFor(message), + correlationData); + } + catch (Exception e) { + throw new AmqpException("Failed to send request", e); + } + } + + @Override + public synchronized void start() { + if (!this.running) { + if (this.internalTaskScheduler) { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.setThreadNamePrefix(getBeanName() == null ? "asyncTemplate-" : (getBeanName() + "-")); + scheduler.afterPropertiesSet(); + this.taskScheduler = scheduler; + } + if (this.container != null) { + this.container.start(); + } + if (this.directReplyToContainer != null) { + this.directReplyToContainer.setTaskScheduler(this.taskScheduler); + this.directReplyToContainer.start(); + } + } + this.running = true; + } + + @Override + public synchronized void stop() { + if (this.running) { + if (this.container != null) { + this.container.stop(); + } + if (this.directReplyToContainer != null) { + this.directReplyToContainer.stop(); + } + for (RabbitFuture2 future : this.pending.values()) { + future.setNackCause("AsyncRabbitTemplate was stopped while waiting for reply"); + future.cancel(true); + } + if (this.internalTaskScheduler) { + ((ThreadPoolTaskScheduler) this.taskScheduler).destroy(); + this.taskScheduler = null; + } + } + this.running = false; + } + + @Override + public boolean isRunning() { + return this.running; + } + + @Override + public int getPhase() { + return this.phase; + } + + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + + @SuppressWarnings("unchecked") + @Override + public void onMessage(Message message, Channel channel) { + MessageProperties messageProperties = message.getMessageProperties(); + if (messageProperties != null) { + String correlationId = messageProperties.getCorrelationId(); + if (StringUtils.hasText(correlationId)) { + if (this.logger.isDebugEnabled()) { + this.logger.debug("onMessage: " + message); + } + RabbitFuture2 future = this.pending.remove(correlationId); + if (future != null) { + if (future instanceof AsyncRabbitTemplate2.RabbitConverterFuture2) { + MessageConverter messageConverter = this.template.getMessageConverter(); + RabbitConverterFuture2 rabbitFuture = (RabbitConverterFuture2) future; + Object converted = rabbitFuture.getReturnType() != null + && messageConverter instanceof SmartMessageConverter + ? ((SmartMessageConverter) messageConverter).fromMessage(message, + rabbitFuture.getReturnType()) + : messageConverter.fromMessage(message); + rabbitFuture.complete(converted); + } + else { + ((RabbitMessageFuture2) future).complete(message); + } + } + else { + if (this.logger.isWarnEnabled()) { + this.logger.warn("No pending reply - perhaps timed out: " + message); + } + } + } + } + } + + @Override + public void returnedMessage(ReturnedMessage returned) { + MessageProperties messageProperties = returned.getMessage().getMessageProperties(); + String correlationId = messageProperties.getCorrelationId(); + if (StringUtils.hasText(correlationId)) { + RabbitFuture2 future = this.pending.remove(correlationId); + if (future != null) { + future.completeExceptionally(new AmqpMessageReturnedException("Message returned", returned)); + } + else { + if (this.logger.isWarnEnabled()) { + this.logger + .warn("No pending reply - perhaps timed out? Message returned: " + returned.getMessage()); + } + } + } + } + + @Override + public void confirm(@NonNull CorrelationData correlationData, boolean ack, @Nullable String cause) { + if (this.logger.isDebugEnabled()) { + this.logger.debug("Confirm: " + correlationData + ", ack=" + ack + + (cause == null ? "" : (", cause: " + cause))); + } + String correlationId = correlationData.getId(); + if (correlationId != null) { + RabbitFuture2 future = this.pending.get(correlationId); + if (future != null) { + future.setNackCause(cause); + future.getConfirm().complete(ack); + } + else { + if (this.logger.isDebugEnabled()) { + this.logger.debug("Confirm: " + correlationData + ", ack=" + ack + + (cause == null ? "" : (", cause: " + cause)) + + " no pending future - either canceled or the reply is already received"); + } + } + } + } + + private String getOrSetCorrelationIdAndSetReplyTo(Message message, + @Nullable AsyncCorrelationData correlationData) { + + String correlationId; + MessageProperties messageProperties = message.getMessageProperties(); + Assert.notNull(messageProperties, "the message properties cannot be null"); + String currentCorrelationId = messageProperties.getCorrelationId(); + if (!StringUtils.hasText(currentCorrelationId)) { + correlationId = correlationData != null ? correlationData.getId() : UUID.randomUUID().toString(); + messageProperties.setCorrelationId(correlationId); + Assert.isNull(messageProperties.getReplyTo(), "'replyTo' property must be null"); + } + else { + correlationId = currentCorrelationId; + } + messageProperties.setReplyTo(this.replyAddress); + return correlationId; + } + + @Override + public String toString() { + return this.beanName == null ? super.toString() : (this.getClass().getSimpleName() + ": " + this.beanName); + } + + /** + * Base class for {@link CompletableFuture}s returned by {@link AsyncRabbitTemplate2}. + * @param the type. + * @since 1.6 + */ + public abstract class RabbitFuture2 extends CompletableFuture { + + private final String correlationId; + + private final Message requestMessage; + + private ScheduledFuture timeoutTask; + + private volatile CompletableFuture confirm; + + private String nackCause; + + private ChannelHolder channelHolder; + + public RabbitFuture2(String correlationId, Message requestMessage) { + this.correlationId = correlationId; + this.requestMessage = requestMessage; + } + + void setChannelHolder(ChannelHolder channel) { + this.channelHolder = channel; + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + if (this.timeoutTask != null) { + this.timeoutTask.cancel(true); + } + AsyncRabbitTemplate2.this.pending.remove(this.correlationId); + if (this.channelHolder != null && AsyncRabbitTemplate2.this.directReplyToContainer != null) { + AsyncRabbitTemplate2.this.directReplyToContainer + .releaseConsumerFor(this.channelHolder, false, null); // NOSONAR + } + return super.cancel(mayInterruptIfRunning); + } + + /** + * When confirms are enabled contains a {@link CompletableFuture} + * for the confirmation. + * @return the future. + */ + public CompletableFuture getConfirm() { + return this.confirm; + } + + void setConfirm(CompletableFuture 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() { + if (AsyncRabbitTemplate2.this.receiveTimeout > 0) { + synchronized (AsyncRabbitTemplate2.this) { + if (!AsyncRabbitTemplate2.this.running) { + AsyncRabbitTemplate2.this.pending.remove(this.correlationId); + throw new IllegalStateException("'AsyncRabbitTemplate' must be started."); + } + this.timeoutTask = AsyncRabbitTemplate2.this.taskScheduler.schedule(new TimeoutTask(), + new Date(System.currentTimeMillis() + AsyncRabbitTemplate2.this.receiveTimeout)); + } + } + else { + this.timeoutTask = null; + } + } + + private class TimeoutTask implements Runnable { + + @Override + public void run() { + AsyncRabbitTemplate2.this.pending.remove(RabbitFuture2.this.correlationId); + if (RabbitFuture2.this.channelHolder != null + && AsyncRabbitTemplate2.this.directReplyToContainer != null) { + AsyncRabbitTemplate2.this.directReplyToContainer + .releaseConsumerFor(RabbitFuture2.this.channelHolder, false, null); // NOSONAR + } + completeExceptionally( + new AmqpReplyTimeoutException("Reply timed out", RabbitFuture2.this.requestMessage)); + } + + } + + } + + /** + * A {@link RabbitFuture} with a return type of {@link Message}. + * @since 1.6 + */ + public class RabbitMessageFuture2 extends RabbitFuture2 { + + public RabbitMessageFuture2(String correlationId, Message requestMessage) { + super(correlationId, requestMessage); + } + + } + + /** + * A {@link RabbitFuture} with a return type of the template's + * generic parameter. + * @param the type. + * @since 1.6 + */ + public class RabbitConverterFuture2 extends RabbitFuture2 { + + private volatile ParameterizedTypeReference returnType; + + public RabbitConverterFuture2(String correlationId, Message requestMessage) { + super(correlationId, requestMessage); + } + + public ParameterizedTypeReference getReturnType() { + return this.returnType; + } + + public void setReturnType(ParameterizedTypeReference returnType) { + this.returnType = returnType; + } + + } + + private final class CorrelationMessagePostProcessor implements MessagePostProcessor { + + CorrelationMessagePostProcessor() { + } + + @Override + public Message postProcessMessage(Message message) throws AmqpException { + throw new UnsupportedOperationException(); + } + + @SuppressWarnings("unchecked") + @Override + public Message postProcessMessage(Message message, Correlation correlation) throws AmqpException { + Message messageToSend = message; + AsyncCorrelationData correlationData = (AsyncCorrelationData) correlation; + if (correlationData.userPostProcessor != null) { + messageToSend = correlationData.userPostProcessor.postProcessMessage(message); + } + String correlationId = getOrSetCorrelationIdAndSetReplyTo(messageToSend, correlationData); + correlationData.future = new RabbitConverterFuture2(correlationId, message); + if (correlationData.enableConfirms) { + correlationData.setId(correlationId); + correlationData.future.setConfirm(new CompletableFuture<>()); + } + correlationData.future.setReturnType(correlationData.returnType); + AsyncRabbitTemplate2.this.pending.put(correlationId, correlationData.future); + return messageToSend; + } + + } + + private static class AsyncCorrelationData extends CorrelationData { + + final MessagePostProcessor userPostProcessor; // NOSONAR + + final ParameterizedTypeReference returnType; // NOSONAR + + final boolean enableConfirms; // NOSONAR + + volatile RabbitConverterFuture2 future; // NOSONAR + + AsyncCorrelationData(MessagePostProcessor userPostProcessor, ParameterizedTypeReference returnType, + boolean enableConfirms) { + + this.userPostProcessor = userPostProcessor; + this.returnType = returnType; + this.enableConfirms = enableConfirms; + } + + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java index 14162a82..312a523b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-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. @@ -17,6 +17,7 @@ package org.springframework.amqp.rabbit.connection; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import org.springframework.amqp.core.Correlation; import org.springframework.amqp.core.Message; @@ -43,6 +44,8 @@ public class CorrelationData implements Correlation { private final SettableListenableFuture future = new SettableListenableFuture<>(); + private final CompletableFuture completable = this.future.completable(); + private volatile String id; private volatile ReturnedMessage returnedMessage; @@ -90,11 +93,22 @@ public class CorrelationData implements Correlation { * Return a future to check the success/failure of the publish operation. * @return the future. * @since 2.1 + * @deprecated in favor of {@link #getCompletableFuture()}. */ + @Deprecated public SettableListenableFuture getFuture() { return this.future; } + /** + * Return a future to check the success/failure of the publish operation. + * @return the future. + * @since 2.4.7 + */ + public CompletableFuture getCompletableFuture() { + return this.completable; + } + /** * Return a returned message, if any; requires a unique * {@link #CorrelationData(String) id}. Guaranteed to be populated before the future diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java index 01f710b2..4c01b4d5 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-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. @@ -935,6 +935,7 @@ public class PublisherCallbackChannelImpl } } + @SuppressWarnings("deprecation") private void doProcessAck(long seq, boolean ack, boolean multiple, boolean remove) { if (multiple) { processMultipleAck(seq, ack); @@ -971,6 +972,7 @@ public class PublisherCallbackChannelImpl } } + @SuppressWarnings("deprecation") private void processMultipleAck(long seq, boolean ack) { /* * Piggy-backed ack - extract all Listeners for this and earlier diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java index 8746f177..5c0d74bd 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2021 the original author or authors. + * Copyright 2014-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. @@ -21,6 +21,7 @@ import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.lang.reflect.WildcardType; import java.util.Arrays; +import java.util.concurrent.CompletableFuture; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -386,6 +387,21 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe }, t -> asyncFailure(request, channel, t)); } + else if (resultArg.getReturnValue() instanceof CompletableFuture) { + if (!this.isManualAck) { + this.logger.warn("Container AcknowledgeMode must be MANUAL for a Future return type; " + + "otherwise the container will ack the message immediately"); + } + ((CompletableFuture) resultArg.getReturnValue()).whenComplete((r, t) -> { + if (t == null) { + asyncSuccess(resultArg, request, channel, source, r); + basicAck(request, channel); + } + else { + asyncFailure(request, channel, t); + } + }); + } else if (monoPresent && MonoHandler.isMono(resultArg.getReturnValue())) { if (!this.isManualAck) { this.logger.warn("Container AcknowledgeMode must be MANUAL for a Mono return type; " diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java index 96bb30fc..3a663237 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2021 the original author or authors. + * Copyright 2015-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. @@ -23,6 +23,7 @@ import java.util.Arrays; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -147,7 +148,8 @@ public class DelegatingInvocableHandler { private boolean isAsyncReply(InvocableHandlerMethod method) { return (AbstractAdaptableMessageListener.monoPresent && MonoHandler.isMono(method.getMethod().getReturnType())) - || ListenableFuture.class.isAssignableFrom(method.getMethod().getReturnType()); + || ListenableFuture.class.isAssignableFrom(method.getMethod().getReturnType()) + || CompletableFuture.class.isAssignableFrom(method.getMethod().getReturnType()); } /** diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java index 29d3bd45..898cf9a5 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2021 the original author or authors. + * Copyright 2015-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. @@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.listener.adapter; import java.lang.reflect.Method; import java.lang.reflect.Type; +import java.util.concurrent.CompletableFuture; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; @@ -50,7 +51,8 @@ public class HandlerAdapter { this.delegatingHandler = null; this.asyncReplies = (AbstractAdaptableMessageListener.monoPresent && MonoHandler.isMono(invokerHandlerMethod.getMethod().getReturnType())) - || ListenableFuture.class.isAssignableFrom(invokerHandlerMethod.getMethod().getReturnType()); + || ListenableFuture.class.isAssignableFrom(invokerHandlerMethod.getMethod().getReturnType()) + || CompletableFuture.class.isAssignableFrom(invokerHandlerMethod.getMethod().getReturnType()); } /** diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java index 1c65f593..cbdd107b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.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. @@ -108,6 +108,7 @@ public class RepublishMessageRecovererWithConfirms extends RepublishMessageRecov } } + @SuppressWarnings("deprecation") private void doSendCorrelated(String exchange, String routingKey, Message message) { CorrelationData cd = new CorrelationData(); if (exchange != null) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2Tests.java new file mode 100644 index 00000000..695e7d40 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2Tests.java @@ -0,0 +1,536 @@ +/* + * Copyright 2016-2020 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 static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; +import static org.awaitility.Awaitility.await; + +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiConsumer; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.AmqpMessageReturnedException; +import org.springframework.amqp.core.AmqpReplyTimeoutException; +import org.springframework.amqp.core.AnonymousQueue; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.AsyncRabbitTemplate2.RabbitConverterFuture2; +import org.springframework.amqp.rabbit.AsyncRabbitTemplate2.RabbitMessageFuture2; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.junit.RabbitAvailable; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; +import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener; +import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.amqp.support.postprocessor.GUnzipPostProcessor; +import org.springframework.amqp.support.postprocessor.GZipPostProcessor; +import org.springframework.amqp.utils.test.TestUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Primary; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Gary Russell + * @author Artem Bilan + * + * @since 1.6 + */ +@SpringJUnitConfig +@DirtiesContext +@RabbitAvailable +public class AsyncRabbitTemplate2Tests { + + @Autowired + private AsyncRabbitTemplate2 asyncTemplate; + + @Autowired + private AsyncRabbitTemplate2 asyncDirectTemplate; + + @Autowired + private Queue requests; + + @Autowired + private AtomicReference latch; + + private final Message fooMessage = new SimpleMessageConverter().toMessage("foo", new MessageProperties()); + + @Test + public void testConvert1Arg() throws Exception { + final AtomicBoolean mppCalled = new AtomicBoolean(); + CompletableFuture future = this.asyncTemplate.convertSendAndReceive("foo", m -> { + mppCalled.set(true); + return m; + }); + checkConverterResult(future, "FOO"); + assertThat(mppCalled.get()).isTrue(); + } + + @Test + public void testConvert1ArgDirect() throws Exception { + this.latch.set(new CountDownLatch(1)); + CompletableFuture future1 = this.asyncDirectTemplate.convertSendAndReceive("foo"); + CompletableFuture future2 = this.asyncDirectTemplate.convertSendAndReceive("bar"); + this.latch.get().countDown(); + checkConverterResult(future1, "FOO"); + checkConverterResult(future2, "BAR"); + this.latch.set(null); + waitForZeroInUseConsumers(); + assertThat(TestUtils + .getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount", + Integer.class)).isEqualTo(2); + final String missingQueue = UUID.randomUUID().toString(); + this.asyncDirectTemplate.convertSendAndReceive("", missingQueue, "foo"); // send to nowhere + this.asyncDirectTemplate.stop(); // should clear the inUse channel map + waitForZeroInUseConsumers(); + this.asyncDirectTemplate.start(); + this.asyncDirectTemplate.setReceiveTimeout(1); + this.asyncDirectTemplate.convertSendAndReceive("", missingQueue, "foo"); // send to nowhere + waitForZeroInUseConsumers(); + + this.asyncDirectTemplate.setReceiveTimeout(10000); + this.asyncDirectTemplate.convertSendAndReceive("", missingQueue, "foo").cancel(true); + waitForZeroInUseConsumers(); + } + + @Test + public void testConvert2Args() throws Exception { + CompletableFuture future = this.asyncTemplate.convertSendAndReceive(this.requests.getName(), "foo"); + checkConverterResult(future, "FOO"); + } + + @Test + public void testConvert3Args() throws Exception { + CompletableFuture future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo"); + checkConverterResult(future, "FOO"); + } + + @Test + public void testConvert4Args() throws Exception { + CompletableFuture future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo", + message -> { + String body = new String(message.getBody()); + return new Message((body + "bar").getBytes(), message.getMessageProperties()); + }); + checkConverterResult(future, "FOOBAR"); + } + + @Test + public void testMessage1Arg() throws Exception { + CompletableFuture future = this.asyncTemplate.sendAndReceive(getFooMessage()); + checkMessageResult(future, "FOO"); + } + + @Test + public void testMessage1ArgDirect() throws Exception { + this.latch.set(new CountDownLatch(1)); + CompletableFuture future1 = this.asyncDirectTemplate.sendAndReceive(getFooMessage()); + CompletableFuture future2 = this.asyncDirectTemplate.sendAndReceive(getFooMessage()); + this.latch.get().countDown(); + Message reply1 = checkMessageResult(future1, "FOO"); + assertThat(reply1.getMessageProperties().getConsumerQueue()).isEqualTo(Address.AMQ_RABBITMQ_REPLY_TO); + Message reply2 = checkMessageResult(future2, "FOO"); + assertThat(reply2.getMessageProperties().getConsumerQueue()).isEqualTo(Address.AMQ_RABBITMQ_REPLY_TO); + this.latch.set(null); + waitForZeroInUseConsumers(); + assertThat(TestUtils + .getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount", + Integer.class)).isEqualTo(2); + this.asyncDirectTemplate.stop(); + this.asyncDirectTemplate.start(); + assertThat(TestUtils + .getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount", + Integer.class)).isEqualTo(0); + } + + private void waitForZeroInUseConsumers() throws InterruptedException { + Map inUseConsumers = TestUtils + .getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.inUseConsumerChannels", Map.class); + await().until(() -> inUseConsumers.size() == 0); + } + + @Test + public void testMessage2Args() throws Exception { + CompletableFuture future = this.asyncTemplate.sendAndReceive(this.requests.getName(), getFooMessage()); + checkMessageResult(future, "FOO"); + } + + @Test + public void testMessage3Args() throws Exception { + CompletableFuture future = this.asyncTemplate.sendAndReceive("", this.requests.getName(), + getFooMessage()); + checkMessageResult(future, "FOO"); + } + + @SuppressWarnings("unchecked") + @Test + public void testCancel() { + CompletableFuture future = this.asyncTemplate.convertSendAndReceive("foo"); + future.cancel(false); + assertThat(TestUtils.getPropertyValue(asyncTemplate, "pending", Map.class)).hasSize(0); + } + + @Test + public void testMessageCustomCorrelation() throws Exception { + Message message = getFooMessage(); + message.getMessageProperties().setCorrelationId("foo"); + CompletableFuture future = this.asyncTemplate.sendAndReceive(message); + Message result = checkMessageResult(future, "FOO"); + assertThat(result.getMessageProperties().getCorrelationId()).isEqualTo("foo"); + } + + private Message getFooMessage() { + this.fooMessage.getMessageProperties().setCorrelationId(null); + this.fooMessage.getMessageProperties().setReplyTo(null); + return this.fooMessage; + } + + @Test + @DirtiesContext + public void testReturn() throws Exception { + this.asyncTemplate.setMandatory(true); + CompletableFuture future = this.asyncTemplate.convertSendAndReceive(this.requests.getName() + "x", + "foo"); + try { + future.get(10, TimeUnit.SECONDS); + fail("Expected exception"); + } + catch (ExecutionException e) { + assertThat(e.getCause()).isInstanceOf(AmqpMessageReturnedException.class); + assertThat(((AmqpMessageReturnedException) e.getCause()).getRoutingKey()).isEqualTo(this.requests.getName() + "x"); + } + } + + @Test + @DirtiesContext + public void testReturnDirect() throws Exception { + this.asyncDirectTemplate.setMandatory(true); + CompletableFuture future = this.asyncDirectTemplate.convertSendAndReceive(this.requests.getName() + "x", + "foo"); + try { + future.get(10, TimeUnit.SECONDS); + fail("Expected exception"); + } + catch (ExecutionException e) { + assertThat(e.getCause()).isInstanceOf(AmqpMessageReturnedException.class); + assertThat(((AmqpMessageReturnedException) e.getCause()).getRoutingKey()).isEqualTo(this.requests.getName() + "x"); + } + } + + @Test + @DirtiesContext + public void testConvertWithConfirm() throws Exception { + this.asyncTemplate.setEnableConfirms(true); + RabbitConverterFuture2 future = this.asyncTemplate.convertSendAndReceive("sleep"); + CompletableFuture confirm = future.getConfirm(); + assertThat(confirm).isNotNull(); + assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); + checkConverterResult(future, "SLEEP"); + } + + @Test + @DirtiesContext + public void testMessageWithConfirm() throws Exception { + this.asyncTemplate.setEnableConfirms(true); + RabbitMessageFuture2 future = this.asyncTemplate + .sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties())); + CompletableFuture confirm = future.getConfirm(); + assertThat(confirm).isNotNull(); + assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); + checkMessageResult(future, "SLEEP"); + } + + @Test + @DirtiesContext + public void testConvertWithConfirmDirect() throws Exception { + this.asyncDirectTemplate.setEnableConfirms(true); + RabbitConverterFuture2 future = this.asyncDirectTemplate.convertSendAndReceive("sleep"); + CompletableFuture confirm = future.getConfirm(); + assertThat(confirm).isNotNull(); + assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); + checkConverterResult(future, "SLEEP"); + } + + @Test + @DirtiesContext + public void testMessageWithConfirmDirect() throws Exception { + this.asyncDirectTemplate.setEnableConfirms(true); + RabbitMessageFuture2 future = this.asyncDirectTemplate + .sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties())); + CompletableFuture confirm = future.getConfirm(); + assertThat(confirm).isNotNull(); + assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); + checkMessageResult(future, "SLEEP"); + } + + @SuppressWarnings("unchecked") + @Test + @DirtiesContext + public void testReceiveTimeout() throws Exception { + this.asyncTemplate.setReceiveTimeout(500); + CompletableFuture future = this.asyncTemplate.convertSendAndReceive("noReply"); + TheCallback callback = new TheCallback(); + future.whenComplete(callback); + assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1); + try { + future.get(10, TimeUnit.SECONDS); + fail("Expected ExecutionException"); + } + catch (ExecutionException e) { + assertThat(e.getCause()).isInstanceOf(AmqpReplyTimeoutException.class); + } + assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(0); + assertThat(callback.latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(callback.ex).isInstanceOf(AmqpReplyTimeoutException.class); + } + + @SuppressWarnings("unchecked") + @Test + @DirtiesContext + public void testReplyAfterReceiveTimeout() throws Exception { + this.asyncTemplate.setReceiveTimeout(100); + RabbitConverterFuture2 future = this.asyncTemplate.convertSendAndReceive("sleep"); + TheCallback callback = new TheCallback(); + future.whenComplete(callback); + assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1); + try { + future.get(10, TimeUnit.SECONDS); + fail("Expected ExecutionException"); + } + catch (ExecutionException e) { + assertThat(e.getCause()).isInstanceOf(AmqpReplyTimeoutException.class); + } + assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(0); + assertThat(callback.latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(callback.ex).isInstanceOf(AmqpReplyTimeoutException.class); + + /* + * Test there's no harm if the reply is received after the timeout. This + * is unlikely to happen because the future is removed from the pending + * map when it times out. However, there is a small race condition where + * the reply arrives at the same time as the timeout. + */ + future.complete("foo"); + assertThat(callback.result).isNull(); + } + + @SuppressWarnings("unchecked") + @Test + @DirtiesContext + public void testStopCancelled() throws Exception { + this.asyncTemplate.setReceiveTimeout(5000); + RabbitConverterFuture2 future = this.asyncTemplate.convertSendAndReceive("noReply"); + TheCallback callback = new TheCallback(); + future.whenComplete(callback); + assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1); + this.asyncTemplate.stop(); + // Second stop() to be sure that it is idempotent + this.asyncTemplate.stop(); + try { + future.get(10, TimeUnit.SECONDS); + fail("Expected CancellationException"); + } + catch (CancellationException e) { + assertThat(future.getNackCause()).isEqualTo("AsyncRabbitTemplate was stopped while waiting for reply"); + } + assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(0); + assertThat(callback.latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(future.isCancelled()).isTrue(); + assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "taskScheduler")).isNull(); + + /* + * Test there's no harm if the reply is received after the cancel. This + * should never happen because the container is stopped before canceling + * and the future is removed from the pending map. + */ + future.complete("foo"); + assertThat(callback.result).isNull(); + } + + private void checkConverterResult(CompletableFuture future, String expected) throws InterruptedException { + final CountDownLatch cdl = new CountDownLatch(1); + final AtomicReference resultRef = new AtomicReference<>(); + future.whenComplete((result, ex) -> { + resultRef.set(result); + cdl.countDown(); + }); + assertThat(cdl.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(resultRef.get()).isEqualTo(expected); + } + + private Message checkMessageResult(CompletableFuture future, String expected) throws InterruptedException { + final CountDownLatch cdl = new CountDownLatch(1); + final AtomicReference resultRef = new AtomicReference<>(); + future.whenComplete((result, ex) -> { + resultRef.set(result); + cdl.countDown(); + }); + assertThat(cdl.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(new String(resultRef.get().getBody())).isEqualTo(expected); + return resultRef.get(); + } + + public static class TheCallback implements BiConsumer { + + private final CountDownLatch latch = new CountDownLatch(1); + + private volatile String result; + + private volatile Throwable ex; + + + @Override + public void accept(String result, Throwable ex) { + this.result = result; + this.ex = ex; + latch.countDown(); + } + + } + + @Configuration + public static class Config { + + @Bean + public AtomicReference latch() { + return new AtomicReference<>(); + } + + @Bean + public ConnectionFactory connectionFactory() { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost"); + connectionFactory.setPublisherConfirmType(ConfirmType.CORRELATED); + connectionFactory.setPublisherReturns(true); + return connectionFactory; + } + + @Bean + public Queue requests() { + return new AnonymousQueue(); + } + + @Bean + public Queue replies() { + return new AnonymousQueue(); + } + + @Bean + public RabbitAdmin admin(ConnectionFactory connectionFactory) { + return new RabbitAdmin(connectionFactory); + } + + @Bean + public GZipPostProcessor gZipPostProcessor() { + GZipPostProcessor gZipPostProcessor = new GZipPostProcessor(); + gZipPostProcessor.setCopyProperties(true); + return gZipPostProcessor; + } + + @Bean + public RabbitTemplate template(ConnectionFactory connectionFactory) { + RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); + rabbitTemplate.setRoutingKey(requests().getName()); + rabbitTemplate.addBeforePublishPostProcessors(gZipPostProcessor()); + rabbitTemplate.addAfterReceivePostProcessors(new GUnzipPostProcessor()); + return rabbitTemplate; + } + + @Bean + public RabbitTemplate templateForDirect(ConnectionFactory connectionFactory) { + RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); + rabbitTemplate.setRoutingKey(requests().getName()); + rabbitTemplate.addBeforePublishPostProcessors(gZipPostProcessor()); + rabbitTemplate.addAfterReceivePostProcessors(new GUnzipPostProcessor()); + return rabbitTemplate; + } + + @Bean + @Primary + public SimpleMessageListenerContainer replyContainer(ConnectionFactory connectionFactory) { + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); + container.setAfterReceivePostProcessors(new GUnzipPostProcessor()); + container.setQueueNames(replies().getName()); + return container; + } + + @Bean + public AsyncRabbitTemplate2 asyncTemplate(RabbitTemplate template, SimpleMessageListenerContainer container) { + return new AsyncRabbitTemplate2(template, container); + } + + @Bean + public AsyncRabbitTemplate2 asyncDirectTemplate(RabbitTemplate templateForDirect) { + return new AsyncRabbitTemplate2(templateForDirect); + } + + @Bean + public SimpleMessageListenerContainer remoteContainer(ConnectionFactory connectionFactory) { + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); + container.setQueueNames(requests().getName()); + container.setAfterReceivePostProcessors(new GUnzipPostProcessor()); + MessageListenerAdapter messageListener = + new MessageListenerAdapter((ReplyingMessageListener) + message -> { + CountDownLatch countDownLatch = latch().get(); + if (countDownLatch != null) { + try { + countDownLatch.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + if ("sleep".equals(message)) { + try { + Thread.sleep(500); // time for confirm to be delivered, or timeout to occur + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + else if ("noReply".equals(message)) { + return null; + } + return message.toUpperCase(); + }); + + messageListener.setBeforeSendReplyPostProcessors(gZipPostProcessor()); + container.setMessageListener(messageListener); + return container; + } + + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java index 75eddf50..ff684b1f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-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. @@ -38,8 +38,6 @@ import org.springframework.amqp.core.AnonymousQueue; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.core.Queue; -import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture; -import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -71,6 +69,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback; @SpringJUnitConfig @DirtiesContext @RabbitAvailable +@SuppressWarnings("deprecation") public class AsyncRabbitTemplateTests { @Autowired @@ -253,7 +252,7 @@ public class AsyncRabbitTemplateTests { @DirtiesContext public void testConvertWithConfirm() throws Exception { this.asyncTemplate.setEnableConfirms(true); - RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("sleep"); + org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("sleep"); ListenableFuture confirm = future.getConfirm(); assertThat(confirm).isNotNull(); assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); @@ -264,7 +263,7 @@ public class AsyncRabbitTemplateTests { @DirtiesContext public void testMessageWithConfirm() throws Exception { this.asyncTemplate.setEnableConfirms(true); - RabbitMessageFuture future = this.asyncTemplate + org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture future = this.asyncTemplate .sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties())); ListenableFuture confirm = future.getConfirm(); assertThat(confirm).isNotNull(); @@ -276,7 +275,7 @@ public class AsyncRabbitTemplateTests { @DirtiesContext public void testConvertWithConfirmDirect() throws Exception { this.asyncDirectTemplate.setEnableConfirms(true); - RabbitConverterFuture future = this.asyncDirectTemplate.convertSendAndReceive("sleep"); + org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture future = this.asyncDirectTemplate.convertSendAndReceive("sleep"); ListenableFuture confirm = future.getConfirm(); assertThat(confirm).isNotNull(); assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); @@ -287,7 +286,7 @@ public class AsyncRabbitTemplateTests { @DirtiesContext public void testMessageWithConfirmDirect() throws Exception { this.asyncDirectTemplate.setEnableConfirms(true); - RabbitMessageFuture future = this.asyncDirectTemplate + org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture future = this.asyncDirectTemplate .sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties())); ListenableFuture confirm = future.getConfirm(); assertThat(confirm).isNotNull(); @@ -321,7 +320,7 @@ public class AsyncRabbitTemplateTests { @DirtiesContext public void testReplyAfterReceiveTimeout() throws Exception { this.asyncTemplate.setReceiveTimeout(100); - RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("sleep"); + org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("sleep"); TheCallback callback = new TheCallback(); future.addCallback(callback); assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1); @@ -351,7 +350,7 @@ public class AsyncRabbitTemplateTests { @DirtiesContext public void testStopCancelled() throws Exception { this.asyncTemplate.setReceiveTimeout(5000); - RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("noReply"); + org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("noReply"); TheCallback callback = new TheCallback(); future.addCallback(callback); assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AsyncListenerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AsyncListenerTests.java index d5ce8640..4fe63dee 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AsyncListenerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AsyncListenerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-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. @@ -24,6 +24,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -35,8 +36,8 @@ import org.springframework.amqp.ImmediateRequeueAmqpException; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.AnonymousQueue; import org.springframework.amqp.core.Queue; -import org.springframework.amqp.rabbit.AsyncRabbitTemplate; -import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture; +import org.springframework.amqp.rabbit.AsyncRabbitTemplate2; +import org.springframework.amqp.rabbit.AsyncRabbitTemplate2.RabbitConverterFuture2; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -53,8 +54,6 @@ import org.springframework.context.annotation.Configuration; import org.springframework.stereotype.Component; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SettableListenableFuture; import reactor.core.publisher.Mono; @@ -75,7 +74,7 @@ public class AsyncListenerTests { private RabbitTemplate rabbitTemplate; @Autowired - private AsyncRabbitTemplate asyncTemplate; + private AsyncRabbitTemplate2 asyncTemplate; @Autowired private Queue queue1; @@ -107,7 +106,7 @@ public class AsyncListenerTests { @Test public void testAsyncListener() throws Exception { assertThat(this.rabbitTemplate.convertSendAndReceive(this.queue1.getName(), "foo")).isEqualTo("FOO"); - RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive(this.queue1.getName(), "foo"); + RabbitConverterFuture2 future = this.asyncTemplate.convertSendAndReceive(this.queue1.getName(), "foo"); assertThat(future.get(10, TimeUnit.SECONDS)).isEqualTo("FOO"); assertThat(this.config.typeId).isEqualTo("java.lang.String"); assertThat(this.rabbitTemplate.convertSendAndReceive(this.queue2.getName(), "foo")).isEqualTo("FOO"); @@ -194,8 +193,8 @@ public class AsyncListenerTests { } @Bean - public AsyncRabbitTemplate asyncTemplate() { - return new AsyncRabbitTemplate(rabbitTemplate()); + public AsyncRabbitTemplate2 asyncTemplate() { + return new AsyncRabbitTemplate2(rabbitTemplate()); } @Bean @@ -284,13 +283,13 @@ public class AsyncListenerTests { private final AtomicBoolean first7 = new AtomicBoolean(true); @RabbitListener(id = "foo", queues = "#{queue1.name}") - public ListenableFuture listen1(String foo) { - SettableListenableFuture future = new SettableListenableFuture<>(); + public CompletableFuture listen1(String foo) { + CompletableFuture future = new CompletableFuture<>(); if (fooFirst.getAndSet(false)) { - future.setException(new RuntimeException("Future.exception")); + future.completeExceptionally(new RuntimeException("Future.exception")); } else { - future.set(foo.toUpperCase()); + future.complete(foo.toUpperCase()); } return future; } @@ -311,17 +310,17 @@ public class AsyncListenerTests { } @RabbitListener(id = "qux", queues = "#{queue4.name}") - public ListenableFuture listen4(@SuppressWarnings("unused") String foo) { - SettableListenableFuture future = new SettableListenableFuture<>(); - future.set(null); + public CompletableFuture listen4(@SuppressWarnings("unused") String foo) { + CompletableFuture future = new CompletableFuture<>(); + future.complete(null); this.latch4.countDown(); return future; } @RabbitListener(id = "fiz", queues = "#{queue5.name}") - public ListenableFuture listen5(@SuppressWarnings("unused") String foo) { - SettableListenableFuture future = new SettableListenableFuture<>(); - future.setException(new AmqpRejectAndDontRequeueException("asyncToDLQ")); + public CompletableFuture listen5(@SuppressWarnings("unused") String foo) { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new AmqpRejectAndDontRequeueException("asyncToDLQ")); return future; } @@ -331,9 +330,9 @@ public class AsyncListenerTests { } @RabbitListener(id = "fix", queues = "#{queue6.name}", containerFactory = "dontRequeueFactory") - public ListenableFuture listen6(@SuppressWarnings("unused") String foo) { - SettableListenableFuture future = new SettableListenableFuture<>(); - future.setException(new IllegalStateException("asyncDefaultToDLQ")); + public CompletableFuture listen6(@SuppressWarnings("unused") String foo) { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new IllegalStateException("asyncDefaultToDLQ")); return future; } @@ -344,13 +343,13 @@ public class AsyncListenerTests { @RabbitListener(id = "overrideFactoryRequeue", queues = "#{queue7.name}", containerFactory = "dontRequeueFactory") - public ListenableFuture listen7(@SuppressWarnings("unused") String foo) { - SettableListenableFuture future = new SettableListenableFuture<>(); + public CompletableFuture listen7(@SuppressWarnings("unused") String foo) { + CompletableFuture future = new CompletableFuture<>(); if (this.first7.compareAndSet(true, false)) { - future.setException(new ImmediateRequeueAmqpException("asyncOverrideDefaultToDLQ")); + future.completeExceptionally(new ImmediateRequeueAmqpException("asyncOverrideDefaultToDLQ")); } else { - future.set("listen7"); + future.complete("listen7"); } return future; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java index 1c2a504f..0e9c2fd9 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-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. @@ -23,8 +23,8 @@ import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; -import org.springframework.amqp.rabbit.AsyncRabbitTemplate; -import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture; +import org.springframework.amqp.rabbit.AsyncRabbitTemplate2; +import org.springframework.amqp.rabbit.AsyncRabbitTemplate2.RabbitConverterFuture2; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -63,7 +63,7 @@ public class ComplexTypeJsonIntegrationTests { private RabbitTemplate rabbitTemplate; @Autowired - private AsyncRabbitTemplate asyncTemplate; + private AsyncRabbitTemplate2 asyncTemplate; private static Foo> makeAFoo() { Foo> foo = new Foo<>(); @@ -137,7 +137,7 @@ public class ComplexTypeJsonIntegrationTests { new ParameterizedTypeReference>>() { })); } - private void verifyFooBarBazQux(RabbitConverterFuture>> future) throws Exception { + private void verifyFooBarBazQux(RabbitConverterFuture2>> future) throws Exception { verifyFooBarBazQux(future.get(10, TimeUnit.SECONDS)); } @@ -169,8 +169,8 @@ public class ComplexTypeJsonIntegrationTests { } @Bean - public AsyncRabbitTemplate asyncTemplate() { - return new AsyncRabbitTemplate(template()); + public AsyncRabbitTemplate2 asyncTemplate() { + return new AsyncRabbitTemplate2(template()); } @Bean diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java index e48dc3b2..62e8871b 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2021 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. @@ -49,7 +49,7 @@ public class MessagingTemplateConfirmsTests { CorrelationData data = new CorrelationData(); rmt.send("messaging.confirms", new GenericMessage<>("foo", Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, data))); - assertThat(data.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); + assertThat(data.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); ccf.destroy(); } @@ -65,7 +65,7 @@ public class MessagingTemplateConfirmsTests { CorrelationData data = new CorrelationData("foo"); rmt.send("messaging.confirms.unroutable", new GenericMessage<>("foo", Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, data))); - assertThat(data.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); + assertThat(data.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); assertThat(data.getReturned()).isNotNull(); ccf.destroy(); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index 2c9808fe..c8d89a47 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -839,14 +839,14 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { admin.declareQueue(queue); CorrelationData cd1 = new CorrelationData(); this.templateWithConfirmsEnabled.convertAndSend("", queue.getName(), "foo", cd1); - assertThat(cd1.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); + assertThat(cd1.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); CorrelationData cd2 = new CorrelationData(); this.templateWithConfirmsEnabled.convertAndSend("", queue.getName(), "bar", cd2); - assertThat(cd2.getFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse(); + assertThat(cd2.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse(); CorrelationData cd3 = new CorrelationData(); this.templateWithConfirmsEnabled.convertAndSend("NO_EXCHANGE_HERE", queue.getName(), "foo", cd3); - assertThat(cd3.getFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse(); - assertThat(cd3.getFuture().get().getReason()).contains("NOT_FOUND"); + assertThat(cd3.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse(); + assertThat(cd3.getCompletableFuture().get().getReason()).contains("NOT_FOUND"); CorrelationData cd4 = new CorrelationData("42"); AtomicBoolean resent = new AtomicBoolean(); AtomicReference callbackThreadName = new AtomicReference<>(); @@ -858,7 +858,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { callbackLatch.countDown(); }); this.templateWithConfirmsAndReturnsEnabled.convertAndSend("", "NO_QUEUE_HERE", "foo", cd4); - assertThat(cd4.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); + assertThat(cd4.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); assertThat(callbackLatch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(cd4.getReturned()).isNotNull(); assertThat(resent.get()).isTrue(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java index 8d46fb24..2774bdd3 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-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. @@ -118,13 +118,13 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests2 { this.templateWithConfirmsEnabled.setMandatory(true); CorrelationData corr = new CorrelationData(); this.templateWithConfirmsEnabled.convertAndSend("", ROUTE2, "foo", corr); - assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); + assertThat(corr.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); if (listener) { assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); } corr = new CorrelationData(); this.templateWithConfirmsEnabled.convertAndSend("", "bad route", "foo", corr); - assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); + assertThat(corr.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue(); assertThat(corr.getReturned()).isNotNull(); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateRoutingConnectionFactoryIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateRoutingConnectionFactoryIntegrationTests.java index b2ebbbb7..7084ef31 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateRoutingConnectionFactoryIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateRoutingConnectionFactoryIntegrationTests.java @@ -109,7 +109,7 @@ class RabbitTemplateRoutingConnectionFactoryIntegrationTests { rabbitTemplate.send(ROUTE, message, correlationData); assertThat(rabbitTemplate.getUnconfirmedCount()).isEqualTo(1); - final CorrelationData.Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS); + final CorrelationData.Confirm confirm = correlationData.getCompletableFuture().get(10, TimeUnit.SECONDS); assertThat(confirm.isAck()).isTrue(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/AsyncReplyToTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/AsyncReplyToTests.java index 3bfab605..84da4969 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/AsyncReplyToTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/AsyncReplyToTests.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. @@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.listener; import static org.assertj.core.api.Assertions.assertThat; import java.io.IOException; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -97,7 +98,7 @@ public class AsyncReplyToTests { .build()); assertThat(config.dmlcLatch.await(10, TimeUnit.SECONDS)).isTrue(); registry.getListenerContainer("dmlc").stop(); - assertThat(admin.getQueueInfo("async2").getMessageCount()).isEqualTo(1); + assertThat(admin.getQueueInfo("async2").getMessageCount()).isEqualTo(0); } @Configuration @@ -114,8 +115,10 @@ public class AsyncReplyToTests { } @RabbitListener(id = "dmlc", queues = "async2", containerFactory = "dmlcf") - ListenableFuture listen2(String in, Channel channel) { - return new SettableListenableFuture<>(); + CompletableFuture listen2(String in, Channel channel) { + CompletableFuture future = new CompletableFuture<>(); + future.complete("test"); + return future; } @Bean diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java index 15b65019..0db0b7aa 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-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. @@ -27,6 +27,7 @@ import static org.mockito.Mockito.verify; import java.io.IOException; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -43,8 +44,6 @@ import org.springframework.aop.framework.ProxyFactory; import org.springframework.retry.RetryPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SettableListenableFuture; import com.rabbitmq.client.Channel; import reactor.core.publisher.Mono; @@ -220,13 +219,13 @@ public class MessageListenerAdapterTests { } @Test - public void testListenableFutureReturn() throws Exception { + public void testCompletableFutureReturn() throws Exception { class Delegate { @SuppressWarnings("unused") - public ListenableFuture myPojoMessageMethod(String input) { - SettableListenableFuture future = new SettableListenableFuture<>(); - future.set("processed" + input); + public CompletableFuture myPojoMessageMethod(String input) { + CompletableFuture future = new CompletableFuture<>(); + future.complete("processed" + input); return future; } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 5e84157f..6846bd4a 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -3570,6 +3570,7 @@ IMPORTANT: Containers created this way are normal `@Bean` instances and are not ===== Asynchronous `@RabbitListener` Return Types Starting with version 2.1, `@RabbitListener` (and `@RabbitHandler`) methods can be specified with asynchronous return types `ListenableFuture` and `Mono`, letting the reply be sent asynchronously. +`CompletableFuture` was added in 2.4.7 and `ListenableFuture` will be removed in 3.0. IMPORTANT: The listener container factory must be configured with `AcknowledgeMode.MANUAL` so that the consumer thread will not ack the message; instead, the asynchronous completion will ack or nack the message when the async operation completes. When the async result is completed with an error, whether the message is requeued or not depends on the exception type thrown, the container configuration, and the container error handler. @@ -4647,6 +4648,9 @@ Version 2.0 introduced variants of these methods (`convertSendAndReceiveAsType`) You must configure the underlying `RabbitTemplate` with a `SmartMessageConverter`. See <> for more information. +Starting with version 2.4.7, the `AsyncRabbitTemplate` is deprecated in favor of `AsyncRabbitTemplate2` which returns `CompletableFuture` s instead of `ListenableFuture` s. +In 3.0, `AsyncRabbitTemplate2` will be renamed to `AsyncRabbitTemplate`. + [[remoting]] ===== Spring Remoting with AMQP diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 453b8b35..5b4068dc 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -11,6 +11,9 @@ See <> for changes in previous versions. `MessageProperties` is now available for argument matching. See <> for more information. +Async reply types now include `CompleteableFuture` +See <> for more information. + ==== `RabbitAdmin` Changes A new property `recoverManualDeclarations` allows recovery of manually declared queues/exchanges/bindings. @@ -25,3 +28,8 @@ See <> for more information. The `Jackson2JsonMessageConverter` can now determine the charset from the `contentEncoding` header. See <> for more information. + +==== AsyncRabbitTemplate + +The `AsyncRabbitTemplate` is deprecated in favor of `AsyncRabbitTemplate2` which returns `CompletableFuture` s instead of `ListenableFuture` s. +See <> for more information.