GH-1473: Switch to CompletableFuture
Resolves https://github.com/spring-projects/spring-amqp/issues/1473 Given the stability of the project, it was simplest to copy the `AsyncRabbitTemplate` rather than adding a lot of conditional code. **2.4.x only; I will submit a separate PR for main**
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
37109ad07d
commit
439ccd174c
@@ -0,0 +1,205 @@
|
||||
/*
|
||||
* Copyright 2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.amqp.core;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
import org.springframework.core.ParameterizedTypeReference;
|
||||
|
||||
/**
|
||||
* Classes implementing this interface can perform asynchronous send and
|
||||
* receive operations using {@link CompletableFuture}s.
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @since 2.4.7
|
||||
*
|
||||
*/
|
||||
public interface AsyncAmqpTemplate2 {
|
||||
|
||||
/**
|
||||
* Send a message to the default exchange with the default routing key. If the message
|
||||
* contains a correlationId property, it must be unique.
|
||||
* @param message the message.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
CompletableFuture<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 CompletableFuture}.
|
||||
*/
|
||||
CompletableFuture<Message> sendAndReceive(String routingKey, Message message);
|
||||
|
||||
/**
|
||||
* Send a message to the supplied exchange and routing key. If the message
|
||||
* contains a correlationId property, it must be unique.
|
||||
* @param exchange the exchange.
|
||||
* @param routingKey the routing key.
|
||||
* @param message the message.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
CompletableFuture<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 CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceive(Object object);
|
||||
|
||||
/**
|
||||
* Convert the object to a message and send it to the default exchange with the
|
||||
* provided routing key.
|
||||
* @param routingKey the routing key.
|
||||
* @param object the object to convert.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceive(String routingKey, Object object);
|
||||
|
||||
/**
|
||||
* Convert the object to a message and send it to the provided exchange and
|
||||
* routing key.
|
||||
* @param exchange the exchange.
|
||||
* @param routingKey the routing key.
|
||||
* @param object the object to convert.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<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
|
||||
* default routing key after invoking the {@link MessagePostProcessor}.
|
||||
* If the post processor adds a correlationId property, it must be unique.
|
||||
* @param object the object to convert.
|
||||
* @param messagePostProcessor the post processor.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor);
|
||||
|
||||
/**
|
||||
* Convert the object to a message and send it to the default exchange with the
|
||||
* provided routing key after invoking the {@link MessagePostProcessor}.
|
||||
* If the post processor adds a correlationId property, it must be unique.
|
||||
* @param routingKey the routing key.
|
||||
* @param object the object to convert.
|
||||
* @param messagePostProcessor the post processor.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceive(String routingKey, Object object,
|
||||
MessagePostProcessor messagePostProcessor);
|
||||
|
||||
/**
|
||||
* Convert the object to a message and send it to the provided exchange and
|
||||
* routing key after invoking the {@link MessagePostProcessor}.
|
||||
* If the post processor adds a correlationId property, it must be unique.
|
||||
* @param exchange the exchange
|
||||
* @param routingKey the routing key.
|
||||
* @param object the object to convert.
|
||||
* @param messagePostProcessor the post processor.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object,
|
||||
MessagePostProcessor messagePostProcessor);
|
||||
|
||||
/**
|
||||
* Convert the object to a message and send it to the default exchange with the
|
||||
* default routing key.
|
||||
* @param object the object to convert.
|
||||
* @param responseType the response type.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, ParameterizedTypeReference<C> responseType);
|
||||
|
||||
/**
|
||||
* Convert the object to a message and send it to the default exchange with the
|
||||
* provided routing key.
|
||||
* @param routingKey the routing key.
|
||||
* @param object the object to convert.
|
||||
* @param responseType the response type.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
|
||||
ParameterizedTypeReference<C> responseType);
|
||||
|
||||
/**
|
||||
* Convert the object to a message and send it to the provided exchange and
|
||||
* routing key.
|
||||
* @param exchange the exchange.
|
||||
* @param routingKey the routing key.
|
||||
* @param object the object to convert.
|
||||
* @param responseType the response type.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
|
||||
ParameterizedTypeReference<C> responseType);
|
||||
|
||||
/**
|
||||
* Convert the object to a message and send it to the default exchange with the
|
||||
* default routing key after invoking the {@link MessagePostProcessor}.
|
||||
* If the post processor adds a correlationId property, it must be unique.
|
||||
* @param object the object to convert.
|
||||
* @param messagePostProcessor the post processor.
|
||||
* @param responseType the response type.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor,
|
||||
ParameterizedTypeReference<C> responseType);
|
||||
|
||||
/**
|
||||
* Convert the object to a message and send it to the default exchange with the
|
||||
* provided routing key after invoking the {@link MessagePostProcessor}.
|
||||
* If the post processor adds a correlationId property, it must be unique.
|
||||
* @param routingKey the routing key.
|
||||
* @param object the object to convert.
|
||||
* @param messagePostProcessor the post processor.
|
||||
* @param responseType the response type.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
|
||||
MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType);
|
||||
|
||||
/**
|
||||
* Convert the object to a message and send it to the provided exchange and
|
||||
* routing key after invoking the {@link MessagePostProcessor}.
|
||||
* If the post processor adds a correlationId property, it must be unique.
|
||||
* @param exchange the exchange
|
||||
* @param routingKey the routing key.
|
||||
* @param object the object to convert.
|
||||
* @param messagePostProcessor the post processor.
|
||||
* @param responseType the response type.
|
||||
* @param <C> the expected result type.
|
||||
* @return the {@link CompletableFuture}.
|
||||
*/
|
||||
<C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
|
||||
MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType);
|
||||
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2016-2021 the original author or authors.
|
||||
* Copyright 2016-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -89,7 +89,9 @@ import com.rabbitmq.client.Channel;
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 1.6
|
||||
* @deprecated in favor of {@link AsyncRabbitTemplate2}.
|
||||
*/
|
||||
@Deprecated
|
||||
public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessageListener, ReturnsCallback,
|
||||
ConfirmCallback, BeanNameAware, SmartLifecycle {
|
||||
|
||||
|
||||
@@ -0,0 +1,857 @@
|
||||
/*
|
||||
* Copyright 2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.amqp.rabbit;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.amqp.AmqpException;
|
||||
import org.springframework.amqp.AmqpIllegalStateException;
|
||||
import org.springframework.amqp.core.Address;
|
||||
import org.springframework.amqp.core.AmqpMessageReturnedException;
|
||||
import org.springframework.amqp.core.AmqpReplyTimeoutException;
|
||||
import org.springframework.amqp.core.AsyncAmqpTemplate2;
|
||||
import org.springframework.amqp.core.Correlation;
|
||||
import org.springframework.amqp.core.Message;
|
||||
import org.springframework.amqp.core.MessagePostProcessor;
|
||||
import org.springframework.amqp.core.MessageProperties;
|
||||
import org.springframework.amqp.core.ReturnedMessage;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitFuture;
|
||||
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
|
||||
import org.springframework.amqp.rabbit.connection.CorrelationData;
|
||||
import org.springframework.amqp.rabbit.connection.PublisherCallbackChannel;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnsCallback;
|
||||
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
|
||||
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer;
|
||||
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
|
||||
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
|
||||
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
|
||||
import org.springframework.amqp.support.converter.MessageConverter;
|
||||
import org.springframework.amqp.support.converter.SmartMessageConverter;
|
||||
import org.springframework.amqp.utils.JavaUtils;
|
||||
import org.springframework.beans.factory.BeanNameAware;
|
||||
import org.springframework.context.SmartLifecycle;
|
||||
import org.springframework.core.ParameterizedTypeReference;
|
||||
import org.springframework.expression.Expression;
|
||||
import org.springframework.lang.NonNull;
|
||||
import org.springframework.lang.Nullable;
|
||||
import org.springframework.scheduling.TaskScheduler;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import com.rabbitmq.client.Channel;
|
||||
|
||||
/**
|
||||
* Provides asynchronous send and receive operations returning a {@link CompletableFuture}
|
||||
* allowing the caller to obtain the reply later, using {@code get()} or a callback.
|
||||
* <p>
|
||||
* When confirms are enabled, the future has a confirm property which is itself a
|
||||
* {@link CompletableFuture}. If the reply is received before the publisher confirm,
|
||||
* the confirm is discarded since the reply implicitly indicates the message was
|
||||
* published.
|
||||
* <p>
|
||||
* Returned (undeliverable) request messages are presented as a
|
||||
* {@link AmqpMessageReturnedException} cause of an
|
||||
* {@link java.util.concurrent.ExecutionException}.
|
||||
* <p>
|
||||
* Internally, the template uses a {@link RabbitTemplate} and an
|
||||
* {@link AbstractMessageListenerContainer} either provided or constructed internally
|
||||
* (a {@link SimpleMessageListenerContainer}).
|
||||
* If an external {@link RabbitTemplate} is provided and confirms/returns are enabled,
|
||||
* it must not previously have had callbacks registered because this object needs to
|
||||
* be the callback.
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 1.6
|
||||
*/
|
||||
public class AsyncRabbitTemplate2 implements AsyncAmqpTemplate2, ChannelAwareMessageListener, ReturnsCallback,
|
||||
ConfirmCallback, BeanNameAware, SmartLifecycle {
|
||||
|
||||
public static final int DEFAULT_RECEIVE_TIMEOUT = 30000;
|
||||
|
||||
private final Log logger = LogFactory.getLog(this.getClass());
|
||||
|
||||
private final RabbitTemplate template;
|
||||
|
||||
private final AbstractMessageListenerContainer container;
|
||||
|
||||
private final DirectReplyToMessageListenerContainer directReplyToContainer;
|
||||
|
||||
private final String replyAddress;
|
||||
|
||||
private final ConcurrentMap<String, RabbitFuture2<?>> pending = new ConcurrentHashMap<>();
|
||||
|
||||
private final CorrelationMessagePostProcessor<?> messagePostProcessor = new CorrelationMessagePostProcessor<>();
|
||||
|
||||
private volatile boolean running;
|
||||
|
||||
private volatile boolean enableConfirms;
|
||||
|
||||
private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
|
||||
|
||||
private int phase;
|
||||
|
||||
private boolean autoStartup = true;
|
||||
|
||||
private String beanName;
|
||||
|
||||
private TaskScheduler taskScheduler;
|
||||
|
||||
private boolean internalTaskScheduler = true;
|
||||
|
||||
/**
|
||||
* Construct an instance using the provided arguments. Replies will be
|
||||
* routed to the default exchange using the reply queue name as the routing
|
||||
* key.
|
||||
* @param connectionFactory the connection factory.
|
||||
* @param exchange the default exchange to which requests will be sent.
|
||||
* @param routingKey the default routing key.
|
||||
* @param replyQueue the name of the reply queue to listen for replies.
|
||||
*/
|
||||
public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey,
|
||||
String replyQueue) {
|
||||
this(connectionFactory, exchange, routingKey, replyQueue, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct an instance using the provided arguments. If 'replyAddress' is null,
|
||||
* replies will be routed to the default exchange using the reply queue name as the
|
||||
* routing key. Otherwise it should have the form exchange/routingKey and must
|
||||
* cause messages to be routed to the reply queue.
|
||||
* @param connectionFactory the connection factory.
|
||||
* @param exchange the default exchange to which requests will be sent.
|
||||
* @param routingKey the default routing key.
|
||||
* @param replyQueue the name of the reply queue to listen for replies.
|
||||
* @param replyAddress the reply address (exchange/routingKey).
|
||||
*/
|
||||
public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey,
|
||||
String replyQueue, String replyAddress) {
|
||||
Assert.notNull(connectionFactory, "'connectionFactory' cannot be null");
|
||||
Assert.notNull(routingKey, "'routingKey' cannot be null");
|
||||
Assert.notNull(replyQueue, "'replyQueue' cannot be null");
|
||||
this.template = new RabbitTemplate(connectionFactory);
|
||||
this.template.setExchange(exchange == null ? "" : exchange);
|
||||
this.template.setRoutingKey(routingKey);
|
||||
this.container = new SimpleMessageListenerContainer(connectionFactory);
|
||||
JavaUtils.INSTANCE
|
||||
.acceptIfNotNull(this.template.getAfterReceivePostProcessors(),
|
||||
(value) -> this.container.setAfterReceivePostProcessors(
|
||||
value.toArray(new MessagePostProcessor[0])));
|
||||
this.container.setQueueNames(replyQueue);
|
||||
this.container.setMessageListener(this);
|
||||
this.container.afterPropertiesSet();
|
||||
this.directReplyToContainer = null;
|
||||
if (replyAddress == null) {
|
||||
this.replyAddress = replyQueue;
|
||||
}
|
||||
else {
|
||||
this.replyAddress = replyAddress;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct an instance using the provided arguments. The first queue the container
|
||||
* is configured to listen to will be used as the reply queue. Replies will be
|
||||
* routed using the default exchange with that queue name as the routing key.
|
||||
* @param template a {@link RabbitTemplate}
|
||||
* @param container a {@link AbstractMessageListenerContainer}.
|
||||
*/
|
||||
public AsyncRabbitTemplate2(RabbitTemplate template, AbstractMessageListenerContainer container) {
|
||||
this(template, container, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct an instance using the provided arguments. The first queue the container
|
||||
* is configured to listen to will be used as the reply queue. If 'replyAddress' is
|
||||
* null, replies will be routed using the default exchange with that queue name as the
|
||||
* routing key. Otherwise it should have the form exchange/routingKey and must
|
||||
* cause messages to be routed to the reply queue.
|
||||
* @param template a {@link RabbitTemplate}.
|
||||
* @param container a {@link AbstractMessageListenerContainer}.
|
||||
* @param replyAddress the reply address.
|
||||
*/
|
||||
public AsyncRabbitTemplate2(RabbitTemplate template, AbstractMessageListenerContainer container,
|
||||
String replyAddress) {
|
||||
Assert.notNull(template, "'template' cannot be null");
|
||||
Assert.notNull(container, "'container' cannot be null");
|
||||
this.template = template;
|
||||
this.container = container;
|
||||
this.container.setMessageListener(this);
|
||||
this.directReplyToContainer = null;
|
||||
if (replyAddress == null) {
|
||||
this.replyAddress = container.getQueueNames()[0];
|
||||
}
|
||||
else {
|
||||
this.replyAddress = replyAddress;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct an instance using the provided arguments. "Direct replyTo" is used for
|
||||
* replies.
|
||||
* @param connectionFactory the connection factory.
|
||||
* @param exchange the default exchange to which requests will be sent.
|
||||
* @param routingKey the default routing key.
|
||||
* @since 2.0
|
||||
*/
|
||||
public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey) {
|
||||
this(new RabbitTemplate(connectionFactory));
|
||||
Assert.notNull(routingKey, "'routingKey' cannot be null");
|
||||
this.template.setExchange(exchange == null ? "" : exchange);
|
||||
this.template.setRoutingKey(routingKey);
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct an instance using the provided arguments. "Direct replyTo" is used for
|
||||
* replies.
|
||||
* @param template a {@link RabbitTemplate}
|
||||
* @since 2.0
|
||||
*/
|
||||
public AsyncRabbitTemplate2(RabbitTemplate template) {
|
||||
Assert.notNull(template, "'template' cannot be null");
|
||||
this.template = template;
|
||||
this.container = null;
|
||||
this.replyAddress = null;
|
||||
this.directReplyToContainer = new DirectReplyToMessageListenerContainer(this.template.getConnectionFactory());
|
||||
JavaUtils.INSTANCE
|
||||
.acceptIfNotNull(template.getAfterReceivePostProcessors(),
|
||||
(value) -> this.directReplyToContainer.setAfterReceivePostProcessors(
|
||||
value.toArray(new MessagePostProcessor[0])));
|
||||
this.directReplyToContainer.setMessageListener(this);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param autoStartup true for auto start.
|
||||
* @see #isAutoStartup()
|
||||
*/
|
||||
public void setAutoStartup(boolean autoStartup) {
|
||||
this.autoStartup = autoStartup;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param phase the phase.
|
||||
* @see #getPhase()
|
||||
*/
|
||||
public void setPhase(int phase) {
|
||||
this.phase = phase;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set to true to enable the receipt of returned messages that cannot be delivered
|
||||
* in the form of a {@link AmqpMessageReturnedException}.
|
||||
* @param mandatory true to enable returns.
|
||||
*/
|
||||
public void setMandatory(boolean mandatory) {
|
||||
this.template.setReturnsCallback(this);
|
||||
this.template.setMandatory(mandatory);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mandatoryExpression a SpEL {@link Expression} to evaluate against each request
|
||||
* message. The result of the evaluation must be a {@code boolean} value.
|
||||
* @since 2.0
|
||||
*/
|
||||
public void setMandatoryExpression(Expression mandatoryExpression) {
|
||||
this.template.setReturnsCallback(this);
|
||||
this.template.setMandatoryExpression(mandatoryExpression);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mandatoryExpression a SpEL {@link Expression} to evaluate against each request
|
||||
* message. The result of the evaluation must be a {@code boolean} value.
|
||||
* @since 2.0
|
||||
*/
|
||||
public void setMandatoryExpressionString(String mandatoryExpression) {
|
||||
this.template.setReturnsCallback(this);
|
||||
this.template.setMandatoryExpressionString(mandatoryExpression);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set to true to enable publisher confirms. When enabled, the {@link RabbitFuture}
|
||||
* returned by the send and receive operation will have a
|
||||
* {@code CompletableFuture<Boolean>} in its {@code confirm} property.
|
||||
* @param enableConfirms true to enable publisher confirms.
|
||||
*/
|
||||
public void setEnableConfirms(boolean enableConfirms) {
|
||||
this.enableConfirms = enableConfirms;
|
||||
if (enableConfirms) {
|
||||
this.template.setConfirmCallback(this);
|
||||
}
|
||||
}
|
||||
|
||||
public String getBeanName() {
|
||||
return this.beanName;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setBeanName(String beanName) {
|
||||
this.beanName = beanName;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return a reference to the underlying connection factory in the
|
||||
* {@link RabbitTemplate}.
|
||||
*/
|
||||
public ConnectionFactory getConnectionFactory() {
|
||||
return this.template.getConnectionFactory();
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the receive timeout - the future returned by the send and receive
|
||||
* methods will be canceled when this timeout expires. {@code <= 0} means
|
||||
* futures never expire. Beware that this will cause a memory leak if a
|
||||
* reply is not received. Default: 30000 (30 seconds).
|
||||
* @param receiveTimeout the timeout in milliseconds.
|
||||
*/
|
||||
public void setReceiveTimeout(long receiveTimeout) {
|
||||
this.receiveTimeout = receiveTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the task scheduler to expire timed out futures.
|
||||
* @param taskScheduler the task scheduler
|
||||
* @see #setReceiveTimeout(long)
|
||||
*/
|
||||
public synchronized void setTaskScheduler(TaskScheduler taskScheduler) {
|
||||
Assert.notNull(taskScheduler, "'taskScheduler' cannot be null");
|
||||
this.internalTaskScheduler = false;
|
||||
this.taskScheduler = taskScheduler;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return a reference to the underlying {@link RabbitTemplate}'s
|
||||
* {@link MessageConverter}.
|
||||
*/
|
||||
public MessageConverter getMessageConverter() {
|
||||
return this.template.getMessageConverter();
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the underlying {@link RabbitTemplate} used for sending.
|
||||
* @return the template.
|
||||
* @since 2.2
|
||||
*/
|
||||
public RabbitTemplate getRabbitTemplate() {
|
||||
return this.template;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RabbitMessageFuture2 sendAndReceive(Message message) {
|
||||
return sendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), message);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RabbitMessageFuture2 sendAndReceive(String routingKey, Message message) {
|
||||
return sendAndReceive(this.template.getExchange(), routingKey, message);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RabbitMessageFuture2 sendAndReceive(String exchange, String routingKey, Message message) {
|
||||
String correlationId = getOrSetCorrelationIdAndSetReplyTo(message, null);
|
||||
RabbitMessageFuture2 future = new RabbitMessageFuture2(correlationId, message);
|
||||
CorrelationData correlationData = null;
|
||||
if (this.enableConfirms) {
|
||||
correlationData = new CorrelationData(correlationId);
|
||||
future.setConfirm(new CompletableFuture<>());
|
||||
}
|
||||
this.pending.put(correlationId, future);
|
||||
if (this.container != null) {
|
||||
this.template.send(exchange, routingKey, message, correlationData);
|
||||
}
|
||||
else {
|
||||
ChannelHolder channelHolder = this.directReplyToContainer.getChannelHolder();
|
||||
future.setChannelHolder(channelHolder);
|
||||
sendDirect(channelHolder.getChannel(), exchange, routingKey, message, correlationData);
|
||||
}
|
||||
future.startTimer();
|
||||
return future;
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceive(Object object) {
|
||||
return convertSendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), object, null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceive(String routingKey, Object object) {
|
||||
return convertSendAndReceive(this.template.getExchange(), routingKey, object, null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceive(String exchange, String routingKey, Object object) {
|
||||
return convertSendAndReceive(exchange, routingKey, object, null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceive(Object object,
|
||||
MessagePostProcessor messagePostProcessor) {
|
||||
return convertSendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), object,
|
||||
messagePostProcessor);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceive(String routingKey, Object object,
|
||||
MessagePostProcessor messagePostProcessor) {
|
||||
return convertSendAndReceive(this.template.getExchange(), routingKey, object, messagePostProcessor);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceive(String exchange, String routingKey, Object object,
|
||||
MessagePostProcessor messagePostProcessor) {
|
||||
return convertSendAndReceive(exchange, routingKey, object, messagePostProcessor, null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceiveAsType(Object object,
|
||||
ParameterizedTypeReference<C> responseType) {
|
||||
return convertSendAndReceiveAsType(this.template.getExchange(), this.template.getRoutingKey(), object,
|
||||
null, responseType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceiveAsType(String routingKey, Object object,
|
||||
ParameterizedTypeReference<C> responseType) {
|
||||
return convertSendAndReceiveAsType(this.template.getExchange(), routingKey, object, null, responseType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
|
||||
ParameterizedTypeReference<C> responseType) {
|
||||
return convertSendAndReceiveAsType(exchange, routingKey, object, null, responseType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceiveAsType(Object object,
|
||||
MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType) {
|
||||
return convertSendAndReceiveAsType(this.template.getExchange(), this.template.getRoutingKey(), object,
|
||||
messagePostProcessor, responseType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceiveAsType(String routingKey, Object object,
|
||||
MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType) {
|
||||
return convertSendAndReceiveAsType(this.template.getExchange(), routingKey, object, messagePostProcessor,
|
||||
responseType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <C> RabbitConverterFuture2<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
|
||||
MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType) {
|
||||
Assert.state(this.template.getMessageConverter() instanceof SmartMessageConverter,
|
||||
"template's message converter must be a SmartMessageConverter");
|
||||
return convertSendAndReceive(exchange, routingKey, object, messagePostProcessor, responseType);
|
||||
}
|
||||
|
||||
private <C> RabbitConverterFuture2<C> convertSendAndReceive(String exchange, String routingKey, Object object,
|
||||
MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType) {
|
||||
|
||||
AsyncCorrelationData<C> correlationData = new AsyncCorrelationData<C>(messagePostProcessor, responseType,
|
||||
this.enableConfirms);
|
||||
if (this.container != null) {
|
||||
this.template.convertAndSend(exchange, routingKey, object, this.messagePostProcessor, correlationData);
|
||||
}
|
||||
else {
|
||||
MessageConverter converter = this.template.getMessageConverter();
|
||||
if (converter == null) {
|
||||
throw new AmqpIllegalStateException(
|
||||
"No 'messageConverter' specified. Check configuration of RabbitTemplate.");
|
||||
}
|
||||
Message message = converter.toMessage(object, new MessageProperties());
|
||||
this.messagePostProcessor.postProcessMessage(message, correlationData,
|
||||
this.template.nullSafeExchange(exchange), this.template.nullSafeRoutingKey(routingKey));
|
||||
ChannelHolder channelHolder = this.directReplyToContainer.getChannelHolder();
|
||||
correlationData.future.setChannelHolder(channelHolder);
|
||||
sendDirect(channelHolder.getChannel(), exchange, routingKey, message, correlationData);
|
||||
}
|
||||
RabbitConverterFuture2<C> future = correlationData.future;
|
||||
future.startTimer();
|
||||
return future;
|
||||
}
|
||||
|
||||
private void sendDirect(Channel channel, String exchange, String routingKey, Message message,
|
||||
CorrelationData correlationData) {
|
||||
message.getMessageProperties().setReplyTo(Address.AMQ_RABBITMQ_REPLY_TO);
|
||||
try {
|
||||
if (channel instanceof PublisherCallbackChannel) {
|
||||
this.template.addListener(channel);
|
||||
}
|
||||
this.template.doSend(channel, exchange, routingKey, message, this.template.isMandatoryFor(message),
|
||||
correlationData);
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new AmqpException("Failed to send request", e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void start() {
|
||||
if (!this.running) {
|
||||
if (this.internalTaskScheduler) {
|
||||
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
|
||||
scheduler.setThreadNamePrefix(getBeanName() == null ? "asyncTemplate-" : (getBeanName() + "-"));
|
||||
scheduler.afterPropertiesSet();
|
||||
this.taskScheduler = scheduler;
|
||||
}
|
||||
if (this.container != null) {
|
||||
this.container.start();
|
||||
}
|
||||
if (this.directReplyToContainer != null) {
|
||||
this.directReplyToContainer.setTaskScheduler(this.taskScheduler);
|
||||
this.directReplyToContainer.start();
|
||||
}
|
||||
}
|
||||
this.running = true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void stop() {
|
||||
if (this.running) {
|
||||
if (this.container != null) {
|
||||
this.container.stop();
|
||||
}
|
||||
if (this.directReplyToContainer != null) {
|
||||
this.directReplyToContainer.stop();
|
||||
}
|
||||
for (RabbitFuture2<?> future : this.pending.values()) {
|
||||
future.setNackCause("AsyncRabbitTemplate was stopped while waiting for reply");
|
||||
future.cancel(true);
|
||||
}
|
||||
if (this.internalTaskScheduler) {
|
||||
((ThreadPoolTaskScheduler) this.taskScheduler).destroy();
|
||||
this.taskScheduler = null;
|
||||
}
|
||||
}
|
||||
this.running = false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return this.running;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getPhase() {
|
||||
return this.phase;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAutoStartup() {
|
||||
return this.autoStartup;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Override
|
||||
public void onMessage(Message message, Channel channel) {
|
||||
MessageProperties messageProperties = message.getMessageProperties();
|
||||
if (messageProperties != null) {
|
||||
String correlationId = messageProperties.getCorrelationId();
|
||||
if (StringUtils.hasText(correlationId)) {
|
||||
if (this.logger.isDebugEnabled()) {
|
||||
this.logger.debug("onMessage: " + message);
|
||||
}
|
||||
RabbitFuture2<?> future = this.pending.remove(correlationId);
|
||||
if (future != null) {
|
||||
if (future instanceof AsyncRabbitTemplate2.RabbitConverterFuture2) {
|
||||
MessageConverter messageConverter = this.template.getMessageConverter();
|
||||
RabbitConverterFuture2<Object> rabbitFuture = (RabbitConverterFuture2<Object>) future;
|
||||
Object converted = rabbitFuture.getReturnType() != null
|
||||
&& messageConverter instanceof SmartMessageConverter
|
||||
? ((SmartMessageConverter) messageConverter).fromMessage(message,
|
||||
rabbitFuture.getReturnType())
|
||||
: messageConverter.fromMessage(message);
|
||||
rabbitFuture.complete(converted);
|
||||
}
|
||||
else {
|
||||
((RabbitMessageFuture2) future).complete(message);
|
||||
}
|
||||
}
|
||||
else {
|
||||
if (this.logger.isWarnEnabled()) {
|
||||
this.logger.warn("No pending reply - perhaps timed out: " + message);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void returnedMessage(ReturnedMessage returned) {
|
||||
MessageProperties messageProperties = returned.getMessage().getMessageProperties();
|
||||
String correlationId = messageProperties.getCorrelationId();
|
||||
if (StringUtils.hasText(correlationId)) {
|
||||
RabbitFuture2<?> future = this.pending.remove(correlationId);
|
||||
if (future != null) {
|
||||
future.completeExceptionally(new AmqpMessageReturnedException("Message returned", returned));
|
||||
}
|
||||
else {
|
||||
if (this.logger.isWarnEnabled()) {
|
||||
this.logger
|
||||
.warn("No pending reply - perhaps timed out? Message returned: " + returned.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void confirm(@NonNull CorrelationData correlationData, boolean ack, @Nullable String cause) {
|
||||
if (this.logger.isDebugEnabled()) {
|
||||
this.logger.debug("Confirm: " + correlationData + ", ack=" + ack
|
||||
+ (cause == null ? "" : (", cause: " + cause)));
|
||||
}
|
||||
String correlationId = correlationData.getId();
|
||||
if (correlationId != null) {
|
||||
RabbitFuture2<?> future = this.pending.get(correlationId);
|
||||
if (future != null) {
|
||||
future.setNackCause(cause);
|
||||
future.getConfirm().complete(ack);
|
||||
}
|
||||
else {
|
||||
if (this.logger.isDebugEnabled()) {
|
||||
this.logger.debug("Confirm: " + correlationData + ", ack=" + ack
|
||||
+ (cause == null ? "" : (", cause: " + cause))
|
||||
+ " no pending future - either canceled or the reply is already received");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private String getOrSetCorrelationIdAndSetReplyTo(Message message,
|
||||
@Nullable AsyncCorrelationData<?> correlationData) {
|
||||
|
||||
String correlationId;
|
||||
MessageProperties messageProperties = message.getMessageProperties();
|
||||
Assert.notNull(messageProperties, "the message properties cannot be null");
|
||||
String currentCorrelationId = messageProperties.getCorrelationId();
|
||||
if (!StringUtils.hasText(currentCorrelationId)) {
|
||||
correlationId = correlationData != null ? correlationData.getId() : UUID.randomUUID().toString();
|
||||
messageProperties.setCorrelationId(correlationId);
|
||||
Assert.isNull(messageProperties.getReplyTo(), "'replyTo' property must be null");
|
||||
}
|
||||
else {
|
||||
correlationId = currentCorrelationId;
|
||||
}
|
||||
messageProperties.setReplyTo(this.replyAddress);
|
||||
return correlationId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return this.beanName == null ? super.toString() : (this.getClass().getSimpleName() + ": " + this.beanName);
|
||||
}
|
||||
|
||||
/**
|
||||
* Base class for {@link CompletableFuture}s returned by {@link AsyncRabbitTemplate2}.
|
||||
* @param <T> the type.
|
||||
* @since 1.6
|
||||
*/
|
||||
public abstract class RabbitFuture2<T> extends CompletableFuture<T> {
|
||||
|
||||
private final String correlationId;
|
||||
|
||||
private final Message requestMessage;
|
||||
|
||||
private ScheduledFuture<?> timeoutTask;
|
||||
|
||||
private volatile CompletableFuture<Boolean> confirm;
|
||||
|
||||
private String nackCause;
|
||||
|
||||
private ChannelHolder channelHolder;
|
||||
|
||||
public RabbitFuture2(String correlationId, Message requestMessage) {
|
||||
this.correlationId = correlationId;
|
||||
this.requestMessage = requestMessage;
|
||||
}
|
||||
|
||||
void setChannelHolder(ChannelHolder channel) {
|
||||
this.channelHolder = channel;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean cancel(boolean mayInterruptIfRunning) {
|
||||
if (this.timeoutTask != null) {
|
||||
this.timeoutTask.cancel(true);
|
||||
}
|
||||
AsyncRabbitTemplate2.this.pending.remove(this.correlationId);
|
||||
if (this.channelHolder != null && AsyncRabbitTemplate2.this.directReplyToContainer != null) {
|
||||
AsyncRabbitTemplate2.this.directReplyToContainer
|
||||
.releaseConsumerFor(this.channelHolder, false, null); // NOSONAR
|
||||
}
|
||||
return super.cancel(mayInterruptIfRunning);
|
||||
}
|
||||
|
||||
/**
|
||||
* When confirms are enabled contains a {@link CompletableFuture}
|
||||
* for the confirmation.
|
||||
* @return the future.
|
||||
*/
|
||||
public CompletableFuture<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() {
|
||||
if (AsyncRabbitTemplate2.this.receiveTimeout > 0) {
|
||||
synchronized (AsyncRabbitTemplate2.this) {
|
||||
if (!AsyncRabbitTemplate2.this.running) {
|
||||
AsyncRabbitTemplate2.this.pending.remove(this.correlationId);
|
||||
throw new IllegalStateException("'AsyncRabbitTemplate' must be started.");
|
||||
}
|
||||
this.timeoutTask = AsyncRabbitTemplate2.this.taskScheduler.schedule(new TimeoutTask(),
|
||||
new Date(System.currentTimeMillis() + AsyncRabbitTemplate2.this.receiveTimeout));
|
||||
}
|
||||
}
|
||||
else {
|
||||
this.timeoutTask = null;
|
||||
}
|
||||
}
|
||||
|
||||
private class TimeoutTask implements Runnable {
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
AsyncRabbitTemplate2.this.pending.remove(RabbitFuture2.this.correlationId);
|
||||
if (RabbitFuture2.this.channelHolder != null
|
||||
&& AsyncRabbitTemplate2.this.directReplyToContainer != null) {
|
||||
AsyncRabbitTemplate2.this.directReplyToContainer
|
||||
.releaseConsumerFor(RabbitFuture2.this.channelHolder, false, null); // NOSONAR
|
||||
}
|
||||
completeExceptionally(
|
||||
new AmqpReplyTimeoutException("Reply timed out", RabbitFuture2.this.requestMessage));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* A {@link RabbitFuture} with a return type of {@link Message}.
|
||||
* @since 1.6
|
||||
*/
|
||||
public class RabbitMessageFuture2 extends RabbitFuture2<Message> {
|
||||
|
||||
public RabbitMessageFuture2(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 RabbitConverterFuture2<C> extends RabbitFuture2<C> {
|
||||
|
||||
private volatile ParameterizedTypeReference<C> returnType;
|
||||
|
||||
public RabbitConverterFuture2(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() {
|
||||
}
|
||||
|
||||
@Override
|
||||
public Message postProcessMessage(Message message) throws AmqpException {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Override
|
||||
public Message postProcessMessage(Message message, Correlation correlation) throws AmqpException {
|
||||
Message messageToSend = message;
|
||||
AsyncCorrelationData<C> correlationData = (AsyncCorrelationData<C>) correlation;
|
||||
if (correlationData.userPostProcessor != null) {
|
||||
messageToSend = correlationData.userPostProcessor.postProcessMessage(message);
|
||||
}
|
||||
String correlationId = getOrSetCorrelationIdAndSetReplyTo(messageToSend, correlationData);
|
||||
correlationData.future = new RabbitConverterFuture2<C>(correlationId, message);
|
||||
if (correlationData.enableConfirms) {
|
||||
correlationData.setId(correlationId);
|
||||
correlationData.future.setConfirm(new CompletableFuture<>());
|
||||
}
|
||||
correlationData.future.setReturnType(correlationData.returnType);
|
||||
AsyncRabbitTemplate2.this.pending.put(correlationId, correlationData.future);
|
||||
return messageToSend;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static class AsyncCorrelationData<C> extends CorrelationData {
|
||||
|
||||
final MessagePostProcessor userPostProcessor; // NOSONAR
|
||||
|
||||
final ParameterizedTypeReference<C> returnType; // NOSONAR
|
||||
|
||||
final boolean enableConfirms; // NOSONAR
|
||||
|
||||
volatile RabbitConverterFuture2<C> future; // NOSONAR
|
||||
|
||||
AsyncCorrelationData(MessagePostProcessor userPostProcessor, ParameterizedTypeReference<C> returnType,
|
||||
boolean enableConfirms) {
|
||||
|
||||
this.userPostProcessor = userPostProcessor;
|
||||
this.returnType = returnType;
|
||||
this.enableConfirms = enableConfirms;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2021 the original author or authors.
|
||||
* Copyright 2002-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -17,6 +17,7 @@
|
||||
package org.springframework.amqp.rabbit.connection;
|
||||
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
import org.springframework.amqp.core.Correlation;
|
||||
import org.springframework.amqp.core.Message;
|
||||
@@ -43,6 +44,8 @@ public class CorrelationData implements Correlation {
|
||||
|
||||
private final SettableListenableFuture<Confirm> future = new SettableListenableFuture<>();
|
||||
|
||||
private final CompletableFuture<Confirm> completable = this.future.completable();
|
||||
|
||||
private volatile String id;
|
||||
|
||||
private volatile ReturnedMessage returnedMessage;
|
||||
@@ -90,11 +93,22 @@ public class CorrelationData implements Correlation {
|
||||
* Return a future to check the success/failure of the publish operation.
|
||||
* @return the future.
|
||||
* @since 2.1
|
||||
* @deprecated in favor of {@link #getCompletableFuture()}.
|
||||
*/
|
||||
@Deprecated
|
||||
public SettableListenableFuture<Confirm> getFuture() {
|
||||
return this.future;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return a future to check the success/failure of the publish operation.
|
||||
* @return the future.
|
||||
* @since 2.4.7
|
||||
*/
|
||||
public CompletableFuture<Confirm> getCompletableFuture() {
|
||||
return this.completable;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return a returned message, if any; requires a unique
|
||||
* {@link #CorrelationData(String) id}. Guaranteed to be populated before the future
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2021 the original author or authors.
|
||||
* Copyright 2002-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -935,6 +935,7 @@ public class PublisherCallbackChannelImpl
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("deprecation")
|
||||
private void doProcessAck(long seq, boolean ack, boolean multiple, boolean remove) {
|
||||
if (multiple) {
|
||||
processMultipleAck(seq, ack);
|
||||
@@ -971,6 +972,7 @@ public class PublisherCallbackChannelImpl
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("deprecation")
|
||||
private void processMultipleAck(long seq, boolean ack) {
|
||||
/*
|
||||
* Piggy-backed ack - extract all Listeners for this and earlier
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2014-2021 the original author or authors.
|
||||
* Copyright 2014-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -21,6 +21,7 @@ import java.lang.reflect.ParameterizedType;
|
||||
import java.lang.reflect.Type;
|
||||
import java.lang.reflect.WildcardType;
|
||||
import java.util.Arrays;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
@@ -386,6 +387,21 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe
|
||||
},
|
||||
t -> asyncFailure(request, channel, t));
|
||||
}
|
||||
else if (resultArg.getReturnValue() instanceof CompletableFuture) {
|
||||
if (!this.isManualAck) {
|
||||
this.logger.warn("Container AcknowledgeMode must be MANUAL for a Future<?> return type; "
|
||||
+ "otherwise the container will ack the message immediately");
|
||||
}
|
||||
((CompletableFuture<?>) resultArg.getReturnValue()).whenComplete((r, t) -> {
|
||||
if (t == null) {
|
||||
asyncSuccess(resultArg, request, channel, source, r);
|
||||
basicAck(request, channel);
|
||||
}
|
||||
else {
|
||||
asyncFailure(request, channel, t);
|
||||
}
|
||||
});
|
||||
}
|
||||
else if (monoPresent && MonoHandler.isMono(resultArg.getReturnValue())) {
|
||||
if (!this.isManualAck) {
|
||||
this.logger.warn("Container AcknowledgeMode must be MANUAL for a Mono<?> return type; "
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2015-2021 the original author or authors.
|
||||
* Copyright 2015-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -23,6 +23,7 @@ import java.util.Arrays;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
@@ -147,7 +148,8 @@ public class DelegatingInvocableHandler {
|
||||
|
||||
private boolean isAsyncReply(InvocableHandlerMethod method) {
|
||||
return (AbstractAdaptableMessageListener.monoPresent && MonoHandler.isMono(method.getMethod().getReturnType()))
|
||||
|| ListenableFuture.class.isAssignableFrom(method.getMethod().getReturnType());
|
||||
|| ListenableFuture.class.isAssignableFrom(method.getMethod().getReturnType())
|
||||
|| CompletableFuture.class.isAssignableFrom(method.getMethod().getReturnType());
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2015-2021 the original author or authors.
|
||||
* Copyright 2015-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.listener.adapter;
|
||||
|
||||
import java.lang.reflect.Method;
|
||||
import java.lang.reflect.Type;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
import org.springframework.lang.Nullable;
|
||||
import org.springframework.messaging.Message;
|
||||
@@ -50,7 +51,8 @@ public class HandlerAdapter {
|
||||
this.delegatingHandler = null;
|
||||
this.asyncReplies = (AbstractAdaptableMessageListener.monoPresent
|
||||
&& MonoHandler.isMono(invokerHandlerMethod.getMethod().getReturnType()))
|
||||
|| ListenableFuture.class.isAssignableFrom(invokerHandlerMethod.getMethod().getReturnType());
|
||||
|| ListenableFuture.class.isAssignableFrom(invokerHandlerMethod.getMethod().getReturnType())
|
||||
|| CompletableFuture.class.isAssignableFrom(invokerHandlerMethod.getMethod().getReturnType());
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -0,0 +1,536 @@
|
||||
/*
|
||||
* Copyright 2016-2020 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.amqp.rabbit;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.fail;
|
||||
import static org.awaitility.Awaitility.await;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CancellationException;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.BiConsumer;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.amqp.core.Address;
|
||||
import org.springframework.amqp.core.AmqpMessageReturnedException;
|
||||
import org.springframework.amqp.core.AmqpReplyTimeoutException;
|
||||
import org.springframework.amqp.core.AnonymousQueue;
|
||||
import org.springframework.amqp.core.Message;
|
||||
import org.springframework.amqp.core.MessageProperties;
|
||||
import org.springframework.amqp.core.Queue;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate2.RabbitConverterFuture2;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate2.RabbitMessageFuture2;
|
||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
|
||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
|
||||
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
|
||||
import org.springframework.amqp.rabbit.core.RabbitAdmin;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
|
||||
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
|
||||
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
|
||||
import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener;
|
||||
import org.springframework.amqp.support.converter.SimpleMessageConverter;
|
||||
import org.springframework.amqp.support.postprocessor.GUnzipPostProcessor;
|
||||
import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
|
||||
import org.springframework.amqp.utils.test.TestUtils;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 1.6
|
||||
*/
|
||||
@SpringJUnitConfig
|
||||
@DirtiesContext
|
||||
@RabbitAvailable
|
||||
public class AsyncRabbitTemplate2Tests {
|
||||
|
||||
@Autowired
|
||||
private AsyncRabbitTemplate2 asyncTemplate;
|
||||
|
||||
@Autowired
|
||||
private AsyncRabbitTemplate2 asyncDirectTemplate;
|
||||
|
||||
@Autowired
|
||||
private Queue requests;
|
||||
|
||||
@Autowired
|
||||
private AtomicReference<CountDownLatch> latch;
|
||||
|
||||
private final Message fooMessage = new SimpleMessageConverter().toMessage("foo", new MessageProperties());
|
||||
|
||||
@Test
|
||||
public void testConvert1Arg() throws Exception {
|
||||
final AtomicBoolean mppCalled = new AtomicBoolean();
|
||||
CompletableFuture<String> future = this.asyncTemplate.convertSendAndReceive("foo", m -> {
|
||||
mppCalled.set(true);
|
||||
return m;
|
||||
});
|
||||
checkConverterResult(future, "FOO");
|
||||
assertThat(mppCalled.get()).isTrue();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testConvert1ArgDirect() throws Exception {
|
||||
this.latch.set(new CountDownLatch(1));
|
||||
CompletableFuture<String> future1 = this.asyncDirectTemplate.convertSendAndReceive("foo");
|
||||
CompletableFuture<String> future2 = this.asyncDirectTemplate.convertSendAndReceive("bar");
|
||||
this.latch.get().countDown();
|
||||
checkConverterResult(future1, "FOO");
|
||||
checkConverterResult(future2, "BAR");
|
||||
this.latch.set(null);
|
||||
waitForZeroInUseConsumers();
|
||||
assertThat(TestUtils
|
||||
.getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount",
|
||||
Integer.class)).isEqualTo(2);
|
||||
final String missingQueue = UUID.randomUUID().toString();
|
||||
this.asyncDirectTemplate.convertSendAndReceive("", missingQueue, "foo"); // send to nowhere
|
||||
this.asyncDirectTemplate.stop(); // should clear the inUse channel map
|
||||
waitForZeroInUseConsumers();
|
||||
this.asyncDirectTemplate.start();
|
||||
this.asyncDirectTemplate.setReceiveTimeout(1);
|
||||
this.asyncDirectTemplate.convertSendAndReceive("", missingQueue, "foo"); // send to nowhere
|
||||
waitForZeroInUseConsumers();
|
||||
|
||||
this.asyncDirectTemplate.setReceiveTimeout(10000);
|
||||
this.asyncDirectTemplate.convertSendAndReceive("", missingQueue, "foo").cancel(true);
|
||||
waitForZeroInUseConsumers();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testConvert2Args() throws Exception {
|
||||
CompletableFuture<String> future = this.asyncTemplate.convertSendAndReceive(this.requests.getName(), "foo");
|
||||
checkConverterResult(future, "FOO");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testConvert3Args() throws Exception {
|
||||
CompletableFuture<String> future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo");
|
||||
checkConverterResult(future, "FOO");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testConvert4Args() throws Exception {
|
||||
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());
|
||||
});
|
||||
checkConverterResult(future, "FOOBAR");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessage1Arg() throws Exception {
|
||||
CompletableFuture<Message> future = this.asyncTemplate.sendAndReceive(getFooMessage());
|
||||
checkMessageResult(future, "FOO");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessage1ArgDirect() throws Exception {
|
||||
this.latch.set(new CountDownLatch(1));
|
||||
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);
|
||||
Message reply2 = checkMessageResult(future2, "FOO");
|
||||
assertThat(reply2.getMessageProperties().getConsumerQueue()).isEqualTo(Address.AMQ_RABBITMQ_REPLY_TO);
|
||||
this.latch.set(null);
|
||||
waitForZeroInUseConsumers();
|
||||
assertThat(TestUtils
|
||||
.getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount",
|
||||
Integer.class)).isEqualTo(2);
|
||||
this.asyncDirectTemplate.stop();
|
||||
this.asyncDirectTemplate.start();
|
||||
assertThat(TestUtils
|
||||
.getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount",
|
||||
Integer.class)).isEqualTo(0);
|
||||
}
|
||||
|
||||
private void waitForZeroInUseConsumers() throws InterruptedException {
|
||||
Map<?, ?> inUseConsumers = TestUtils
|
||||
.getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.inUseConsumerChannels", Map.class);
|
||||
await().until(() -> inUseConsumers.size() == 0);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessage2Args() throws Exception {
|
||||
CompletableFuture<Message> future = this.asyncTemplate.sendAndReceive(this.requests.getName(), getFooMessage());
|
||||
checkMessageResult(future, "FOO");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessage3Args() throws Exception {
|
||||
CompletableFuture<Message> future = this.asyncTemplate.sendAndReceive("", this.requests.getName(),
|
||||
getFooMessage());
|
||||
checkMessageResult(future, "FOO");
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
public void testCancel() {
|
||||
CompletableFuture<String> future = this.asyncTemplate.convertSendAndReceive("foo");
|
||||
future.cancel(false);
|
||||
assertThat(TestUtils.getPropertyValue(asyncTemplate, "pending", Map.class)).hasSize(0);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessageCustomCorrelation() throws Exception {
|
||||
Message message = getFooMessage();
|
||||
message.getMessageProperties().setCorrelationId("foo");
|
||||
CompletableFuture<Message> future = this.asyncTemplate.sendAndReceive(message);
|
||||
Message result = checkMessageResult(future, "FOO");
|
||||
assertThat(result.getMessageProperties().getCorrelationId()).isEqualTo("foo");
|
||||
}
|
||||
|
||||
private Message getFooMessage() {
|
||||
this.fooMessage.getMessageProperties().setCorrelationId(null);
|
||||
this.fooMessage.getMessageProperties().setReplyTo(null);
|
||||
return this.fooMessage;
|
||||
}
|
||||
|
||||
@Test
|
||||
@DirtiesContext
|
||||
public void testReturn() throws Exception {
|
||||
this.asyncTemplate.setMandatory(true);
|
||||
CompletableFuture<String> future = this.asyncTemplate.convertSendAndReceive(this.requests.getName() + "x",
|
||||
"foo");
|
||||
try {
|
||||
future.get(10, TimeUnit.SECONDS);
|
||||
fail("Expected exception");
|
||||
}
|
||||
catch (ExecutionException e) {
|
||||
assertThat(e.getCause()).isInstanceOf(AmqpMessageReturnedException.class);
|
||||
assertThat(((AmqpMessageReturnedException) e.getCause()).getRoutingKey()).isEqualTo(this.requests.getName() + "x");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@DirtiesContext
|
||||
public void testReturnDirect() throws Exception {
|
||||
this.asyncDirectTemplate.setMandatory(true);
|
||||
CompletableFuture<String> future = this.asyncDirectTemplate.convertSendAndReceive(this.requests.getName() + "x",
|
||||
"foo");
|
||||
try {
|
||||
future.get(10, TimeUnit.SECONDS);
|
||||
fail("Expected exception");
|
||||
}
|
||||
catch (ExecutionException e) {
|
||||
assertThat(e.getCause()).isInstanceOf(AmqpMessageReturnedException.class);
|
||||
assertThat(((AmqpMessageReturnedException) e.getCause()).getRoutingKey()).isEqualTo(this.requests.getName() + "x");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@DirtiesContext
|
||||
public void testConvertWithConfirm() throws Exception {
|
||||
this.asyncTemplate.setEnableConfirms(true);
|
||||
RabbitConverterFuture2<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
|
||||
CompletableFuture<Boolean> confirm = future.getConfirm();
|
||||
assertThat(confirm).isNotNull();
|
||||
assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue();
|
||||
checkConverterResult(future, "SLEEP");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DirtiesContext
|
||||
public void testMessageWithConfirm() throws Exception {
|
||||
this.asyncTemplate.setEnableConfirms(true);
|
||||
RabbitMessageFuture2 future = this.asyncTemplate
|
||||
.sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties()));
|
||||
CompletableFuture<Boolean> confirm = future.getConfirm();
|
||||
assertThat(confirm).isNotNull();
|
||||
assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue();
|
||||
checkMessageResult(future, "SLEEP");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DirtiesContext
|
||||
public void testConvertWithConfirmDirect() throws Exception {
|
||||
this.asyncDirectTemplate.setEnableConfirms(true);
|
||||
RabbitConverterFuture2<String> future = this.asyncDirectTemplate.convertSendAndReceive("sleep");
|
||||
CompletableFuture<Boolean> confirm = future.getConfirm();
|
||||
assertThat(confirm).isNotNull();
|
||||
assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue();
|
||||
checkConverterResult(future, "SLEEP");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DirtiesContext
|
||||
public void testMessageWithConfirmDirect() throws Exception {
|
||||
this.asyncDirectTemplate.setEnableConfirms(true);
|
||||
RabbitMessageFuture2 future = this.asyncDirectTemplate
|
||||
.sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties()));
|
||||
CompletableFuture<Boolean> confirm = future.getConfirm();
|
||||
assertThat(confirm).isNotNull();
|
||||
assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue();
|
||||
checkMessageResult(future, "SLEEP");
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
@DirtiesContext
|
||||
public void testReceiveTimeout() throws Exception {
|
||||
this.asyncTemplate.setReceiveTimeout(500);
|
||||
CompletableFuture<String> future = this.asyncTemplate.convertSendAndReceive("noReply");
|
||||
TheCallback callback = new TheCallback();
|
||||
future.whenComplete(callback);
|
||||
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1);
|
||||
try {
|
||||
future.get(10, TimeUnit.SECONDS);
|
||||
fail("Expected ExecutionException");
|
||||
}
|
||||
catch (ExecutionException e) {
|
||||
assertThat(e.getCause()).isInstanceOf(AmqpReplyTimeoutException.class);
|
||||
}
|
||||
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(0);
|
||||
assertThat(callback.latch.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
assertThat(callback.ex).isInstanceOf(AmqpReplyTimeoutException.class);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
@DirtiesContext
|
||||
public void testReplyAfterReceiveTimeout() throws Exception {
|
||||
this.asyncTemplate.setReceiveTimeout(100);
|
||||
RabbitConverterFuture2<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
|
||||
TheCallback callback = new TheCallback();
|
||||
future.whenComplete(callback);
|
||||
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1);
|
||||
try {
|
||||
future.get(10, TimeUnit.SECONDS);
|
||||
fail("Expected ExecutionException");
|
||||
}
|
||||
catch (ExecutionException e) {
|
||||
assertThat(e.getCause()).isInstanceOf(AmqpReplyTimeoutException.class);
|
||||
}
|
||||
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(0);
|
||||
assertThat(callback.latch.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
assertThat(callback.ex).isInstanceOf(AmqpReplyTimeoutException.class);
|
||||
|
||||
/*
|
||||
* Test there's no harm if the reply is received after the timeout. This
|
||||
* is unlikely to happen because the future is removed from the pending
|
||||
* map when it times out. However, there is a small race condition where
|
||||
* the reply arrives at the same time as the timeout.
|
||||
*/
|
||||
future.complete("foo");
|
||||
assertThat(callback.result).isNull();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
@DirtiesContext
|
||||
public void testStopCancelled() throws Exception {
|
||||
this.asyncTemplate.setReceiveTimeout(5000);
|
||||
RabbitConverterFuture2<String> future = this.asyncTemplate.convertSendAndReceive("noReply");
|
||||
TheCallback callback = new TheCallback();
|
||||
future.whenComplete(callback);
|
||||
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1);
|
||||
this.asyncTemplate.stop();
|
||||
// Second stop() to be sure that it is idempotent
|
||||
this.asyncTemplate.stop();
|
||||
try {
|
||||
future.get(10, TimeUnit.SECONDS);
|
||||
fail("Expected CancellationException");
|
||||
}
|
||||
catch (CancellationException e) {
|
||||
assertThat(future.getNackCause()).isEqualTo("AsyncRabbitTemplate was stopped while waiting for reply");
|
||||
}
|
||||
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(0);
|
||||
assertThat(callback.latch.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
assertThat(future.isCancelled()).isTrue();
|
||||
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "taskScheduler")).isNull();
|
||||
|
||||
/*
|
||||
* Test there's no harm if the reply is received after the cancel. This
|
||||
* should never happen because the container is stopped before canceling
|
||||
* and the future is removed from the pending map.
|
||||
*/
|
||||
future.complete("foo");
|
||||
assertThat(callback.result).isNull();
|
||||
}
|
||||
|
||||
private void checkConverterResult(CompletableFuture<String> future, String expected) throws InterruptedException {
|
||||
final CountDownLatch cdl = new CountDownLatch(1);
|
||||
final AtomicReference<String> resultRef = new AtomicReference<>();
|
||||
future.whenComplete((result, ex) -> {
|
||||
resultRef.set(result);
|
||||
cdl.countDown();
|
||||
});
|
||||
assertThat(cdl.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
assertThat(resultRef.get()).isEqualTo(expected);
|
||||
}
|
||||
|
||||
private Message checkMessageResult(CompletableFuture<Message> future, String expected) throws InterruptedException {
|
||||
final CountDownLatch cdl = new CountDownLatch(1);
|
||||
final AtomicReference<Message> resultRef = new AtomicReference<>();
|
||||
future.whenComplete((result, ex) -> {
|
||||
resultRef.set(result);
|
||||
cdl.countDown();
|
||||
});
|
||||
assertThat(cdl.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
assertThat(new String(resultRef.get().getBody())).isEqualTo(expected);
|
||||
return resultRef.get();
|
||||
}
|
||||
|
||||
public static class TheCallback implements BiConsumer<String, Throwable> {
|
||||
|
||||
private final CountDownLatch latch = new CountDownLatch(1);
|
||||
|
||||
private volatile String result;
|
||||
|
||||
private volatile Throwable ex;
|
||||
|
||||
|
||||
@Override
|
||||
public void accept(String result, Throwable ex) {
|
||||
this.result = result;
|
||||
this.ex = ex;
|
||||
latch.countDown();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Configuration
|
||||
public static class Config {
|
||||
|
||||
@Bean
|
||||
public AtomicReference<CountDownLatch> latch() {
|
||||
return new AtomicReference<>();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ConnectionFactory connectionFactory() {
|
||||
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
|
||||
connectionFactory.setPublisherConfirmType(ConfirmType.CORRELATED);
|
||||
connectionFactory.setPublisherReturns(true);
|
||||
return connectionFactory;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Queue requests() {
|
||||
return new AnonymousQueue();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Queue replies() {
|
||||
return new AnonymousQueue();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RabbitAdmin admin(ConnectionFactory connectionFactory) {
|
||||
return new RabbitAdmin(connectionFactory);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public GZipPostProcessor gZipPostProcessor() {
|
||||
GZipPostProcessor gZipPostProcessor = new GZipPostProcessor();
|
||||
gZipPostProcessor.setCopyProperties(true);
|
||||
return gZipPostProcessor;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RabbitTemplate template(ConnectionFactory connectionFactory) {
|
||||
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
|
||||
rabbitTemplate.setRoutingKey(requests().getName());
|
||||
rabbitTemplate.addBeforePublishPostProcessors(gZipPostProcessor());
|
||||
rabbitTemplate.addAfterReceivePostProcessors(new GUnzipPostProcessor());
|
||||
return rabbitTemplate;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RabbitTemplate templateForDirect(ConnectionFactory connectionFactory) {
|
||||
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
|
||||
rabbitTemplate.setRoutingKey(requests().getName());
|
||||
rabbitTemplate.addBeforePublishPostProcessors(gZipPostProcessor());
|
||||
rabbitTemplate.addAfterReceivePostProcessors(new GUnzipPostProcessor());
|
||||
return rabbitTemplate;
|
||||
}
|
||||
|
||||
@Bean
|
||||
@Primary
|
||||
public SimpleMessageListenerContainer replyContainer(ConnectionFactory connectionFactory) {
|
||||
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
|
||||
container.setAfterReceivePostProcessors(new GUnzipPostProcessor());
|
||||
container.setQueueNames(replies().getName());
|
||||
return container;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public AsyncRabbitTemplate2 asyncTemplate(RabbitTemplate template, SimpleMessageListenerContainer container) {
|
||||
return new AsyncRabbitTemplate2(template, container);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public AsyncRabbitTemplate2 asyncDirectTemplate(RabbitTemplate templateForDirect) {
|
||||
return new AsyncRabbitTemplate2(templateForDirect);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public SimpleMessageListenerContainer remoteContainer(ConnectionFactory connectionFactory) {
|
||||
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
|
||||
container.setQueueNames(requests().getName());
|
||||
container.setAfterReceivePostProcessors(new GUnzipPostProcessor());
|
||||
MessageListenerAdapter messageListener =
|
||||
new MessageListenerAdapter((ReplyingMessageListener<String, String>)
|
||||
message -> {
|
||||
CountDownLatch countDownLatch = latch().get();
|
||||
if (countDownLatch != null) {
|
||||
try {
|
||||
countDownLatch.await(10, TimeUnit.SECONDS);
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
if ("sleep".equals(message)) {
|
||||
try {
|
||||
Thread.sleep(500); // time for confirm to be delivered, or timeout to occur
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
else if ("noReply".equals(message)) {
|
||||
return null;
|
||||
}
|
||||
return message.toUpperCase();
|
||||
});
|
||||
|
||||
messageListener.setBeforeSendReplyPostProcessors(gZipPostProcessor());
|
||||
container.setMessageListener(messageListener);
|
||||
return container;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2016-2020 the original author or authors.
|
||||
* Copyright 2016-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -38,8 +38,6 @@ import org.springframework.amqp.core.AnonymousQueue;
|
||||
import org.springframework.amqp.core.Message;
|
||||
import org.springframework.amqp.core.MessageProperties;
|
||||
import org.springframework.amqp.core.Queue;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture;
|
||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
|
||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
|
||||
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
|
||||
@@ -71,6 +69,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
|
||||
@SpringJUnitConfig
|
||||
@DirtiesContext
|
||||
@RabbitAvailable
|
||||
@SuppressWarnings("deprecation")
|
||||
public class AsyncRabbitTemplateTests {
|
||||
|
||||
@Autowired
|
||||
@@ -253,7 +252,7 @@ public class AsyncRabbitTemplateTests {
|
||||
@DirtiesContext
|
||||
public void testConvertWithConfirm() throws Exception {
|
||||
this.asyncTemplate.setEnableConfirms(true);
|
||||
RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
|
||||
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
|
||||
ListenableFuture<Boolean> confirm = future.getConfirm();
|
||||
assertThat(confirm).isNotNull();
|
||||
assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue();
|
||||
@@ -264,7 +263,7 @@ public class AsyncRabbitTemplateTests {
|
||||
@DirtiesContext
|
||||
public void testMessageWithConfirm() throws Exception {
|
||||
this.asyncTemplate.setEnableConfirms(true);
|
||||
RabbitMessageFuture future = this.asyncTemplate
|
||||
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture future = this.asyncTemplate
|
||||
.sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties()));
|
||||
ListenableFuture<Boolean> confirm = future.getConfirm();
|
||||
assertThat(confirm).isNotNull();
|
||||
@@ -276,7 +275,7 @@ public class AsyncRabbitTemplateTests {
|
||||
@DirtiesContext
|
||||
public void testConvertWithConfirmDirect() throws Exception {
|
||||
this.asyncDirectTemplate.setEnableConfirms(true);
|
||||
RabbitConverterFuture<String> future = this.asyncDirectTemplate.convertSendAndReceive("sleep");
|
||||
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture<String> future = this.asyncDirectTemplate.convertSendAndReceive("sleep");
|
||||
ListenableFuture<Boolean> confirm = future.getConfirm();
|
||||
assertThat(confirm).isNotNull();
|
||||
assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue();
|
||||
@@ -287,7 +286,7 @@ public class AsyncRabbitTemplateTests {
|
||||
@DirtiesContext
|
||||
public void testMessageWithConfirmDirect() throws Exception {
|
||||
this.asyncDirectTemplate.setEnableConfirms(true);
|
||||
RabbitMessageFuture future = this.asyncDirectTemplate
|
||||
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture future = this.asyncDirectTemplate
|
||||
.sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties()));
|
||||
ListenableFuture<Boolean> confirm = future.getConfirm();
|
||||
assertThat(confirm).isNotNull();
|
||||
@@ -321,7 +320,7 @@ public class AsyncRabbitTemplateTests {
|
||||
@DirtiesContext
|
||||
public void testReplyAfterReceiveTimeout() throws Exception {
|
||||
this.asyncTemplate.setReceiveTimeout(100);
|
||||
RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
|
||||
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
|
||||
TheCallback callback = new TheCallback();
|
||||
future.addCallback(callback);
|
||||
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1);
|
||||
@@ -351,7 +350,7 @@ public class AsyncRabbitTemplateTests {
|
||||
@DirtiesContext
|
||||
public void testStopCancelled() throws Exception {
|
||||
this.asyncTemplate.setReceiveTimeout(5000);
|
||||
RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("noReply");
|
||||
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("noReply");
|
||||
TheCallback callback = new TheCallback();
|
||||
future.addCallback(callback);
|
||||
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1);
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2018-2019 the original author or authors.
|
||||
* Copyright 2018-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -24,6 +24,7 @@ import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Properties;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
@@ -35,8 +36,8 @@ import org.springframework.amqp.ImmediateRequeueAmqpException;
|
||||
import org.springframework.amqp.core.AcknowledgeMode;
|
||||
import org.springframework.amqp.core.AnonymousQueue;
|
||||
import org.springframework.amqp.core.Queue;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate2;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate2.RabbitConverterFuture2;
|
||||
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
|
||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
|
||||
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
|
||||
@@ -53,8 +54,6 @@ import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
|
||||
import org.springframework.util.concurrent.ListenableFuture;
|
||||
import org.springframework.util.concurrent.SettableListenableFuture;
|
||||
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
@@ -75,7 +74,7 @@ public class AsyncListenerTests {
|
||||
private RabbitTemplate rabbitTemplate;
|
||||
|
||||
@Autowired
|
||||
private AsyncRabbitTemplate asyncTemplate;
|
||||
private AsyncRabbitTemplate2 asyncTemplate;
|
||||
|
||||
@Autowired
|
||||
private Queue queue1;
|
||||
@@ -107,7 +106,7 @@ public class AsyncListenerTests {
|
||||
@Test
|
||||
public void testAsyncListener() throws Exception {
|
||||
assertThat(this.rabbitTemplate.convertSendAndReceive(this.queue1.getName(), "foo")).isEqualTo("FOO");
|
||||
RabbitConverterFuture<Object> future = this.asyncTemplate.convertSendAndReceive(this.queue1.getName(), "foo");
|
||||
RabbitConverterFuture2<Object> future = this.asyncTemplate.convertSendAndReceive(this.queue1.getName(), "foo");
|
||||
assertThat(future.get(10, TimeUnit.SECONDS)).isEqualTo("FOO");
|
||||
assertThat(this.config.typeId).isEqualTo("java.lang.String");
|
||||
assertThat(this.rabbitTemplate.convertSendAndReceive(this.queue2.getName(), "foo")).isEqualTo("FOO");
|
||||
@@ -194,8 +193,8 @@ public class AsyncListenerTests {
|
||||
}
|
||||
|
||||
@Bean
|
||||
public AsyncRabbitTemplate asyncTemplate() {
|
||||
return new AsyncRabbitTemplate(rabbitTemplate());
|
||||
public AsyncRabbitTemplate2 asyncTemplate() {
|
||||
return new AsyncRabbitTemplate2(rabbitTemplate());
|
||||
}
|
||||
|
||||
@Bean
|
||||
@@ -284,13 +283,13 @@ public class AsyncListenerTests {
|
||||
private final AtomicBoolean first7 = new AtomicBoolean(true);
|
||||
|
||||
@RabbitListener(id = "foo", queues = "#{queue1.name}")
|
||||
public ListenableFuture<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;
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2016-2020 the original author or authors.
|
||||
* Copyright 2016-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -23,8 +23,8 @@ import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate2;
|
||||
import org.springframework.amqp.rabbit.AsyncRabbitTemplate2.RabbitConverterFuture2;
|
||||
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
|
||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
|
||||
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
|
||||
@@ -63,7 +63,7 @@ public class ComplexTypeJsonIntegrationTests {
|
||||
private RabbitTemplate rabbitTemplate;
|
||||
|
||||
@Autowired
|
||||
private AsyncRabbitTemplate asyncTemplate;
|
||||
private AsyncRabbitTemplate2 asyncTemplate;
|
||||
|
||||
private static Foo<Bar<Baz, Qux>> makeAFoo() {
|
||||
Foo<Bar<Baz, Qux>> foo = new Foo<>();
|
||||
@@ -137,7 +137,7 @@ public class ComplexTypeJsonIntegrationTests {
|
||||
new ParameterizedTypeReference<Foo<Bar<Baz, Qux>>>() { }));
|
||||
}
|
||||
|
||||
private void verifyFooBarBazQux(RabbitConverterFuture<Foo<Bar<Baz, Qux>>> future) throws Exception {
|
||||
private void verifyFooBarBazQux(RabbitConverterFuture2<Foo<Bar<Baz, Qux>>> future) throws Exception {
|
||||
verifyFooBarBazQux(future.get(10, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
@@ -169,8 +169,8 @@ public class ComplexTypeJsonIntegrationTests {
|
||||
}
|
||||
|
||||
@Bean
|
||||
public AsyncRabbitTemplate asyncTemplate() {
|
||||
return new AsyncRabbitTemplate(template());
|
||||
public AsyncRabbitTemplate2 asyncTemplate() {
|
||||
return new AsyncRabbitTemplate2(template());
|
||||
}
|
||||
|
||||
@Bean
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2020-2021 the original author or authors.
|
||||
* Copyright 2020-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -49,7 +49,7 @@ public class MessagingTemplateConfirmsTests {
|
||||
CorrelationData data = new CorrelationData();
|
||||
rmt.send("messaging.confirms",
|
||||
new GenericMessage<>("foo", Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, data)));
|
||||
assertThat(data.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
assertThat(data.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
ccf.destroy();
|
||||
}
|
||||
|
||||
@@ -65,7 +65,7 @@ public class MessagingTemplateConfirmsTests {
|
||||
CorrelationData data = new CorrelationData("foo");
|
||||
rmt.send("messaging.confirms.unroutable",
|
||||
new GenericMessage<>("foo", Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, data)));
|
||||
assertThat(data.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
assertThat(data.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
assertThat(data.getReturned()).isNotNull();
|
||||
ccf.destroy();
|
||||
}
|
||||
|
||||
@@ -839,14 +839,14 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
|
||||
admin.declareQueue(queue);
|
||||
CorrelationData cd1 = new CorrelationData();
|
||||
this.templateWithConfirmsEnabled.convertAndSend("", queue.getName(), "foo", cd1);
|
||||
assertThat(cd1.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
assertThat(cd1.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
CorrelationData cd2 = new CorrelationData();
|
||||
this.templateWithConfirmsEnabled.convertAndSend("", queue.getName(), "bar", cd2);
|
||||
assertThat(cd2.getFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse();
|
||||
assertThat(cd2.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse();
|
||||
CorrelationData cd3 = new CorrelationData();
|
||||
this.templateWithConfirmsEnabled.convertAndSend("NO_EXCHANGE_HERE", queue.getName(), "foo", cd3);
|
||||
assertThat(cd3.getFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse();
|
||||
assertThat(cd3.getFuture().get().getReason()).contains("NOT_FOUND");
|
||||
assertThat(cd3.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse();
|
||||
assertThat(cd3.getCompletableFuture().get().getReason()).contains("NOT_FOUND");
|
||||
CorrelationData cd4 = new CorrelationData("42");
|
||||
AtomicBoolean resent = new AtomicBoolean();
|
||||
AtomicReference<String> callbackThreadName = new AtomicReference<>();
|
||||
@@ -858,7 +858,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
|
||||
callbackLatch.countDown();
|
||||
});
|
||||
this.templateWithConfirmsAndReturnsEnabled.convertAndSend("", "NO_QUEUE_HERE", "foo", cd4);
|
||||
assertThat(cd4.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
assertThat(cd4.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
assertThat(callbackLatch.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
assertThat(cd4.getReturned()).isNotNull();
|
||||
assertThat(resent.get()).isTrue();
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2016-2021 the original author or authors.
|
||||
* Copyright 2016-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -118,13 +118,13 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests2 {
|
||||
this.templateWithConfirmsEnabled.setMandatory(true);
|
||||
CorrelationData corr = new CorrelationData();
|
||||
this.templateWithConfirmsEnabled.convertAndSend("", ROUTE2, "foo", corr);
|
||||
assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
assertThat(corr.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
if (listener) {
|
||||
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
}
|
||||
corr = new CorrelationData();
|
||||
this.templateWithConfirmsEnabled.convertAndSend("", "bad route", "foo", corr);
|
||||
assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
assertThat(corr.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
|
||||
assertThat(corr.getReturned()).isNotNull();
|
||||
}
|
||||
|
||||
|
||||
@@ -109,7 +109,7 @@ class RabbitTemplateRoutingConnectionFactoryIntegrationTests {
|
||||
rabbitTemplate.send(ROUTE, message, correlationData);
|
||||
assertThat(rabbitTemplate.getUnconfirmedCount()).isEqualTo(1);
|
||||
|
||||
final CorrelationData.Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS);
|
||||
final CorrelationData.Confirm confirm = correlationData.getCompletableFuture().get(10, TimeUnit.SECONDS);
|
||||
|
||||
assertThat(confirm.isAck()).isTrue();
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2021 the original author or authors.
|
||||
* Copyright 2021-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.listener;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
@@ -97,7 +98,7 @@ public class AsyncReplyToTests {
|
||||
.build());
|
||||
assertThat(config.dmlcLatch.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
registry.getListenerContainer("dmlc").stop();
|
||||
assertThat(admin.getQueueInfo("async2").getMessageCount()).isEqualTo(1);
|
||||
assertThat(admin.getQueueInfo("async2").getMessageCount()).isEqualTo(0);
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@@ -114,8 +115,10 @@ public class AsyncReplyToTests {
|
||||
}
|
||||
|
||||
@RabbitListener(id = "dmlc", queues = "async2", containerFactory = "dmlcf")
|
||||
ListenableFuture<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
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -3570,6 +3570,7 @@ IMPORTANT: Containers created this way are normal `@Bean` instances and are not
|
||||
===== Asynchronous `@RabbitListener` Return Types
|
||||
|
||||
Starting with version 2.1, `@RabbitListener` (and `@RabbitHandler`) methods can be specified with asynchronous return types `ListenableFuture<?>` and `Mono<?>`, letting the reply be sent asynchronously.
|
||||
`CompletableFuture<?>` was added in 2.4.7 and `ListenableFuture<?>` will be removed in 3.0.
|
||||
|
||||
IMPORTANT: The listener container factory must be configured with `AcknowledgeMode.MANUAL` so that the consumer thread will not ack the message; instead, the asynchronous completion will ack or nack the message when the async operation completes.
|
||||
When the async result is completed with an error, whether the message is requeued or not depends on the exception type thrown, the container configuration, and the container error handler.
|
||||
@@ -4647,6 +4648,9 @@ Version 2.0 introduced variants of these methods (`convertSendAndReceiveAsType`)
|
||||
You must configure the underlying `RabbitTemplate` with a `SmartMessageConverter`.
|
||||
See <<json-complex>> for more information.
|
||||
|
||||
Starting with version 2.4.7, the `AsyncRabbitTemplate` is deprecated in favor of `AsyncRabbitTemplate2` which returns `CompletableFuture` s instead of `ListenableFuture` s.
|
||||
In 3.0, `AsyncRabbitTemplate2` will be renamed to `AsyncRabbitTemplate`.
|
||||
|
||||
[[remoting]]
|
||||
===== Spring Remoting with AMQP
|
||||
|
||||
|
||||
@@ -11,6 +11,9 @@ See <<change-history>> for changes in previous versions.
|
||||
`MessageProperties` is now available for argument matching.
|
||||
See <<async-annotation-driven-enable-signature>> for more information.
|
||||
|
||||
Async reply types now include `CompleteableFuture`
|
||||
See <<async-returns>> for more information.
|
||||
|
||||
==== `RabbitAdmin` Changes
|
||||
|
||||
A new property `recoverManualDeclarations` allows recovery of manually declared queues/exchanges/bindings.
|
||||
@@ -25,3 +28,8 @@ See <<remoting>> for more information.
|
||||
|
||||
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.
|
||||
|
||||
Reference in New Issue
Block a user