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

GH-1473: Revert CompletableFuture Changes

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

- caused problems with Boot auto configuration
- since 3.0 is a major release, switching to CF there is not onerous
- retain `CompletableFuture` as an async return type
This commit is contained in:
Gary Russell
2022-09-13 10:17:23 -04:00
committed by Artem Bilan
parent e8f12b2156
commit 2746ebeaca
26 changed files with 49 additions and 2199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

205
spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate2.java
View File

@@ -1,205 +0,0 @@
/*
* 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);
}

97
spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamOperations2.java
View File

@@ -1,97 +0,0 @@
/*
* 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.rabbit.stream.producer;
import java.util.concurrent.CompletableFuture;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.lang.Nullable;
import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
import com.rabbitmq.stream.MessageBuilder;
/**
* Provides methods for sending messages using a RabbitMQ Stream producer,
* returning {@link CompletableFuture}.
*
* @author Gary Russell
* @since 2.4.7
*
*/
public interface RabbitStreamOperations2 extends AutoCloseable {
/**
* Send a Spring AMQP message.
* @param message the message.
* @return a future to indicate success/failure.
*/
CompletableFuture<Boolean> send(Message message);
/**
* Convert to and send a Spring AMQP message.
* @param message the payload.
* @return a future to indicate success/failure.
*/
CompletableFuture<Boolean> convertAndSend(Object message);
/**
* Convert to and send a Spring AMQP message. If a {@link MessagePostProcessor} is
* provided and returns {@code null}, the message is not sent and the future is
* completed with {@code false}.
* @param message the payload.
* @param mpp a message post processor.
* @return a future to indicate success/failure.
*/
CompletableFuture<Boolean> convertAndSend(Object message, @Nullable MessagePostProcessor mpp);
/**
* Send a native stream message.
* @param message the message.
* @return a future to indicate success/failure.
* @see #messageBuilder()
*/
CompletableFuture<Boolean> send(com.rabbitmq.stream.Message message);
/**
* Return the producer's {@link MessageBuilder} to create native stream messages.
* @return the builder.
* @see #send(com.rabbitmq.stream.Message)
*/
MessageBuilder messageBuilder();
/**
* Return the message converter.
* @return the converter.
*/
MessageConverter messageConverter();
/**
* Return the stream message converter.
* @return the converter;
*/
StreamMessageConverter streamMessageConverter();
@Override
default void close() throws AmqpException {
// narrow exception to avoid compiler warning - see
// https://bugs.openjdk.java.net/browse/JDK-8155591
}
}

2
spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java
View File

