diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java index cd84b043..123c3b6f 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2020-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,12 +16,13 @@ package org.springframework.amqp.core; +import java.util.concurrent.CompletableFuture; + import org.springframework.core.ParameterizedTypeReference; -import org.springframework.util.concurrent.ListenableFuture; /** * Classes implementing this interface can perform asynchronous send and - * receive operations. + * receive operations using {@link CompletableFuture}s. * * @author Gary Russell * @since 2.0 @@ -33,18 +34,18 @@ public interface AsyncAmqpTemplate { * Send a message to the default exchange with the default routing key. If the message * contains a correlationId property, it must be unique. * @param message the message. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture sendAndReceive(Message message); + 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 ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture sendAndReceive(String routingKey, Message message); + CompletableFuture sendAndReceive(String routingKey, Message message); /** * Send a message to the supplied exchange and routing key. If the message @@ -52,18 +53,18 @@ public interface AsyncAmqpTemplate { * @param exchange the exchange. * @param routingKey the routing key. * @param message the message. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture sendAndReceive(String exchange, String routingKey, Message message); + 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 ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceive(Object object); + CompletableFuture convertSendAndReceive(Object object); /** * Convert the object to a message and send it to the default exchange with the @@ -71,9 +72,9 @@ public interface AsyncAmqpTemplate { * @param routingKey the routing key. * @param object the object to convert. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceive(String routingKey, Object object); + CompletableFuture convertSendAndReceive(String routingKey, Object object); /** * Convert the object to a message and send it to the provided exchange and @@ -82,9 +83,9 @@ public interface AsyncAmqpTemplate { * @param routingKey the routing key. * @param object the object to convert. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceive(String exchange, String routingKey, Object object); + CompletableFuture convertSendAndReceive(String exchange, String routingKey, Object object); /** * Convert the object to a message and send it to the default exchange with the @@ -93,9 +94,9 @@ public interface AsyncAmqpTemplate { * @param object the object to convert. * @param messagePostProcessor the post processor. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor); + CompletableFuture convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor); /** * Convert the object to a message and send it to the default exchange with the @@ -105,9 +106,9 @@ public interface AsyncAmqpTemplate { * @param object the object to convert. * @param messagePostProcessor the post processor. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceive(String routingKey, Object object, + CompletableFuture convertSendAndReceive(String routingKey, Object object, MessagePostProcessor messagePostProcessor); /** @@ -119,9 +120,9 @@ public interface AsyncAmqpTemplate { * @param object the object to convert. * @param messagePostProcessor the post processor. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceive(String exchange, String routingKey, Object object, + CompletableFuture convertSendAndReceive(String exchange, String routingKey, Object object, MessagePostProcessor messagePostProcessor); /** @@ -130,9 +131,9 @@ public interface AsyncAmqpTemplate { * @param object the object to convert. * @param responseType the response type. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceiveAsType(Object object, ParameterizedTypeReference responseType); + CompletableFuture convertSendAndReceiveAsType(Object object, ParameterizedTypeReference responseType); /** * Convert the object to a message and send it to the default exchange with the @@ -141,9 +142,9 @@ public interface AsyncAmqpTemplate { * @param object the object to convert. * @param responseType the response type. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceiveAsType(String routingKey, Object object, + CompletableFuture convertSendAndReceiveAsType(String routingKey, Object object, ParameterizedTypeReference responseType); /** @@ -154,9 +155,9 @@ public interface AsyncAmqpTemplate { * @param object the object to convert. * @param responseType the response type. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, + CompletableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, ParameterizedTypeReference responseType); /** @@ -167,9 +168,9 @@ public interface AsyncAmqpTemplate { * @param messagePostProcessor the post processor. * @param responseType the response type. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor, + CompletableFuture convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType); /** @@ -181,9 +182,9 @@ public interface AsyncAmqpTemplate { * @param messagePostProcessor the post processor. * @param responseType the response type. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceiveAsType(String routingKey, Object object, + CompletableFuture convertSendAndReceiveAsType(String routingKey, Object object, MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType); /** @@ -196,9 +197,9 @@ public interface AsyncAmqpTemplate { * @param messagePostProcessor the post processor. * @param responseType the response type. * @param the expected result type. - * @return the {@link ListenableFuture}. + * @return the {@link CompletableFuture}. */ - ListenableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, + 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 3d31b4cd..56dcfe9b 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-2022 the original author or authors. + * 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. @@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit; import java.time.Instant; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ScheduledFuture; @@ -29,7 +30,6 @@ 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.AsyncAmqpTemplate; import org.springframework.amqp.core.Correlation; import org.springframework.amqp.core.Message; @@ -60,17 +60,15 @@ import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; import org.springframework.util.StringUtils; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SettableListenableFuture; import com.rabbitmq.client.Channel; /** - * Provides asynchronous send and receive operations returning a {@link ListenableFuture} + * 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 ListenableFuture}. If the reply is received before the publisher confirm, + * {@link CompletableFuture}. If the reply is received before the publisher confirm, * the confirm is discarded since the reply implicitly indicates the message was * published. *

@@ -296,7 +294,7 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa /** * Set to true to enable publisher confirms. When enabled, the {@link RabbitFuture} * returned by the send and receive operation will have a - * {@code ListenableFuture} in its {@code confirm} property. + * {@code CompletableFuture} in its {@code confirm} property. * @param enableConfirms true to enable publisher confirms. */ public void setEnableConfirms(boolean enableConfirms) { @@ -375,11 +373,12 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa @Override public RabbitMessageFuture sendAndReceive(String exchange, String routingKey, Message message) { String correlationId = getOrSetCorrelationIdAndSetReplyTo(message, null); - RabbitMessageFuture future = new RabbitMessageFuture(correlationId, message); + RabbitMessageFuture future = new RabbitMessageFuture(correlationId, message, this::canceler, + this::timeoutTask); CorrelationData correlationData = null; if (this.enableConfirms) { correlationData = new CorrelationData(correlationId); - future.setConfirm(new SettableListenableFuture<>()); + future.setConfirm(new CompletableFuture<>()); } this.pending.put(correlationId, future); if (this.container != null) { @@ -578,7 +577,7 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa } RabbitFuture future = this.pending.remove(correlationId); if (future != null) { - if (future instanceof AsyncRabbitTemplate.RabbitConverterFuture) { + if (future instanceof RabbitConverterFuture) { MessageConverter messageConverter = this.template.getMessageConverter(); RabbitConverterFuture rabbitFuture = (RabbitConverterFuture) future; Object converted = rabbitFuture.getReturnType() != null @@ -586,10 +585,10 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa ? ((SmartMessageConverter) messageConverter).fromMessage(message, rabbitFuture.getReturnType()) : messageConverter.fromMessage(message); - rabbitFuture.set(converted); + rabbitFuture.complete(converted); } else { - ((RabbitMessageFuture) future).set(message); + ((RabbitMessageFuture) future).complete(message); } } else { @@ -608,7 +607,7 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa if (StringUtils.hasText(correlationId)) { RabbitFuture future = this.pending.remove(correlationId); if (future != null) { - future.setException(new AmqpMessageReturnedException("Message returned", returned)); + future.completeExceptionally(new AmqpMessageReturnedException("Message returned", returned)); } else { if (this.logger.isWarnEnabled()) { @@ -630,7 +629,7 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa RabbitFuture future = this.pending.get(correlationId); if (future != null) { future.setNackCause(cause); - ((SettableListenableFuture) future.getConfirm()).set(ack); + future.getConfirm().complete(ack); } else { if (this.logger.isDebugEnabled()) { @@ -661,147 +660,35 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa return correlationId; } + private void canceler(String correlationId, @Nullable ChannelHolder channelHolder) { + this.pending.remove(correlationId); + if (channelHolder != null && this.directReplyToContainer != null) { + this.directReplyToContainer + .releaseConsumerFor(channelHolder, false, null); // NOSONAR + } + } + + @Nullable + private ScheduledFuture timeoutTask(RabbitFuture future) { + if (this.receiveTimeout > 0) { + synchronized (this) { + if (!this.running) { + this.pending.remove(future.getCorrelationId()); + throw new IllegalStateException("'AsyncRabbitTemplate' must be started."); + } + return this.taskScheduler.schedule( + new TimeoutTask(future, this.pending, this.directReplyToContainer), + Instant.now().plusMillis(this.receiveTimeout)); + } + } + return null; + } + @Override public String toString() { return this.beanName == null ? super.toString() : (this.getClass().getSimpleName() + ": " + this.beanName); } - /** - * Base class for {@link ListenableFuture}s returned by {@link AsyncRabbitTemplate}. - * @param the type. - * @since 1.6 - */ - public abstract class RabbitFuture extends SettableListenableFuture { - - private final String correlationId; - - private final Message requestMessage; - - private ScheduledFuture timeoutTask; - - private volatile ListenableFuture confirm; - - private String nackCause; - - private ChannelHolder channelHolder; - - public RabbitFuture(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); - } - AsyncRabbitTemplate.this.pending.remove(this.correlationId); - if (this.channelHolder != null && AsyncRabbitTemplate.this.directReplyToContainer != null) { - AsyncRabbitTemplate.this.directReplyToContainer - .releaseConsumerFor(this.channelHolder, false, null); // NOSONAR - } - return super.cancel(mayInterruptIfRunning); - } - - /** - * When confirms are enabled contains a {@link ListenableFuture} - * for the confirmation. - * @return the future. - */ - public ListenableFuture getConfirm() { - return this.confirm; - } - - void setConfirm(ListenableFuture 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 (AsyncRabbitTemplate.this.receiveTimeout > 0) { - synchronized (AsyncRabbitTemplate.this) { - if (!AsyncRabbitTemplate.this.running) { - AsyncRabbitTemplate.this.pending.remove(this.correlationId); - throw new IllegalStateException("'AsyncRabbitTemplate' must be started."); - } - this.timeoutTask = AsyncRabbitTemplate.this.taskScheduler.schedule(new TimeoutTask(), - Instant.now().plusMillis(AsyncRabbitTemplate.this.receiveTimeout)); - } - } - else { - this.timeoutTask = null; - } - } - - private class TimeoutTask implements Runnable { - - @Override - public void run() { - AsyncRabbitTemplate.this.pending.remove(RabbitFuture.this.correlationId); - if (RabbitFuture.this.channelHolder != null - && AsyncRabbitTemplate.this.directReplyToContainer != null) { - AsyncRabbitTemplate.this.directReplyToContainer - .releaseConsumerFor(RabbitFuture.this.channelHolder, false, null); // NOSONAR - } - setException(new AmqpReplyTimeoutException("Reply timed out", RabbitFuture.this.requestMessage)); - } - - } - - } - - /** - * A {@link RabbitFuture} with a return type of {@link Message}. - * @since 1.6 - */ - public class RabbitMessageFuture extends RabbitFuture { - - public RabbitMessageFuture(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 RabbitConverterFuture extends RabbitFuture { - - private volatile ParameterizedTypeReference returnType; - - public RabbitConverterFuture(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() { @@ -821,10 +708,11 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa messageToSend = correlationData.userPostProcessor.postProcessMessage(message); } String correlationId = getOrSetCorrelationIdAndSetReplyTo(messageToSend, correlationData); - correlationData.future = new RabbitConverterFuture(correlationId, message); + correlationData.future = new RabbitConverterFuture(correlationId, message, + AsyncRabbitTemplate.this::canceler, AsyncRabbitTemplate.this::timeoutTask); if (correlationData.enableConfirms) { correlationData.setId(correlationId); - correlationData.future.setConfirm(new SettableListenableFuture<>()); + correlationData.future.setConfirm(new CompletableFuture<>()); } correlationData.future.setReturnType(correlationData.returnType); AsyncRabbitTemplate.this.pending.put(correlationId, correlationData.future); 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..a02af02c --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2.java @@ -0,0 +1,64 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit; + +import java.util.concurrent.CompletableFuture; + +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; + +/** + * This class was added in 2.4.7 to aid migration from methods returning + * {@code ListenableFuture}s to {@link CompletableFuture}s. + * + * @author Gary Russell + * @since 2.4.7 + * @deprecated in favor of {@link AsyncRabbitTemplate}. + * + */ +@Deprecated +public class AsyncRabbitTemplate2 extends AsyncRabbitTemplate { + + public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey, + String replyQueue, String replyAddress) { + super(connectionFactory, exchange, routingKey, replyQueue, replyAddress); + } + + public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey, + String replyQueue) { + super(connectionFactory, exchange, routingKey, replyQueue); + } + + public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey) { + super(connectionFactory, exchange, routingKey); + } + + public AsyncRabbitTemplate2(RabbitTemplate template, AbstractMessageListenerContainer container, + String replyAddress) { + super(template, container, replyAddress); + } + + public AsyncRabbitTemplate2(RabbitTemplate template, AbstractMessageListenerContainer container) { + super(template, container); + } + + public AsyncRabbitTemplate2(RabbitTemplate template) { + super(template); + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitConverterFuture.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitConverterFuture.java new file mode 100644 index 00000000..fe799c57 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitConverterFuture.java @@ -0,0 +1,54 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit; + +import java.util.concurrent.ScheduledFuture; +import java.util.function.BiConsumer; +import java.util.function.Function; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder; +import org.springframework.core.ParameterizedTypeReference; + +/** + * A {@link RabbitFuture} with a return type of the template's + * generic parameter. + * @param the type. + * + * @author Gary Russell + * @since 2.4.7 + */ +public class RabbitConverterFuture extends RabbitFuture { + + private volatile ParameterizedTypeReference returnType; + + RabbitConverterFuture(String correlationId, Message requestMessage, + BiConsumer canceler, + Function, ScheduledFuture> timeoutTaskFunction) { + + super(correlationId, requestMessage, canceler, timeoutTaskFunction); + } + + public ParameterizedTypeReference getReturnType() { + return this.returnType; + } + + public void setReturnType(ParameterizedTypeReference returnType) { + this.returnType = returnType; + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitFuture.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitFuture.java new file mode 100644 index 00000000..17bab188 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitFuture.java @@ -0,0 +1,116 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledFuture; +import java.util.function.BiConsumer; +import java.util.function.Function; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder; + +/** + * Base class for {@link CompletableFuture}s returned by {@link AsyncRabbitTemplate}. + * @param the type. + * + * @author Gary Russell + * @since 2.4.7 + */ +public abstract class RabbitFuture extends CompletableFuture { + + private final String correlationId; + + private final Message requestMessage; + + private final BiConsumer canceler; + + private final Function, ScheduledFuture> timeoutTaskFunction; + + private ScheduledFuture timeoutTask; + + private volatile CompletableFuture confirm; + + private String nackCause; + + private ChannelHolder channelHolder; + + protected RabbitFuture(String correlationId, Message requestMessage, BiConsumer canceler, + Function, ScheduledFuture> timeoutTaskFunction) { + + this.correlationId = correlationId; + this.requestMessage = requestMessage; + this.canceler = canceler; + this.timeoutTaskFunction = timeoutTaskFunction; + } + + void setChannelHolder(ChannelHolder channel) { + this.channelHolder = channel; + } + + String getCorrelationId() { + return this.correlationId; + } + + ChannelHolder getChannelHolder() { + return this.channelHolder; + } + + Message getRequestMessage() { + return this.requestMessage; + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + if (this.timeoutTask != null) { + this.timeoutTask.cancel(true); + } + this.canceler.accept(this.correlationId, this.channelHolder); + return super.cancel(mayInterruptIfRunning); + } + + /** + * When confirms are enabled contains a {@link CompletableFuture} + * for the confirmation. + * @return the future. + */ + public CompletableFuture 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() { + this.timeoutTask = this.timeoutTaskFunction.apply(this); + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitMessageFuture.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitMessageFuture.java new file mode 100644 index 00000000..467053a3 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitMessageFuture.java @@ -0,0 +1,40 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit; + +import java.util.concurrent.ScheduledFuture; +import java.util.function.BiConsumer; +import java.util.function.Function; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder; + +/** + * A {@link RabbitFuture} with a return type of {@link Message}. + * + * @author Gary Russell + * @since 2.4.7 + */ +public class RabbitMessageFuture extends RabbitFuture { + + RabbitMessageFuture(String correlationId, Message requestMessage, BiConsumer canceler, + Function, ScheduledFuture> timeoutTaskFunction) { + + super(correlationId, requestMessage, canceler, timeoutTaskFunction); + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/TimeoutTask.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/TimeoutTask.java new file mode 100644 index 00000000..8ca3af16 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/TimeoutTask.java @@ -0,0 +1,59 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit; + +import java.util.concurrent.ConcurrentMap; + +import org.springframework.amqp.core.AmqpReplyTimeoutException; +import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder; +import org.springframework.lang.Nullable; + +/** + * A {@link Runnable} used to time out a {@link RabbitFuture}. + * + * @author Gary Russell + * @since 2.4.7 + */ +public class TimeoutTask implements Runnable { + + private final RabbitFuture future; + + private final ConcurrentMap> pending; + + private final DirectReplyToMessageListenerContainer container; + + TimeoutTask(RabbitFuture future, ConcurrentMap> pending, + @Nullable DirectReplyToMessageListenerContainer container) { + + this.future = future; + this.pending = pending; + this.container = container; + } + + @Override + public void run() { + this.pending.remove(this.future.getCorrelationId()); + ChannelHolder holder = this.future.getChannelHolder(); + if (holder != null && this.container != null) { + this.container.releaseConsumerFor(holder, false, null); // NOSONAR + } + this.future.completeExceptionally( + new AmqpReplyTimeoutException("Reply timed out", this.future.getRequestMessage())); + } + +} 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..46ffaedc 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,13 +17,13 @@ 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; import org.springframework.amqp.core.ReturnedMessage; import org.springframework.lang.Nullable; import org.springframework.util.Assert; -import org.springframework.util.concurrent.SettableListenableFuture; /** * Base class for correlating publisher confirms to sent messages. Use the @@ -41,7 +41,7 @@ import org.springframework.util.concurrent.SettableListenableFuture; */ public class CorrelationData implements Correlation { - private final SettableListenableFuture future = new SettableListenableFuture<>(); + private final CompletableFuture future = new CompletableFuture<>(); private volatile String id; @@ -91,7 +91,18 @@ public class CorrelationData implements Correlation { * @return the future. * @since 2.1 */ - public SettableListenableFuture getFuture() { + public CompletableFuture getFuture() { + return this.future; + } + + /** + * Return a future to check the success/failure of the publish operation. + * @return the future. + * @since 2.4.7 + * @deprecated in favor of {@link #getFuture()}. + */ + @Deprecated + public CompletableFuture getCompletableFuture() { return this.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..d1c695d9 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. @@ -955,7 +955,7 @@ public class PublisherCallbackChannelImpl if (pendingConfirm != null) { CorrelationData correlationData = pendingConfirm.getCorrelationData(); if (correlationData != null) { - correlationData.getFuture().set(new Confirm(ack, pendingConfirm.getCause())); + correlationData.getFuture().complete(new Confirm(ack, pendingConfirm.getCause())); if (StringUtils.hasText(correlationData.getId())) { this.pendingReturns.remove(correlationData.getId()); // NOSONAR never null } @@ -991,7 +991,7 @@ public class PublisherCallbackChannelImpl PendingConfirm value = entry.getValue(); CorrelationData correlationData = value.getCorrelationData(); if (correlationData != null) { - correlationData.getFuture().set(new Confirm(ack, value.getCause())); + correlationData.getFuture().complete(new Confirm(ack, value.getCause())); if (StringUtils.hasText(correlationData.getId())) { this.pendingReturns.remove(correlationData.getId()); // NOSONAR never null } 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..2e0bd818 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; @@ -52,7 +53,6 @@ import org.springframework.retry.RecoveryCallback; import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; -import org.springframework.util.concurrent.ListenableFuture; import com.rabbitmq.client.Channel; @@ -374,17 +374,20 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe */ protected void handleResult(InvocationResult resultArg, Message request, Channel channel, Object source) { if (channel != null) { - if (resultArg.getReturnValue() instanceof ListenableFuture) { + 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"); } - ((ListenableFuture) resultArg.getReturnValue()).addCallback( - r -> { + ((CompletableFuture) resultArg.getReturnValue()).whenComplete((r, t) -> { + if (t == null) { asyncSuccess(resultArg, request, channel, source, r); basicAck(request, channel); - }, - t -> asyncFailure(request, channel, t)); + } + else { + asyncFailure(request, channel, t); + } + }); } else if (monoPresent && MonoHandler.isMono(resultArg.getReturnValue())) { if (!this.isManualAck) { 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..6a99f75a 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; @@ -45,7 +46,6 @@ import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.util.Assert; -import org.springframework.util.concurrent.ListenableFuture; import org.springframework.validation.Validator; @@ -147,7 +147,7 @@ public class DelegatingInvocableHandler { private boolean isAsyncReply(InvocableHandlerMethod method) { return (AbstractAdaptableMessageListener.monoPresent && MonoHandler.isMono(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..e6e0f1a1 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,11 +18,11 @@ 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; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; -import org.springframework.util.concurrent.ListenableFuture; /** * A wrapper for either an {@link InvocableHandlerMethod} or @@ -50,7 +50,7 @@ public class HandlerAdapter { this.delegatingHandler = null; this.asyncReplies = (AbstractAdaptableMessageListener.monoPresent && MonoHandler.isMono(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/AsyncRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java index 75eddf50..039355ec 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. @@ -19,15 +19,18 @@ 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 static org.mockito.Mockito.mock; 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; @@ -38,14 +41,13 @@ 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; 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.AbstractMessageListenerContainer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener; @@ -59,8 +61,6 @@ 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; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.ListenableFutureCallback; /** * @author Gary Russell @@ -90,7 +90,7 @@ public class AsyncRabbitTemplateTests { @Test public void testConvert1Arg() throws Exception { final AtomicBoolean mppCalled = new AtomicBoolean(); - ListenableFuture future = this.asyncTemplate.convertSendAndReceive("foo", m -> { + CompletableFuture future = this.asyncTemplate.convertSendAndReceive("foo", m -> { mppCalled.set(true); return m; }); @@ -101,8 +101,8 @@ public class AsyncRabbitTemplateTests { @Test public void testConvert1ArgDirect() throws Exception { this.latch.set(new CountDownLatch(1)); - ListenableFuture future1 = this.asyncDirectTemplate.convertSendAndReceive("foo"); - ListenableFuture future2 = this.asyncDirectTemplate.convertSendAndReceive("bar"); + CompletableFuture future1 = this.asyncDirectTemplate.convertSendAndReceive("foo"); + CompletableFuture future2 = this.asyncDirectTemplate.convertSendAndReceive("bar"); this.latch.get().countDown(); checkConverterResult(future1, "FOO"); checkConverterResult(future2, "BAR"); @@ -127,19 +127,19 @@ public class AsyncRabbitTemplateTests { @Test public void testConvert2Args() throws Exception { - ListenableFuture future = this.asyncTemplate.convertSendAndReceive(this.requests.getName(), "foo"); + CompletableFuture future = this.asyncTemplate.convertSendAndReceive(this.requests.getName(), "foo"); checkConverterResult(future, "FOO"); } @Test public void testConvert3Args() throws Exception { - ListenableFuture future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo"); + CompletableFuture future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo"); checkConverterResult(future, "FOO"); } @Test public void testConvert4Args() throws Exception { - ListenableFuture future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo", + CompletableFuture future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo", message -> { String body = new String(message.getBody()); return new Message((body + "bar").getBytes(), message.getMessageProperties()); @@ -149,15 +149,15 @@ public class AsyncRabbitTemplateTests { @Test public void testMessage1Arg() throws Exception { - ListenableFuture future = this.asyncTemplate.sendAndReceive(getFooMessage()); + CompletableFuture future = this.asyncTemplate.sendAndReceive(getFooMessage()); checkMessageResult(future, "FOO"); } @Test public void testMessage1ArgDirect() throws Exception { this.latch.set(new CountDownLatch(1)); - ListenableFuture future1 = this.asyncDirectTemplate.sendAndReceive(getFooMessage()); - ListenableFuture future2 = this.asyncDirectTemplate.sendAndReceive(getFooMessage()); + 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); @@ -183,13 +183,13 @@ public class AsyncRabbitTemplateTests { @Test public void testMessage2Args() throws Exception { - ListenableFuture future = this.asyncTemplate.sendAndReceive(this.requests.getName(), getFooMessage()); + CompletableFuture future = this.asyncTemplate.sendAndReceive(this.requests.getName(), getFooMessage()); checkMessageResult(future, "FOO"); } @Test public void testMessage3Args() throws Exception { - ListenableFuture future = this.asyncTemplate.sendAndReceive("", this.requests.getName(), + CompletableFuture future = this.asyncTemplate.sendAndReceive("", this.requests.getName(), getFooMessage()); checkMessageResult(future, "FOO"); } @@ -197,7 +197,7 @@ public class AsyncRabbitTemplateTests { @SuppressWarnings("unchecked") @Test public void testCancel() { - ListenableFuture future = this.asyncTemplate.convertSendAndReceive("foo"); + CompletableFuture future = this.asyncTemplate.convertSendAndReceive("foo"); future.cancel(false); assertThat(TestUtils.getPropertyValue(asyncTemplate, "pending", Map.class)).hasSize(0); } @@ -206,7 +206,7 @@ public class AsyncRabbitTemplateTests { public void testMessageCustomCorrelation() throws Exception { Message message = getFooMessage(); message.getMessageProperties().setCorrelationId("foo"); - ListenableFuture future = this.asyncTemplate.sendAndReceive(message); + CompletableFuture future = this.asyncTemplate.sendAndReceive(message); Message result = checkMessageResult(future, "FOO"); assertThat(result.getMessageProperties().getCorrelationId()).isEqualTo("foo"); } @@ -221,7 +221,7 @@ public class AsyncRabbitTemplateTests { @DirtiesContext public void testReturn() throws Exception { this.asyncTemplate.setMandatory(true); - ListenableFuture future = this.asyncTemplate.convertSendAndReceive(this.requests.getName() + "x", + CompletableFuture future = this.asyncTemplate.convertSendAndReceive(this.requests.getName() + "x", "foo"); try { future.get(10, TimeUnit.SECONDS); @@ -237,7 +237,7 @@ public class AsyncRabbitTemplateTests { @DirtiesContext public void testReturnDirect() throws Exception { this.asyncDirectTemplate.setMandatory(true); - ListenableFuture future = this.asyncDirectTemplate.convertSendAndReceive(this.requests.getName() + "x", + CompletableFuture future = this.asyncDirectTemplate.convertSendAndReceive(this.requests.getName() + "x", "foo"); try { future.get(10, TimeUnit.SECONDS); @@ -254,7 +254,7 @@ public class AsyncRabbitTemplateTests { public void testConvertWithConfirm() throws Exception { this.asyncTemplate.setEnableConfirms(true); RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("sleep"); - ListenableFuture confirm = future.getConfirm(); + CompletableFuture confirm = future.getConfirm(); assertThat(confirm).isNotNull(); assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); checkConverterResult(future, "SLEEP"); @@ -266,7 +266,7 @@ public class AsyncRabbitTemplateTests { this.asyncTemplate.setEnableConfirms(true); RabbitMessageFuture future = this.asyncTemplate .sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties())); - ListenableFuture confirm = future.getConfirm(); + CompletableFuture confirm = future.getConfirm(); assertThat(confirm).isNotNull(); assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); checkMessageResult(future, "SLEEP"); @@ -277,7 +277,7 @@ public class AsyncRabbitTemplateTests { public void testConvertWithConfirmDirect() throws Exception { this.asyncDirectTemplate.setEnableConfirms(true); RabbitConverterFuture future = this.asyncDirectTemplate.convertSendAndReceive("sleep"); - ListenableFuture confirm = future.getConfirm(); + CompletableFuture confirm = future.getConfirm(); assertThat(confirm).isNotNull(); assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); checkConverterResult(future, "SLEEP"); @@ -289,7 +289,7 @@ public class AsyncRabbitTemplateTests { this.asyncDirectTemplate.setEnableConfirms(true); RabbitMessageFuture future = this.asyncDirectTemplate .sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties())); - ListenableFuture confirm = future.getConfirm(); + CompletableFuture confirm = future.getConfirm(); assertThat(confirm).isNotNull(); assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); checkMessageResult(future, "SLEEP"); @@ -300,9 +300,9 @@ public class AsyncRabbitTemplateTests { @DirtiesContext public void testReceiveTimeout() throws Exception { this.asyncTemplate.setReceiveTimeout(500); - ListenableFuture future = this.asyncTemplate.convertSendAndReceive("noReply"); + CompletableFuture future = this.asyncTemplate.convertSendAndReceive("noReply"); TheCallback callback = new TheCallback(); - future.addCallback(callback); + future.whenComplete(callback); assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1); try { future.get(10, TimeUnit.SECONDS); @@ -323,7 +323,7 @@ public class AsyncRabbitTemplateTests { this.asyncTemplate.setReceiveTimeout(100); RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("sleep"); TheCallback callback = new TheCallback(); - future.addCallback(callback); + future.whenComplete(callback); assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1); try { future.get(10, TimeUnit.SECONDS); @@ -342,7 +342,7 @@ public class AsyncRabbitTemplateTests { * map when it times out. However, there is a small race condition where * the reply arrives at the same time as the timeout. */ - future.set("foo"); + future.complete("foo"); assertThat(callback.result).isNull(); } @@ -353,7 +353,7 @@ public class AsyncRabbitTemplateTests { this.asyncTemplate.setReceiveTimeout(5000); RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("noReply"); TheCallback callback = new TheCallback(); - future.addCallback(callback); + 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 @@ -375,54 +375,76 @@ public class AsyncRabbitTemplateTests { * should never happen because the container is stopped before canceling * and the future is removed from the pending map. */ - future.set("foo"); + future.complete("foo"); assertThat(callback.result).isNull(); } - private void checkConverterResult(ListenableFuture future, String expected) throws InterruptedException { + @Test + void ctorCoverage() { + AsyncRabbitTemplate template = new AsyncRabbitTemplate(mock(ConnectionFactory.class), "ex", "rk"); + assertThat(template).extracting(t -> t.getRabbitTemplate()) + .extracting("exchange") + .isEqualTo("ex"); + assertThat(template).extracting(t -> t.getRabbitTemplate()) + .extracting("routingKey") + .isEqualTo("rk"); + template = new AsyncRabbitTemplate(mock(ConnectionFactory.class), "ex", "rk", "rq"); + assertThat(template).extracting(t -> t.getRabbitTemplate()) + .extracting("exchange") + .isEqualTo("ex"); + assertThat(template).extracting(t -> t.getRabbitTemplate()) + .extracting("routingKey") + .isEqualTo("rk"); + assertThat(template) + .extracting("replyAddress") + .isEqualTo("rq"); + assertThat(template).extracting("container") + .extracting("queueNames") + .isEqualTo(new String[] { "rq" }); + template = new AsyncRabbitTemplate(mock(ConnectionFactory.class), "ex", "rk", "rq", "ra"); + assertThat(template).extracting(t -> t.getRabbitTemplate()) + .extracting("exchange") + .isEqualTo("ex"); + assertThat(template).extracting(t -> t.getRabbitTemplate()) + .extracting("routingKey") + .isEqualTo("rk"); + assertThat(template) + .extracting("replyAddress") + .isEqualTo("ra"); + assertThat(template).extracting("container") + .extracting("queueNames") + .isEqualTo(new String[] { "rq" }); + template = new AsyncRabbitTemplate(mock(RabbitTemplate.class), mock(AbstractMessageListenerContainer.class), + "rq"); + assertThat(template) + .extracting("replyAddress") + .isEqualTo("rq"); + } + + private void checkConverterResult(CompletableFuture future, String expected) throws InterruptedException { final CountDownLatch cdl = new CountDownLatch(1); final AtomicReference resultRef = new AtomicReference<>(); - future.addCallback(new ListenableFutureCallback() { - - @Override - public void onSuccess(String result) { - resultRef.set(result); - cdl.countDown(); - } - - @Override - public void onFailure(Throwable ex) { - cdl.countDown(); - } - + future.whenComplete((result, ex) -> { + resultRef.set(result); + cdl.countDown(); }); assertThat(cdl.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(resultRef.get()).isEqualTo(expected); } - private Message checkMessageResult(ListenableFuture future, String expected) throws InterruptedException { + private Message checkMessageResult(CompletableFuture future, String expected) throws InterruptedException { final CountDownLatch cdl = new CountDownLatch(1); final AtomicReference resultRef = new AtomicReference<>(); - future.addCallback(new ListenableFutureCallback() { - - @Override - public void onSuccess(Message result) { - resultRef.set(result); - cdl.countDown(); - } - - @Override - public void onFailure(Throwable ex) { - cdl.countDown(); - } - + 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 ListenableFutureCallback { + public static class TheCallback implements BiConsumer { private final CountDownLatch latch = new CountDownLatch(1); @@ -430,14 +452,10 @@ public class AsyncRabbitTemplateTests { private volatile Throwable ex; - @Override - public void onSuccess(String result) { - this.result = result; - latch.countDown(); - } @Override - public void onFailure(Throwable ex) { + public void accept(String result, Throwable ex) { + this.result = result; this.ex = ex; latch.countDown(); } 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..459cdb07 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; @@ -36,7 +37,7 @@ 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.RabbitConverterFuture; 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; @@ -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..2b9d3427 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. @@ -24,7 +24,7 @@ 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.RabbitConverterFuture; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; 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..51a70bb9 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. 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..450c1566 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. 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..e5c43a5f 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; @@ -44,8 +45,6 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SettableListenableFuture; import com.rabbitmq.client.Channel; @@ -97,7 +96,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 @@ -109,13 +108,15 @@ public class AsyncReplyToTests { volatile CountDownLatch dmlcLatch = new CountDownLatch(1); @RabbitListener(id = "smlc", queues = "async1", containerFactory = "smlcf") - ListenableFuture listen1(String in, Channel channel) { - return new SettableListenableFuture<>(); + CompletableFuture listen1(String in, Channel channel) { + return new CompletableFuture<>(); } @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 a0931518..fae9a422 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -1224,7 +1224,7 @@ This is no longer necessary since the framework now hands off the callback invoc IMPORTANT: The guarantee of receiving a returned message before the ack is still maintained as long as the return callback executes in 60 seconds or less. The confirm is scheduled to be delivered after the return callback exits or after 60 seconds, whichever comes first. -Starting with version 2.1, the `CorrelationData` object has a `ListenableFuture` that you can use to get the result, instead of using a `ConfirmCallback` on the template. +The `CorrelationData` object has a `CompletableFuture` that you can use to get the result, instead of using a `ConfirmCallback` on the template. The following example shows how to configure a `CorrelationData` instance: ==== @@ -1236,7 +1236,7 @@ assertTrue(cd1.getFuture().get(10, TimeUnit.SECONDS).isAck()); ---- ==== -Since it is a `ListenableFuture`, you can either `get()` the result when ready or add listeners for an asynchronous callback. +Since it is a `CompletableFuture`, you can either `get()` the result when ready or use `whenComplete()` for an asynchronous callback. The `Confirm` object is a simple bean with 2 properties: `ack` and `reason` (for `nack` instances). The reason is not populated for broker-generated `nack` instances. It is populated for `nack` instances generated by the framework (for example, closing the connection while `ack` instances are outstanding). @@ -3569,7 +3569,8 @@ IMPORTANT: Containers created this way are normal `@Bean` instances and are not [[async-returns]] ===== 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. +`@RabbitListener` (and `@RabbitHandler`) methods can be specified with asynchronous return types `CompletableFuture` and `Mono`, letting the reply be sent asynchronously. +`ListenableFuture` is no longer supported; it has been deprecated by Spring Framework. 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. @@ -4555,7 +4556,7 @@ You can also take a look at the `FixedReplyQueueDeadLetterTests` test case for a Version 1.6 introduced the `AsyncRabbitTemplate`. This has similar `sendAndReceive` (and `convertSendAndReceive`) methods to those on the <>. -However, instead of blocking, they return a `ListenableFuture`. +However, instead of blocking, they return a `CompletableFuture`. The `sendAndReceive` methods return a `RabbitMessageFuture`. The `convertSendAndReceive` methods return a `RabbitConverterFuture`. @@ -4575,13 +4576,13 @@ public void doSomeWorkAndGetResultLater() { ... - ListenableFuture future = this.template.convertSendAndReceive("foo"); + CompletableFuture future = this.template.convertSendAndReceive("foo"); // do some more work String reply = null; try { - reply = future.get(); + reply = future.get(10, TimeUnit.SECONDS); } catch (ExecutionException e) { ... @@ -4596,18 +4597,13 @@ public void doSomeWorkAndGetResultAsync() { ... RabbitConverterFuture future = this.template.convertSendAndReceive("foo"); - future.addCallback(new ListenableFutureCallback() { - - @Override - public void onSuccess(String result) { - ... + future.whenComplete((result, ex) -> { + if (ex == null) { + // success } - - @Override - public void onFailure(Throwable ex) { - ... + else { + // failure } - }); ... @@ -4618,7 +4614,7 @@ public void doSomeWorkAndGetResultAsync() { If `mandatory` is set and the message cannot be delivered, the future throws an `ExecutionException` with a cause of `AmqpMessageReturnedException`, which encapsulates the returned message and information about the return. -If `enableConfirms` is set, the future has a property called `confirm`, which is itself a `ListenableFuture` with `true` indicating a successful publish. +If `enableConfirms` is set, the future has a property called `confirm`, which is itself a `CompletableFuture` with `true` indicating a successful publish. If the confirm future is `false`, the `RabbitFuture` has a further property called `nackCause`, which contains the reason for the failure, if available. IMPORTANT: The publisher confirm is discarded if it is received after the reply, since the reply implies a successful publish. @@ -4647,6 +4643,8 @@ Version 2.0 introduced variants of these methods (`convertSendAndReceiveAsType`) You must configure the underlying `RabbitTemplate` with a `SmartMessageConverter`. See <> for more information. +The `AsyncRabbitTemplate2` (added to assist with migration to this release) is now deprecated in favor of `AsyncRabbitTemplate` which now returns `CompletableFuture` s instead of `ListenableFuture` s. + [[remoting]] ===== Spring Remoting with AMQP diff --git a/src/reference/asciidoc/appendix.adoc b/src/reference/asciidoc/appendix.adoc index d1f3bf6d..b7e6ce42 100644 --- a/src/reference/asciidoc/appendix.adoc +++ b/src/reference/asciidoc/appendix.adoc @@ -34,6 +34,16 @@ Support remoting using Spring Framework’s RMI support is deprecated and will b 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. + +==== Message Converter Changes + +The `Jackson2JsonMessageConverter` can now determine the charset from the `contentEncoding` header. +See <> for more information. + ==== Changes in 2.3 Since 2.2 This section describes the changes between version 2.2 and version 2.3. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index fb15a9e0..f355f769 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -10,3 +10,9 @@ This version requires Spring Framework 6.0 and Java 17 ==== Remoting The remoting feature (using RMI) is no longer supported. + +==== AsyncRabbitTemplate + +The `AsyncRabbitTemplate2`, which was added in 2.4.7 to aid migration to this release, is deprecated in favor of `AsyncRabbitTemplate`. +The `AsyncRabbitTemplate` now returns `CompletableFuture` s instead of `ListenableFuture` s. +See <> for more information.