Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1473: Switch to CompletableFuture

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/1473

* Deprecate `AsyncRabbitTemplate2`.
* Copy implementation from  `AsyncRabbitTemplate2` to `AsyncRabbitTemplate`
* Fix other `ListenableFuture` usages.
This commit is contained in:
Gary Russell
2022-07-28 16:08:58 -04:00
committed by GitHub
parent 8becb8925e
commit b03a0bc9af
23 changed files with 596 additions and 328 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Message> sendAndReceive(Message message);
CompletableFuture<Message> sendAndReceive(Message message);
/**
* Send a message to the default exchange with the supplied routing key. If the message
* contains a correlationId property, it must be unique.
* @param routingKey the routing key.
* @param message the message.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
ListenableFuture<Message> sendAndReceive(String routingKey, Message message);
CompletableFuture<Message> sendAndReceive(String routingKey, Message message);
/**
* Send a message to the supplied exchange and routing key. If the message
@@ -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<Message> sendAndReceive(String exchange, String routingKey, Message message);
CompletableFuture<Message> sendAndReceive(String exchange, String routingKey, Message message);
/**
* Convert the object to a message and send it to the default exchange with the
* default routing key.
* @param object the object to convert.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceive(Object object);
<C> CompletableFuture<C> convertSendAndReceive(Object object);
/**
* Convert the object to a message and send it to the default exchange with the
@@ -71,9 +72,9 @@ public interface AsyncAmqpTemplate {
* @param routingKey the routing key.
* @param object the object to convert.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceive(String routingKey, Object object);
<C> CompletableFuture<C> convertSendAndReceive(String routingKey, Object object);
/**
* Convert the object to a message and send it to the provided exchange and
@@ -82,9 +83,9 @@ public interface AsyncAmqpTemplate {
* @param routingKey the routing key.
* @param object the object to convert.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object);
<C> CompletableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object);
/**
* Convert the object to a message and send it to the default exchange with the
@@ -93,9 +94,9 @@ public interface AsyncAmqpTemplate {
* @param object the object to convert.
* @param messagePostProcessor the post processor.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor);
<C> CompletableFuture<C> convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor);
/**
* Convert the object to a message and send it to the default exchange with the
@@ -105,9 +106,9 @@ public interface AsyncAmqpTemplate {
* @param object the object to convert.
* @param messagePostProcessor the post processor.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceive(String routingKey, Object object,
<C> CompletableFuture<C> convertSendAndReceive(String routingKey, Object object,
MessagePostProcessor messagePostProcessor);
/**
@@ -119,9 +120,9 @@ public interface AsyncAmqpTemplate {
* @param object the object to convert.
* @param messagePostProcessor the post processor.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object,
<C> CompletableFuture<C> 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 <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceiveAsType(Object object, ParameterizedTypeReference<C> responseType);
<C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, ParameterizedTypeReference<C> responseType);
/**
* Convert the object to a message and send it to the default exchange with the
@@ -141,9 +142,9 @@ public interface AsyncAmqpTemplate {
* @param object the object to convert.
* @param responseType the response type.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
<C> CompletableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
ParameterizedTypeReference<C> responseType);
/**
@@ -154,9 +155,9 @@ public interface AsyncAmqpTemplate {
* @param object the object to convert.
* @param responseType the response type.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
<C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
ParameterizedTypeReference<C> responseType);
/**
@@ -167,9 +168,9 @@ public interface AsyncAmqpTemplate {
* @param messagePostProcessor the post processor.
* @param responseType the response type.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor,
<C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor,
ParameterizedTypeReference<C> responseType);
/**
@@ -181,9 +182,9 @@ public interface AsyncAmqpTemplate {
* @param messagePostProcessor the post processor.
* @param responseType the response type.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
<C> CompletableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType);
/**
@@ -196,9 +197,9 @@ public interface AsyncAmqpTemplate {
* @param messagePostProcessor the post processor.
* @param responseType the response type.
* @param <C> the expected result type.
* @return the {@link ListenableFuture}.
* @return the {@link CompletableFuture}.
*/
<C> ListenableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
<C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType);
}

192
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java
View File