@@ -42,9 +42,7 @@ import com.rabbitmq.stream.ProducerBuilder;
*
* @author Gary Russell
* @since 2.4
* @deprecated in favor of {@link RabbitStreamTemplate2}.
*/
@Deprecated
public class RabbitStreamTemplate implements RabbitStreamOperations, BeanNameAware {
protected final LogAccessor logger = new LogAccessor(getClass()); // NOSONAR

225
spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate2.java
View File

@@ -1,225 +0,0 @@
/*
* Copyright 2021 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.rabbit.stream.producer;
import java.util.concurrent.CompletableFuture;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.core.log.LogAccessor;
import org.springframework.lang.Nullable;
import org.springframework.rabbit.stream.support.StreamMessageProperties;
import org.springframework.rabbit.stream.support.converter.DefaultStreamMessageConverter;
import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
import org.springframework.util.Assert;
import com.rabbitmq.stream.ConfirmationHandler;
import com.rabbitmq.stream.Constants;
import com.rabbitmq.stream.Environment;
import com.rabbitmq.stream.MessageBuilder;
import com.rabbitmq.stream.Producer;
import com.rabbitmq.stream.ProducerBuilder;
/**
* Default implementation of {@link RabbitStreamOperations}.
*
* @author Gary Russell
* @since 2.4.7
*
*/
public class RabbitStreamTemplate2 implements RabbitStreamOperations2, BeanNameAware {
protected final LogAccessor logger = new LogAccessor(getClass()); // NOSONAR
private final Environment environment;
private final String streamName;
private MessageConverter messageConverter = new SimpleMessageConverter();
private StreamMessageConverter streamConverter = new DefaultStreamMessageConverter();
private boolean streamConverterSet;
private Producer producer;
private String beanName;
private ProducerCustomizer producerCustomizer = (name, builder) -> { };
/**
* Construct an instance with the provided {@link Environment}.
* @param environment the environment.
* @param streamName the stream name.
*/
public RabbitStreamTemplate2(Environment environment, String streamName) {
Assert.notNull(environment, "'environment' cannot be null");
Assert.notNull(streamName, "'streamName' cannot be null");
this.environment = environment;
this.streamName = streamName;
}
private synchronized Producer createOrGetProducer() {
if (this.producer == null) {
ProducerBuilder builder = this.environment.producerBuilder();
builder.stream(this.streamName);
this.producerCustomizer.accept(this.beanName, builder);
this.producer = builder.build();
if (!this.streamConverterSet) {
((DefaultStreamMessageConverter) this.streamConverter).setBuilderSupplier(
() -> this.producer.messageBuilder());
}
}
return this.producer;
}
@Override
public synchronized void setBeanName(String name) {
this.beanName = name;
}
/**
* Set a converter for {@link #convertAndSend(Object)} operations.
* @param messageConverter the converter.
*/
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "'messageConverter' cannot be null");
this.messageConverter = messageConverter;
}
/**
* Set a converter to convert from {@link Message} to {@link com.rabbitmq.stream.Message}
* for {@link #send(Message)} and {@link #convertAndSend(Object)} methods.
* @param streamConverter the converter.
*/
public synchronized void setStreamConverter(StreamMessageConverter streamConverter) {
Assert.notNull(streamConverter, "'streamConverter' cannot be null");
this.streamConverter = streamConverter;
this.streamConverterSet = true;
}
/**
* Used to customize the {@link ProducerBuilder} before the {@link Producer} is built.
* @param producerCustomizer the customizer;
*/
public synchronized void setProducerCustomizer(ProducerCustomizer producerCustomizer) {
Assert.notNull(producerCustomizer, "'producerCustomizer' cannot be null");
this.producerCustomizer = producerCustomizer;
}
@Override
public MessageConverter messageConverter() {
return this.messageConverter;
}
@Override
public StreamMessageConverter streamMessageConverter() {
return this.streamConverter;
}
@Override
public CompletableFuture<Boolean> send(Message message) {
CompletableFuture<Boolean> future = new CompletableFuture<>();
createOrGetProducer().send(this.streamConverter.fromMessage(message), handleConfirm(future));
return future;
}
@Override
public CompletableFuture<Boolean> convertAndSend(Object message) {
return convertAndSend(message, null);
}
@Override
public CompletableFuture<Boolean> convertAndSend(Object message, @Nullable MessagePostProcessor mpp) {
Message message2 = this.messageConverter.toMessage(message, new StreamMessageProperties());
Assert.notNull(message2, "The message converter returned null");
if (mpp != null) {
message2 = mpp.postProcessMessage(message2);
if (message2 == null) {
this.logger.debug("Message Post Processor returned null, message not sent");
CompletableFuture<Boolean> future = new CompletableFuture<>();
future.complete(false);
return future;
}
}
return send(message2);
}
@Override
public CompletableFuture<Boolean> send(com.rabbitmq.stream.Message message) {
CompletableFuture<Boolean> future = new CompletableFuture<>();
createOrGetProducer().send(message, handleConfirm(future));
return future;
}
@Override
public MessageBuilder messageBuilder() {
return createOrGetProducer().messageBuilder();
}
private ConfirmationHandler handleConfirm(CompletableFuture<Boolean> future) {
return confStatus -> {
if (confStatus.isConfirmed()) {
future.complete(true);
}
else {
int code = confStatus.getCode();
String errorMessage;
switch (code) {
case Constants.CODE_MESSAGE_ENQUEUEING_FAILED:
errorMessage = "Message Enqueueing Failed";
break;
case Constants.CODE_PRODUCER_CLOSED:
errorMessage = "Producer Closed";
break;
case Constants.CODE_PRODUCER_NOT_AVAILABLE:
errorMessage = "Producer Not Available";
break;
case Constants.CODE_PUBLISH_CONFIRM_TIMEOUT:
errorMessage = "Publish Confirm Timeout";
break;
default:
errorMessage = "Unknown code: " + code;
break;
}
future.completeExceptionally(new StreamSendException(errorMessage, code));
}
};
}
/**
* {@inheritDoc}
* <p>
* <b>Close the underlying producer; a new producer will be created on the next
* operation that requires one.</b>
*/
@Override
public synchronized void close() {
if (this.producer != null) {
this.producer.close();
this.producer = null;
}
}
}

8
spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java
View File

@@ -42,7 +42,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.rabbit.stream.config.StreamRabbitListenerContainerFactory;
import org.springframework.rabbit.stream.producer.RabbitStreamTemplate2;
import org.springframework.rabbit.stream.producer.RabbitStreamTemplate;
import org.springframework.rabbit.stream.retry.StreamRetryOperationsInterceptorFactoryBean;
import org.springframework.rabbit.stream.support.StreamMessageProperties;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
@@ -70,7 +70,7 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
Config config;
@Test
void simple(@Autowired RabbitStreamTemplate2 template) throws Exception {
void simple(@Autowired RabbitStreamTemplate template) throws Exception {
Future<Boolean> future = template.convertAndSend("foo");
assertThat(future.get(10, TimeUnit.SECONDS)).isTrue();
future = template.convertAndSend("bar", msg -> msg);
@@ -247,8 +247,8 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
}
@Bean
RabbitStreamTemplate2 streamTemplate1(Environment env) {
RabbitStreamTemplate2 template = new RabbitStreamTemplate2(env, "test.stream.queue1");
RabbitStreamTemplate streamTemplate1(Environment env) {
RabbitStreamTemplate template = new RabbitStreamTemplate(env, "test.stream.queue1");
template.setProducerCustomizer((name, builder) -> builder.name("test"));
return template;
}

16
spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplateTests.java
View File

@@ -23,7 +23,6 @@ import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.mock;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
@@ -31,6 +30,7 @@ import org.junit.jupiter.api.Test;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
import org.springframework.util.concurrent.ListenableFuture;
import com.rabbitmq.stream.ConfirmationHandler;
import com.rabbitmq.stream.ConfirmationStatus;
@@ -81,7 +81,7 @@ public class RabbitStreamTemplateTests {
handler.handle(status);
return null;
}).given(producer).send(any(), any());
try (RabbitStreamTemplate2 template = new RabbitStreamTemplate2(env, "foo")) {
try (RabbitStreamTemplate template = new RabbitStreamTemplate(env, "foo")) {
SimpleMessageConverter messageConverter = new SimpleMessageConverter();
template.setMessageConverter(messageConverter);
assertThat(template.messageConverter()).isSameAs(messageConverter);
@@ -89,25 +89,25 @@ public class RabbitStreamTemplateTests {
given(converter.fromMessage(any())).willReturn(mock(Message.class));
template.setStreamConverter(converter);
assertThat(template.streamMessageConverter()).isSameAs(converter);
CompletableFuture<Boolean> future = template.convertAndSend("foo");
ListenableFuture<Boolean> future = template.convertAndSend("foo");
assertThat(future.get()).isTrue();
CompletableFuture<Boolean> future1 = template.convertAndSend("foo");
ListenableFuture<Boolean> future1 = template.convertAndSend("foo");
assertThatExceptionOfType(ExecutionException.class).isThrownBy(() -> future1.get())
.withCauseExactlyInstanceOf(StreamSendException.class)
.withStackTraceContaining("Message Enqueueing Failed");
CompletableFuture<Boolean> future2 = template.convertAndSend("foo");
ListenableFuture<Boolean> future2 = template.convertAndSend("foo");
assertThatExceptionOfType(ExecutionException.class).isThrownBy(() -> future2.get())
.withCauseExactlyInstanceOf(StreamSendException.class)
.withStackTraceContaining("Producer Closed");
CompletableFuture<Boolean> future3 = template.convertAndSend("foo");
ListenableFuture<Boolean> future3 = template.convertAndSend("foo");
assertThatExceptionOfType(ExecutionException.class).isThrownBy(() -> future3.get())
.withCauseExactlyInstanceOf(StreamSendException.class)
.withStackTraceContaining("Producer Not Available");
CompletableFuture<Boolean> future4 = template.convertAndSend("foo");
ListenableFuture<Boolean> future4 = template.convertAndSend("foo");
assertThatExceptionOfType(ExecutionException.class).isThrownBy(() -> future4.get())
.withCauseExactlyInstanceOf(StreamSendException.class)
.withStackTraceContaining("Publish Confirm Timeout");
CompletableFuture<Boolean> future5 = template.convertAndSend("foo");
ListenableFuture<Boolean> future5 = template.convertAndSend("foo");
assertThatExceptionOfType(ExecutionException.class).isThrownBy(() -> future5.get())
.withCauseExactlyInstanceOf(StreamSendException.class)
.withStackTraceContaining("Unknown code: " + -1);

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

@@ -89,9 +89,7 @@ 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 {

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

@@ -1,744 +0,0 @@
/*
* 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.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.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, RabbitFuture<?>> 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 RabbitMessageFuture sendAndReceive(Message message) {
return sendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), message);
}
@Override
public RabbitMessageFuture sendAndReceive(String routingKey, Message message) {
return sendAndReceive(this.template.getExchange(), routingKey, message);
}
@Override
public RabbitMessageFuture sendAndReceive(String exchange, String routingKey, Message message) {
String correlationId = getOrSetCorrelationIdAndSetReplyTo(message, null);
RabbitMessageFuture future = new RabbitMessageFuture(correlationId, message, this::canceler,
this::timeoutTask);
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> RabbitConverterFuture<C> convertSendAndReceive(Object object) {
return convertSendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), object, null);
}
@Override
public <C> RabbitConverterFuture<C> convertSendAndReceive(String routingKey, Object object) {
return convertSendAndReceive(this.template.getExchange(), routingKey, object, null);
}
@Override
public <C> RabbitConverterFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object) {
return convertSendAndReceive(exchange, routingKey, object, null);
}
@Override
public <C> RabbitConverterFuture<C> convertSendAndReceive(Object object,
MessagePostProcessor messagePostProcessor) {
return convertSendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), object,
messagePostProcessor);
}
@Override
public <C> RabbitConverterFuture<C> convertSendAndReceive(String routingKey, Object object,
MessagePostProcessor messagePostProcessor) {
return convertSendAndReceive(this.template.getExchange(), routingKey, object, messagePostProcessor);
}
@Override
public <C> RabbitConverterFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object,
MessagePostProcessor messagePostProcessor) {
return convertSendAndReceive(exchange, routingKey, object, messagePostProcessor, null);
}
@Override
public <C> RabbitConverterFuture<C> convertSendAndReceiveAsType(Object object,
ParameterizedTypeReference<C> responseType) {
return convertSendAndReceiveAsType(this.template.getExchange(), this.template.getRoutingKey(), object,
null, responseType);
}
@Override
public <C> RabbitConverterFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
ParameterizedTypeReference<C> responseType) {
return convertSendAndReceiveAsType(this.template.getExchange(), routingKey, object, null, responseType);
}
@Override
public <C> RabbitConverterFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object,
ParameterizedTypeReference<C> responseType) {
return convertSendAndReceiveAsType(exchange, routingKey, object, null, responseType);
}
@Override
public <C> RabbitConverterFuture<C> convertSendAndReceiveAsType(Object object,
MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType) {
return convertSendAndReceiveAsType(this.template.getExchange(), this.template.getRoutingKey(), object,
messagePostProcessor, responseType);
}
@Override
public <C> RabbitConverterFuture<C> convertSendAndReceiveAsType(String routingKey, Object object,
MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType) {
return convertSendAndReceiveAsType(this.template.getExchange(), routingKey, object, messagePostProcessor,
responseType);
}
@Override
public <C> RabbitConverterFuture<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> RabbitConverterFuture<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);
}
RabbitConverterFuture<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 (RabbitFuture<?> 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);
}
RabbitFuture<?> future = this.pending.remove(correlationId);
if (future != null) {
if (future instanceof RabbitConverterFuture) {
MessageConverter messageConverter = this.template.getMessageConverter();
RabbitConverterFuture<Object> rabbitFuture = (RabbitConverterFuture<Object>) future;
Object converted = rabbitFuture.getReturnType() != null
&& messageConverter instanceof SmartMessageConverter
? ((SmartMessageConverter) messageConverter).fromMessage(message,
rabbitFuture.getReturnType())
: messageConverter.fromMessage(message);
rabbitFuture.complete(converted);
}
else {
((RabbitMessageFuture) 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)) {
RabbitFuture<?> 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) {
RabbitFuture<?> 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;
}
private void canceler(String correlationId, @Nullable ChannelHolder channelHolder) {
this.pending.remove(correlationId);
if (channelHolder != null && this.directReplyToContainer != null) {
this.directReplyToContainer
.releaseConsumerFor(channelHolder, false, null); // NOSONAR
}
}
@Nullable
private ScheduledFuture<?> timeoutTask(RabbitFuture<?> future) {
if (this.receiveTimeout > 0) {
synchronized (this) {
if (!this.running) {
this.pending.remove(future.getCorrelationId());
throw new IllegalStateException("'AsyncRabbitTemplate' must be started.");
}
return this.taskScheduler.schedule(
new TimeoutTask(future, this.pending, this.directReplyToContainer),
new Date(System.currentTimeMillis() + this.receiveTimeout));
}
}
return null;
}
@Override
public String toString() {
return this.beanName == null ? super.toString() : (this.getClass().getSimpleName() + ": " + this.beanName);
}
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 RabbitConverterFuture<C>(correlationId, message,
AsyncRabbitTemplate2.this::canceler, AsyncRabbitTemplate2.this::timeoutTask);
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 RabbitConverterFuture<C> future; // NOSONAR
AsyncCorrelationData(MessagePostProcessor userPostProcessor, ParameterizedTypeReference<C> returnType,
boolean enableConfirms) {
this.userPostProcessor = userPostProcessor;
this.returnType = returnType;
this.enableConfirms = enableConfirms;
}
}
}

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

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

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

@@ -1,116 +0,0 @@
/*
* Copyright 2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledFuture;
import java.util.function.BiConsumer;
import java.util.function.Function;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
/**
* Base class for {@link CompletableFuture}s returned by {@link AsyncRabbitTemplate2}.
* @param <T> the type.
*
* @author Gary Russell
* @since 2.4.7
*/
public abstract class RabbitFuture<T> extends CompletableFuture<T> {
private final String correlationId;
private final Message requestMessage;
private final BiConsumer<String, ChannelHolder> canceler;
private final Function<RabbitFuture<?>, ScheduledFuture<?>> timeoutTaskFunction;
private ScheduledFuture<?> timeoutTask;
private volatile CompletableFuture<Boolean> confirm;
private String nackCause;
private ChannelHolder channelHolder;
protected RabbitFuture(String correlationId, Message requestMessage, BiConsumer<String, ChannelHolder> canceler,
Function<RabbitFuture<?>, ScheduledFuture<?>> timeoutTaskFunction) {
this.correlationId = correlationId;
this.requestMessage = requestMessage;
this.canceler = canceler;
this.timeoutTaskFunction = timeoutTaskFunction;
}
void setChannelHolder(ChannelHolder channel) {
this.channelHolder = channel;
}
String getCorrelationId() {
return this.correlationId;
}
ChannelHolder getChannelHolder() {
return this.channelHolder;
}
Message getRequestMessage() {
return this.requestMessage;
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
if (this.timeoutTask != null) {
this.timeoutTask.cancel(true);
}
this.canceler.accept(this.correlationId, this.channelHolder);
return super.cancel(mayInterruptIfRunning);
}
/**
* When confirms are enabled contains a {@link CompletableFuture}
* for the confirmation.
* @return the future.
*/
public CompletableFuture<Boolean> getConfirm() {
return this.confirm;
}
void setConfirm(CompletableFuture<Boolean> confirm) {
this.confirm = confirm;
}
/**
* When confirms are enabled and a nack is received, contains
* the cause for the nack, if any.
* @return the cause.
*/
public String getNackCause() {
return this.nackCause;
}
void setNackCause(String nackCause) {
this.nackCause = nackCause;
}
void startTimer() {
this.timeoutTask = this.timeoutTaskFunction.apply(this);
}
}

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

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

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

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

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2021 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,7 +17,6 @@
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;
@@ -44,8 +43,6 @@ 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;
@@ -93,22 +90,11 @@ 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

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2021 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,7 +935,6 @@ public class PublisherCallbackChannelImpl
}
}
@SuppressWarnings("deprecation")
private void doProcessAck(long seq, boolean ack, boolean multiple, boolean remove) {
if (multiple) {
processMultipleAck(seq, ack);
@@ -972,7 +971,6 @@ public class PublisherCallbackChannelImpl
}
}
@SuppressWarnings("deprecation")
private void processMultipleAck(long seq, boolean ack) {
/*
* Piggy-backed ack - extract all Listeners for this and earlier

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

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021 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,7 +108,6 @@ public class RepublishMessageRecovererWithConfirms extends RepublishMessageRecov
}
}
@SuppressWarnings("deprecation")
private void doSendCorrelated(String exchange, String routingKey, Message message) {
CorrelationData cd = new CorrelationData();
if (exchange != null) {

578
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2Tests.java
View File

@@ -1,578 +0,0 @@
/*
* 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.
* 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 static org.mockito.Mockito.mock;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import org.junit.jupiter.api.Test;
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.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener;
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);
RabbitConverterFuture<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);
RabbitMessageFuture 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);
RabbitConverterFuture<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);
RabbitMessageFuture 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);
RabbitConverterFuture<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);
RabbitConverterFuture<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();
}
@Test
void ctorCoverage() {
AsyncRabbitTemplate2 template = new AsyncRabbitTemplate2(mock(ConnectionFactory.class), "ex", "rk");
assertThat(template).extracting(t -> t.getRabbitTemplate())
.extracting("exchange")
.isEqualTo("ex");
assertThat(template).extracting(t -> t.getRabbitTemplate())
.extracting("routingKey")
.isEqualTo("rk");
template = new AsyncRabbitTemplate2(mock(ConnectionFactory.class), "ex", "rk", "rq");
assertThat(template).extracting(t -> t.getRabbitTemplate())
.extracting("exchange")
.isEqualTo("ex");
assertThat(template).extracting(t -> t.getRabbitTemplate())
.extracting("routingKey")
.isEqualTo("rk");
assertThat(template)
.extracting("replyAddress")
.isEqualTo("rq");
assertThat(template).extracting("container")
.extracting("queueNames")
.isEqualTo(new String[] { "rq" });
template = new AsyncRabbitTemplate2(mock(ConnectionFactory.class), "ex", "rk", "rq", "ra");
assertThat(template).extracting(t -> t.getRabbitTemplate())
.extracting("exchange")
.isEqualTo("ex");
assertThat(template).extracting(t -> t.getRabbitTemplate())
.extracting("routingKey")
.isEqualTo("rk");
assertThat(template)
.extracting("replyAddress")
.isEqualTo("ra");
assertThat(template).extracting("container")
.extracting("queueNames")
.isEqualTo(new String[] { "rq" });
template = new AsyncRabbitTemplate2(mock(RabbitTemplate.class), mock(AbstractMessageListenerContainer.class),
"rq");
assertThat(template)
.extracting("replyAddress")
.isEqualTo("rq");
}
private void checkConverterResult(CompletableFuture<String> future, String expected) throws InterruptedException {
final CountDownLatch cdl = new CountDownLatch(1);
final AtomicReference<String> resultRef = new AtomicReference<>();
future.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;
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* 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.
@@ -38,6 +38,8 @@ 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;
@@ -69,7 +71,6 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
@SpringJUnitConfig
@DirtiesContext
@RabbitAvailable
@SuppressWarnings("deprecation")
public class AsyncRabbitTemplateTests {
@Autowired
@@ -252,7 +253,7 @@ public class AsyncRabbitTemplateTests {
@DirtiesContext
public void testConvertWithConfirm() throws Exception {
this.asyncTemplate.setEnableConfirms(true);
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
ListenableFuture<Boolean> confirm = future.getConfirm();
assertThat(confirm).isNotNull();
assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue();
@@ -263,7 +264,7 @@ public class AsyncRabbitTemplateTests {
@DirtiesContext
public void testMessageWithConfirm() throws Exception {
this.asyncTemplate.setEnableConfirms(true);
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture future = this.asyncTemplate
RabbitMessageFuture future = this.asyncTemplate
.sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties()));
ListenableFuture<Boolean> confirm = future.getConfirm();
assertThat(confirm).isNotNull();
@@ -275,7 +276,7 @@ public class AsyncRabbitTemplateTests {
@DirtiesContext
public void testConvertWithConfirmDirect() throws Exception {
this.asyncDirectTemplate.setEnableConfirms(true);
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture<String> future = this.asyncDirectTemplate.convertSendAndReceive("sleep");
RabbitConverterFuture<String> future = this.asyncDirectTemplate.convertSendAndReceive("sleep");
ListenableFuture<Boolean> confirm = future.getConfirm();
assertThat(confirm).isNotNull();
assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue();
@@ -286,7 +287,7 @@ public class AsyncRabbitTemplateTests {
@DirtiesContext
public void testMessageWithConfirmDirect() throws Exception {
this.asyncDirectTemplate.setEnableConfirms(true);
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture future = this.asyncDirectTemplate
RabbitMessageFuture future = this.asyncDirectTemplate
.sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties()));
ListenableFuture<Boolean> confirm = future.getConfirm();
assertThat(confirm).isNotNull();
@@ -320,7 +321,7 @@ public class AsyncRabbitTemplateTests {
@DirtiesContext
public void testReplyAfterReceiveTimeout() throws Exception {
this.asyncTemplate.setReceiveTimeout(100);
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("sleep");
TheCallback callback = new TheCallback();
future.addCallback(callback);
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1);
@@ -350,7 +351,7 @@ public class AsyncRabbitTemplateTests {
@DirtiesContext
public void testStopCancelled() throws Exception {
this.asyncTemplate.setReceiveTimeout(5000);
org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("noReply");
RabbitConverterFuture<String> future = this.asyncTemplate.convertSendAndReceive("noReply");
TheCallback callback = new TheCallback();
future.addCallback(callback);
assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1);

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

@@ -36,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.AsyncRabbitTemplate2;
import org.springframework.amqp.rabbit.RabbitConverterFuture;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -74,7 +74,7 @@ public class AsyncListenerTests {
private RabbitTemplate rabbitTemplate;
@Autowired
private AsyncRabbitTemplate2 asyncTemplate;
private AsyncRabbitTemplate asyncTemplate;
@Autowired
private Queue queue1;
@@ -193,8 +193,8 @@ public class AsyncListenerTests {
}
@Bean
public AsyncRabbitTemplate2 asyncTemplate() {
return new AsyncRabbitTemplate2(rabbitTemplate());
public AsyncRabbitTemplate asyncTemplate() {
return new AsyncRabbitTemplate(rabbitTemplate());
}
@Bean

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* 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.
@@ -23,8 +23,8 @@ import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate2;
import org.springframework.amqp.rabbit.RabbitConverterFuture;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitConverterFuture;
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 AsyncRabbitTemplate2 asyncTemplate;
private AsyncRabbitTemplate asyncTemplate;
private static Foo<Bar<Baz, Qux>> makeAFoo() {
Foo<Bar<Baz, Qux>> foo = new Foo<>();
@@ -169,8 +169,8 @@ public class ComplexTypeJsonIntegrationTests {
}
@Bean
public AsyncRabbitTemplate2 asyncTemplate() {
return new AsyncRabbitTemplate2(template());
public AsyncRabbitTemplate asyncTemplate() {
return new AsyncRabbitTemplate(template());
}
@Bean

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

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2022 the original author or authors.
* Copyright 2020-2021 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.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(data.getFuture().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.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(data.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(data.getReturned()).isNotNull();
ccf.destroy();
}

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -839,14 +839,14 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
admin.declareQueue(queue);
CorrelationData cd1 = new CorrelationData();
this.templateWithConfirmsEnabled.convertAndSend("", queue.getName(), "foo", cd1);
assertThat(cd1.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(cd1.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
CorrelationData cd2 = new CorrelationData();
this.templateWithConfirmsEnabled.convertAndSend("", queue.getName(), "bar", cd2);
assertThat(cd2.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse();
assertThat(cd2.getFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse();
CorrelationData cd3 = new CorrelationData();
this.templateWithConfirmsEnabled.convertAndSend("NO_EXCHANGE_HERE", queue.getName(), "foo", cd3);
assertThat(cd3.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse();
assertThat(cd3.getCompletableFuture().get().getReason()).contains("NOT_FOUND");
assertThat(cd3.getFuture().get(10, TimeUnit.SECONDS).isAck()).isFalse();
assertThat(cd3.getFuture().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.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(cd4.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(callbackLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(cd4.getReturned()).isNotNull();
assertThat(resent.get()).isTrue();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2021 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.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(corr.getFuture().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.getCompletableFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(corr.getReturned()).isNotNull();
}

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

@@ -109,7 +109,7 @@ class RabbitTemplateRoutingConnectionFactoryIntegrationTests {
rabbitTemplate.send(ROUTE, message, correlationData);
assertThat(rabbitTemplate.getUnconfirmedCount()).isEqualTo(1);
final CorrelationData.Confirm confirm = correlationData.getCompletableFuture().get(10, TimeUnit.SECONDS);
final CorrelationData.Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS);
assertThat(confirm.isAck()).isTrue();

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

@@ -4678,8 +4678,7 @@ 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`.
IMPORTANT: In version 3.0, `AsyncRabbitTemplate` method return types will be `CompletableFuture` instead of `ListenableFuture`.
[[remoting]]
===== Spring Remoting with AMQP

3
src/reference/asciidoc/stream.adoc
View File

@@ -67,8 +67,7 @@ The `ProducerCustomizer` provides a mechanism to customize the producer before i
Refer to the https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/[Java Client Documentation] about customizing the `Environment` and `Producer`.
IMPORTANT: In version 2.4.7 `RabbitStreamOperations` and `RabbitStreamTemplate` have been deprecated in favor of `RabbitStreamOperations2` and `RabbitStreamTemplate2` respectively; they return `CompletableFuture` instead of `ListenableFuture`.
`RabbitStreamOperations` and `RabbitStreamTemplate` will be removed in 3.0.
IMPORTANT: In version 3.0, the method return types will be `CompletableFuture` instead of `ListenableFuture`.
==== Receiving Messages

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

@@ -31,13 +31,3 @@ 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.
==== Stream Support Changes
`RabbitStreamOperations` and `RabbitStreamTemplate` have been deprecated in favor of `RabbitStreamOperations2` and `RabbitStreamTemplate2` respectively; they return `CompletableFuture` instead of `ListenableFuture`.
See <<stream-support>> for more information.