@@ -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.
* <p>
* 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.
* <p>
@@ -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<Boolean>} in its {@code confirm} property.
* {@code CompletableFuture<Boolean>} 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<Object> rabbitFuture = (RabbitConverterFuture<Object>) 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<Boolean>) 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 <T> the type.
* @since 1.6
*/
public abstract class RabbitFuture<T> extends SettableListenableFuture<T> {
private final String correlationId;
private final Message requestMessage;
private ScheduledFuture<?> timeoutTask;
private volatile ListenableFuture<Boolean> 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<Boolean> getConfirm() {
return this.confirm;
}
void setConfirm(ListenableFuture<Boolean> confirm) {
this.confirm = confirm;
}
/**
* When confirms are enabled and a nack is received, contains
* the cause for the nack, if any.
* @return the cause.
*/
public String getNackCause() {
return this.nackCause;
}
void setNackCause(String nackCause) {
this.nackCause = nackCause;
}
void startTimer() {
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<Message> {
public RabbitMessageFuture(String correlationId, Message requestMessage) {
super(correlationId, requestMessage);
}
}
/**
* A {@link RabbitFuture} with a return type of the template's
* generic parameter.
* @param <C> the type.
* @since 1.6
*/
public class RabbitConverterFuture<C> extends RabbitFuture<C> {
private volatile ParameterizedTypeReference<C> returnType;
public RabbitConverterFuture(String correlationId, Message requestMessage) {
super(correlationId, requestMessage);
}
public ParameterizedTypeReference<C> getReturnType() {
return this.returnType;
}
public void setReturnType(ParameterizedTypeReference<C> returnType) {
this.returnType = returnType;
}
}
private final class CorrelationMessagePostProcessor<C> 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<C>(correlationId, message);
correlationData.future = new RabbitConverterFuture<C>(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);

64
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2.java Normal file
View File

@@ -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);
}
}

54
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitConverterFuture.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* Copyright 2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit;
import java.util.concurrent.ScheduledFuture;
import java.util.function.BiConsumer;
import java.util.function.Function;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
import org.springframework.core.ParameterizedTypeReference;
/**
* A {@link RabbitFuture} with a return type of the template's
* generic parameter.
* @param <C> the type.
*
* @author Gary Russell
* @since 2.4.7
*/
public class RabbitConverterFuture<C> extends RabbitFuture<C> {
private volatile ParameterizedTypeReference<C> returnType;
RabbitConverterFuture(String correlationId, Message requestMessage,
BiConsumer<String, ChannelHolder> canceler,
Function<RabbitFuture<?>, ScheduledFuture<?>> timeoutTaskFunction) {
super(correlationId, requestMessage, canceler, timeoutTaskFunction);
}
public ParameterizedTypeReference<C> getReturnType() {
return this.returnType;
}
public void setReturnType(ParameterizedTypeReference<C> returnType) {
this.returnType = returnType;
}
}

116
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitFuture.java Normal file
View File

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

40
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitMessageFuture.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* Copyright 2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit;
import java.util.concurrent.ScheduledFuture;
import java.util.function.BiConsumer;
import java.util.function.Function;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
/**
* A {@link RabbitFuture} with a return type of {@link Message}.
*
* @author Gary Russell
* @since 2.4.7
*/
public class RabbitMessageFuture extends RabbitFuture<Message> {
RabbitMessageFuture(String correlationId, Message requestMessage, BiConsumer<String, ChannelHolder> canceler,
Function<RabbitFuture<?>, ScheduledFuture<?>> timeoutTaskFunction) {
super(correlationId, requestMessage, canceler, timeoutTaskFunction);
}
}

59
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/TimeoutTask.java Normal file
View File

@@ -0,0 +1,59 @@
/*
* Copyright 2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit;
import java.util.concurrent.ConcurrentMap;
import org.springframework.amqp.core.AmqpReplyTimeoutException;
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
import org.springframework.lang.Nullable;
/**
* A {@link Runnable} used to time out a {@link RabbitFuture}.
*
* @author Gary Russell
* @since 2.4.7
*/
public class TimeoutTask implements Runnable {
private final RabbitFuture<?> future;
private final ConcurrentMap<String, RabbitFuture<?>> pending;
private final DirectReplyToMessageListenerContainer container;
TimeoutTask(RabbitFuture<?> future, ConcurrentMap<String, RabbitFuture<?>> pending,
@Nullable DirectReplyToMessageListenerContainer container) {
this.future = future;
this.pending = pending;
this.container = container;
}
@Override
public void run() {
this.pending.remove(this.future.getCorrelationId());
ChannelHolder holder = this.future.getChannelHolder();
if (holder != null && this.container != null) {
this.container.releaseConsumerFor(holder, false, null); // NOSONAR
}
this.future.completeExceptionally(
new AmqpReplyTimeoutException("Reply timed out", this.future.getRequestMessage()));
}
}

19
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java
View File

@@ -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<Confirm> future = new SettableListenableFuture<>();
private final CompletableFuture<Confirm> future = new CompletableFuture<>();
private volatile String id;
@@ -91,7 +91,18 @@ public class CorrelationData implements Correlation {
* @return the future.
* @since 2.1
*/
public SettableListenableFuture<Confirm> getFuture() {
public CompletableFuture<Confirm> 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<Confirm> getCompletableFuture() {
return this.future;
}

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java
View File

@@ -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
}

17
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java
View File

@@ -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) {

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java
View File

@@ -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());
}
/**

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java
View File

@@ -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());
}
/**

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java
View File

@@ -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) {

148
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java
View File

@@ -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<String> future = this.asyncTemplate.convertSendAndReceive("foo", m -> {
CompletableFuture<String> 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<String> future1 = this.asyncDirectTemplate.convertSendAndReceive("foo");
ListenableFuture<String> future2 = this.asyncDirectTemplate.convertSendAndReceive("bar");
CompletableFuture<String> future1 = this.asyncDirectTemplate.convertSendAndReceive("foo");
CompletableFuture<String> 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<String> future = this.asyncTemplate.convertSendAndReceive(this.requests.getName(), "foo");
CompletableFuture<String> future = this.asyncTemplate.convertSendAndReceive(this.requests.getName(), "foo");
checkConverterResult(future, "FOO");
}
@Test
public void testConvert3Args() throws Exception {
ListenableFuture<String> future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo");
CompletableFuture<String> future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo");
checkConverterResult(future, "FOO");
}
@Test
public void testConvert4Args() throws Exception {
ListenableFuture<String> future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo",
CompletableFuture<String> 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<Message> future = this.asyncTemplate.sendAndReceive(getFooMessage());
CompletableFuture<Message> future = this.asyncTemplate.sendAndReceive(getFooMessage());
checkMessageResult(future, "FOO");
}
@Test
public void testMessage1ArgDirect() throws Exception {
this.latch.set(new CountDownLatch(1));
ListenableFuture<Message> future1 = this.asyncDirectTemplate.sendAndReceive(getFooMessage());
ListenableFuture<Message> future2 = this.asyncDirectTemplate.sendAndReceive(getFooMessage());
CompletableFuture<Message> future1 = this.asyncDirectTemplate.sendAndReceive(getFooMessage());
CompletableFuture<Message> 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<Message> future = this.asyncTemplate.sendAndReceive(this.requests.getName(), getFooMessage());
CompletableFuture<Message> future = this.asyncTemplate.sendAndReceive(this.requests.getName(), getFooMessage());
checkMessageResult(future, "FOO");
}
@Test
public void testMessage3Args() throws Exception {
ListenableFuture<Message> future = this.asyncTemplate.sendAndReceive("", this.requests.getName(),
CompletableFuture<Message> 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<String> future = this.asyncTemplate.convertSendAndReceive("foo");
CompletableFuture<String> 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<Message> future = this.asyncTemplate.sendAndReceive(message);
CompletableFuture<Message> 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<String> future = this.asyncTemplate.convertSendAndReceive(this.requests.getName() + "x",
CompletableFuture<String> 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<String> future = this.asyncDirectTemplate.convertSendAndReceive(this.requests.getName() + "x",
CompletableFuture<String> 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<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
ListenableFuture<Boolean> confirm = future.getConfirm();
CompletableFuture<Boolean> 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<Boolean> confirm = future.getConfirm();
CompletableFuture<Boolean> 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<String> future = this.asyncDirectTemplate.convertSendAndReceive("sleep");
ListenableFuture<Boolean> confirm = future.getConfirm();
CompletableFuture<Boolean> 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<Boolean> confirm = future.getConfirm();
CompletableFuture<Boolean> 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<String> future = this.asyncTemplate.convertSendAndReceive("noReply");
CompletableFuture<String> 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<String> 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<String> 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<String> 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<String> future, String expected) throws InterruptedException {
final CountDownLatch cdl = new CountDownLatch(1);
final AtomicReference<String> resultRef = new AtomicReference<>();
future.addCallback(new ListenableFutureCallback<String>() {
@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<Message> future, String expected) throws InterruptedException {
private Message checkMessageResult(CompletableFuture<Message> future, String expected) throws InterruptedException {
final CountDownLatch cdl = new CountDownLatch(1);
final AtomicReference<Message> resultRef = new AtomicReference<>();
future.addCallback(new ListenableFutureCallback<Message>() {
@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<String> {
public static class TheCallback implements BiConsumer<String, Throwable> {
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();
}

41
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AsyncListenerTests.java
View File

@@ -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<String> listen1(String foo) {
SettableListenableFuture<String> future = new SettableListenableFuture<>();
public CompletableFuture<String> listen1(String foo) {
CompletableFuture<String> 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<Void> listen4(@SuppressWarnings("unused") String foo) {
SettableListenableFuture<Void> future = new SettableListenableFuture<>();
future.set(null);
public CompletableFuture<Void> listen4(@SuppressWarnings("unused") String foo) {
CompletableFuture<Void> future = new CompletableFuture<>();
future.complete(null);
this.latch4.countDown();
return future;
}
@RabbitListener(id = "fiz", queues = "#{queue5.name}")
public ListenableFuture<Void> listen5(@SuppressWarnings("unused") String foo) {
SettableListenableFuture<Void> future = new SettableListenableFuture<>();
future.setException(new AmqpRejectAndDontRequeueException("asyncToDLQ"));
public CompletableFuture<Void> listen5(@SuppressWarnings("unused") String foo) {
CompletableFuture<Void> 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<Void> listen6(@SuppressWarnings("unused") String foo) {
SettableListenableFuture<Void> future = new SettableListenableFuture<>();
future.setException(new IllegalStateException("asyncDefaultToDLQ"));
public CompletableFuture<Void> listen6(@SuppressWarnings("unused") String foo) {
CompletableFuture<Void> 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<String> listen7(@SuppressWarnings("unused") String foo) {
SettableListenableFuture<String> future = new SettableListenableFuture<>();
public CompletableFuture<String> listen7(@SuppressWarnings("unused") String foo) {
CompletableFuture<String> 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;
}

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java
View File

@@ -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;

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java
View File

@@ -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.

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java
View File

@@ -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.

17
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/AsyncReplyToTests.java
View File

@@ -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<String> listen1(String in, Channel channel) {
return new SettableListenableFuture<>();
CompletableFuture<String> listen1(String in, Channel channel) {
return new CompletableFuture<>();
}
@RabbitListener(id = "dmlc", queues = "async2", containerFactory = "dmlcf")
ListenableFuture<String> listen2(String in, Channel channel) {
return new SettableListenableFuture<>();
CompletableFuture<String> listen2(String in, Channel channel) {
CompletableFuture<String> future = new CompletableFuture<>();
future.complete("test");
return future;
}
@Bean

13
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java
View File

@@ -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<String> myPojoMessageMethod(String input) {
SettableListenableFuture<String> future = new SettableListenableFuture<>();
future.set("processed" + input);
public CompletableFuture<String> myPojoMessageMethod(String input) {
CompletableFuture<String> future = new CompletableFuture<>();
future.complete("processed" + input);
return future;
}

32
src/reference/asciidoc/amqp.adoc
View File

@@ -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<Confirm>`, you can either `get()` the result when ready or add listeners for an asynchronous callback.
Since it is a `CompletableFuture<Confirm>`, 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 <<amqp-template>>.
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<String> future = this.template.convertSendAndReceive("foo");
CompletableFuture<String> 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<String> future = this.template.convertSendAndReceive("foo");
future.addCallback(new ListenableFutureCallback<String>() {
@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<Boolean>` with `true` indicating a successful publish.
If `enableConfirms` is set, the future has a property called `confirm`, which is itself a `CompletableFuture<Boolean>` 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 <<json-complex>> 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

10
src/reference/asciidoc/appendix.adoc
View File

@@ -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 <<json-message-converter>> for more information.
==== AsyncRabbitTemplate
The `AsyncRabbitTemplate` is deprecated in favor of `AsyncRabbitTemplate2` which returns `CompletableFuture` s instead of `ListenableFuture` s.
See <<async-template>> for more information.
==== Message Converter Changes
The `Jackson2JsonMessageConverter` can now determine the charset from the `contentEncoding` header.
See <<json-message-converter>> for more information.
==== Changes in 2.3 Since 2.2
This section describes the changes between version 2.2 and version 2.3.

6
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<async-template>> for more information.