diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate2.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate2.java deleted file mode 100644 index 3d755305..00000000 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate2.java +++ /dev/null @@ -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 sendAndReceive(Message message); - - /** - * Send a message to the default exchange with the supplied routing key. If the message - * contains a correlationId property, it must be unique. - * @param routingKey the routing key. - * @param message the message. - * @return the {@link CompletableFuture}. - */ - CompletableFuture sendAndReceive(String routingKey, Message message); - - /** - * Send a message to the supplied exchange and routing key. If the message - * contains a correlationId property, it must be unique. - * @param exchange the exchange. - * @param routingKey the routing key. - * @param message the message. - * @return the {@link CompletableFuture}. - */ - CompletableFuture sendAndReceive(String exchange, String routingKey, Message message); - - /** - * Convert the object to a message and send it to the default exchange with the - * default routing key. - * @param object the object to convert. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceive(Object object); - - /** - * Convert the object to a message and send it to the default exchange with the - * provided routing key. - * @param routingKey the routing key. - * @param object the object to convert. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceive(String routingKey, Object object); - - /** - * Convert the object to a message and send it to the provided exchange and - * routing key. - * @param exchange the exchange. - * @param routingKey the routing key. - * @param object the object to convert. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceive(String exchange, String routingKey, Object object); - - /** - * Convert the object to a message and send it to the default exchange with the - * default routing key after invoking the {@link MessagePostProcessor}. - * If the post processor adds a correlationId property, it must be unique. - * @param object the object to convert. - * @param messagePostProcessor the post processor. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor); - - /** - * Convert the object to a message and send it to the default exchange with the - * provided routing key after invoking the {@link MessagePostProcessor}. - * If the post processor adds a correlationId property, it must be unique. - * @param routingKey the routing key. - * @param object the object to convert. - * @param messagePostProcessor the post processor. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceive(String routingKey, Object object, - MessagePostProcessor messagePostProcessor); - - /** - * Convert the object to a message and send it to the provided exchange and - * routing key after invoking the {@link MessagePostProcessor}. - * If the post processor adds a correlationId property, it must be unique. - * @param exchange the exchange - * @param routingKey the routing key. - * @param object the object to convert. - * @param messagePostProcessor the post processor. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceive(String exchange, String routingKey, Object object, - MessagePostProcessor messagePostProcessor); - - /** - * Convert the object to a message and send it to the default exchange with the - * default routing key. - * @param object the object to convert. - * @param responseType the response type. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceiveAsType(Object object, ParameterizedTypeReference responseType); - - /** - * Convert the object to a message and send it to the default exchange with the - * provided routing key. - * @param routingKey the routing key. - * @param object the object to convert. - * @param responseType the response type. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceiveAsType(String routingKey, Object object, - ParameterizedTypeReference responseType); - - /** - * Convert the object to a message and send it to the provided exchange and - * routing key. - * @param exchange the exchange. - * @param routingKey the routing key. - * @param object the object to convert. - * @param responseType the response type. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, - ParameterizedTypeReference responseType); - - /** - * Convert the object to a message and send it to the default exchange with the - * default routing key after invoking the {@link MessagePostProcessor}. - * If the post processor adds a correlationId property, it must be unique. - * @param object the object to convert. - * @param messagePostProcessor the post processor. - * @param responseType the response type. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor, - ParameterizedTypeReference responseType); - - /** - * Convert the object to a message and send it to the default exchange with the - * provided routing key after invoking the {@link MessagePostProcessor}. - * If the post processor adds a correlationId property, it must be unique. - * @param routingKey the routing key. - * @param object the object to convert. - * @param messagePostProcessor the post processor. - * @param responseType the response type. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceiveAsType(String routingKey, Object object, - MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType); - - /** - * Convert the object to a message and send it to the provided exchange and - * routing key after invoking the {@link MessagePostProcessor}. - * If the post processor adds a correlationId property, it must be unique. - * @param exchange the exchange - * @param routingKey the routing key. - * @param object the object to convert. - * @param messagePostProcessor the post processor. - * @param responseType the response type. - * @param the expected result type. - * @return the {@link CompletableFuture}. - */ - CompletableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, - MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType); - -} diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamOperations2.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamOperations2.java deleted file mode 100644 index a60c489c..00000000 --- a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamOperations2.java +++ /dev/null @@ -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 send(Message message); - - /** - * Convert to and send a Spring AMQP message. - * @param message the payload. - * @return a future to indicate success/failure. - */ - CompletableFuture 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 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 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 - } - -} diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java index ca3a89de..487d3cf6 100644 --- a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java +++ b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate.java @@ -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 diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate2.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate2.java deleted file mode 100644 index 236d0675..00000000 --- a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplate2.java +++ /dev/null @@ -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 send(Message message) { - CompletableFuture future = new CompletableFuture<>(); - createOrGetProducer().send(this.streamConverter.fromMessage(message), handleConfirm(future)); - return future; - } - - @Override - public CompletableFuture convertAndSend(Object message) { - return convertAndSend(message, null); - } - - @Override - public CompletableFuture 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 future = new CompletableFuture<>(); - future.complete(false); - return future; - } - } - return send(message2); - } - - - @Override - public CompletableFuture send(com.rabbitmq.stream.Message message) { - CompletableFuture future = new CompletableFuture<>(); - createOrGetProducer().send(message, handleConfirm(future)); - return future; - } - - @Override - public MessageBuilder messageBuilder() { - return createOrGetProducer().messageBuilder(); - } - - private ConfirmationHandler handleConfirm(CompletableFuture 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} - *

- * Close the underlying producer; a new producer will be created on the next - * operation that requires one. - */ - @Override - public synchronized void close() { - if (this.producer != null) { - this.producer.close(); - this.producer = null; - } - } - -} diff --git a/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java b/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java index 652ee1d8..b0f914d7 100644 --- a/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java +++ b/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java @@ -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 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; } diff --git a/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplateTests.java b/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplateTests.java index 888cfa73..fc4c2fa0 100644 --- a/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplateTests.java +++ b/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/producer/RabbitStreamTemplateTests.java @@ -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 future = template.convertAndSend("foo"); + ListenableFuture future = template.convertAndSend("foo"); assertThat(future.get()).isTrue(); - CompletableFuture future1 = template.convertAndSend("foo"); + ListenableFuture future1 = template.convertAndSend("foo"); assertThatExceptionOfType(ExecutionException.class).isThrownBy(() -> future1.get()) .withCauseExactlyInstanceOf(StreamSendException.class) .withStackTraceContaining("Message Enqueueing Failed"); - CompletableFuture future2 = template.convertAndSend("foo"); + ListenableFuture future2 = template.convertAndSend("foo"); assertThatExceptionOfType(ExecutionException.class).isThrownBy(() -> future2.get()) .withCauseExactlyInstanceOf(StreamSendException.class) .withStackTraceContaining("Producer Closed"); - CompletableFuture future3 = template.convertAndSend("foo"); + ListenableFuture future3 = template.convertAndSend("foo"); assertThatExceptionOfType(ExecutionException.class).isThrownBy(() -> future3.get()) .withCauseExactlyInstanceOf(StreamSendException.class) .withStackTraceContaining("Producer Not Available"); - CompletableFuture future4 = template.convertAndSend("foo"); + ListenableFuture future4 = template.convertAndSend("foo"); assertThatExceptionOfType(ExecutionException.class).isThrownBy(() -> future4.get()) .withCauseExactlyInstanceOf(StreamSendException.class) .withStackTraceContaining("Publish Confirm Timeout"); - CompletableFuture future5 = template.convertAndSend("foo"); + ListenableFuture future5 = template.convertAndSend("foo"); assertThatExceptionOfType(ExecutionException.class).isThrownBy(() -> future5.get()) .withCauseExactlyInstanceOf(StreamSendException.class) .withStackTraceContaining("Unknown code: " + -1); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java index a45ff16f..5b767c7b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java @@ -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 { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2.java deleted file mode 100644 index 424c7d0d..00000000 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2.java +++ /dev/null @@ -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. - *

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

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

- * Internally, the template uses a {@link RabbitTemplate} and an - * {@link AbstractMessageListenerContainer} either provided or constructed internally - * (a {@link SimpleMessageListenerContainer}). - * If an external {@link RabbitTemplate} is provided and confirms/returns are enabled, - * it must not previously have had callbacks registered because this object needs to - * be the callback. - * - * @author Gary Russell - * @author Artem Bilan - * - * @since 1.6 - */ -public class AsyncRabbitTemplate2 implements AsyncAmqpTemplate2, ChannelAwareMessageListener, ReturnsCallback, - ConfirmCallback, BeanNameAware, SmartLifecycle { - - public static final int DEFAULT_RECEIVE_TIMEOUT = 30000; - - private final Log logger = LogFactory.getLog(this.getClass()); - - private final RabbitTemplate template; - - private final AbstractMessageListenerContainer container; - - private final DirectReplyToMessageListenerContainer directReplyToContainer; - - private final String replyAddress; - - private final ConcurrentMap> pending = new ConcurrentHashMap<>(); - - private final CorrelationMessagePostProcessor messagePostProcessor = new CorrelationMessagePostProcessor<>(); - - private volatile boolean running; - - private volatile boolean enableConfirms; - - private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; - - private int phase; - - private boolean autoStartup = true; - - private String beanName; - - private TaskScheduler taskScheduler; - - private boolean internalTaskScheduler = true; - - /** - * Construct an instance using the provided arguments. Replies will be - * routed to the default exchange using the reply queue name as the routing - * key. - * @param connectionFactory the connection factory. - * @param exchange the default exchange to which requests will be sent. - * @param routingKey the default routing key. - * @param replyQueue the name of the reply queue to listen for replies. - */ - public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey, - String replyQueue) { - this(connectionFactory, exchange, routingKey, replyQueue, null); - } - - /** - * Construct an instance using the provided arguments. If 'replyAddress' is null, - * replies will be routed to the default exchange using the reply queue name as the - * routing key. Otherwise it should have the form exchange/routingKey and must - * cause messages to be routed to the reply queue. - * @param connectionFactory the connection factory. - * @param exchange the default exchange to which requests will be sent. - * @param routingKey the default routing key. - * @param replyQueue the name of the reply queue to listen for replies. - * @param replyAddress the reply address (exchange/routingKey). - */ - public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey, - String replyQueue, String replyAddress) { - Assert.notNull(connectionFactory, "'connectionFactory' cannot be null"); - Assert.notNull(routingKey, "'routingKey' cannot be null"); - Assert.notNull(replyQueue, "'replyQueue' cannot be null"); - this.template = new RabbitTemplate(connectionFactory); - this.template.setExchange(exchange == null ? "" : exchange); - this.template.setRoutingKey(routingKey); - this.container = new SimpleMessageListenerContainer(connectionFactory); - JavaUtils.INSTANCE - .acceptIfNotNull(this.template.getAfterReceivePostProcessors(), - (value) -> this.container.setAfterReceivePostProcessors( - value.toArray(new MessagePostProcessor[0]))); - this.container.setQueueNames(replyQueue); - this.container.setMessageListener(this); - this.container.afterPropertiesSet(); - this.directReplyToContainer = null; - if (replyAddress == null) { - this.replyAddress = replyQueue; - } - else { - this.replyAddress = replyAddress; - } - - } - - /** - * Construct an instance using the provided arguments. The first queue the container - * is configured to listen to will be used as the reply queue. Replies will be - * routed using the default exchange with that queue name as the routing key. - * @param template a {@link RabbitTemplate} - * @param container a {@link AbstractMessageListenerContainer}. - */ - public AsyncRabbitTemplate2(RabbitTemplate template, AbstractMessageListenerContainer container) { - this(template, container, null); - } - - /** - * Construct an instance using the provided arguments. The first queue the container - * is configured to listen to will be used as the reply queue. If 'replyAddress' is - * null, replies will be routed using the default exchange with that queue name as the - * routing key. Otherwise it should have the form exchange/routingKey and must - * cause messages to be routed to the reply queue. - * @param template a {@link RabbitTemplate}. - * @param container a {@link AbstractMessageListenerContainer}. - * @param replyAddress the reply address. - */ - public AsyncRabbitTemplate2(RabbitTemplate template, AbstractMessageListenerContainer container, - String replyAddress) { - Assert.notNull(template, "'template' cannot be null"); - Assert.notNull(container, "'container' cannot be null"); - this.template = template; - this.container = container; - this.container.setMessageListener(this); - this.directReplyToContainer = null; - if (replyAddress == null) { - this.replyAddress = container.getQueueNames()[0]; - } - else { - this.replyAddress = replyAddress; - } - } - - /** - * Construct an instance using the provided arguments. "Direct replyTo" is used for - * replies. - * @param connectionFactory the connection factory. - * @param exchange the default exchange to which requests will be sent. - * @param routingKey the default routing key. - * @since 2.0 - */ - public AsyncRabbitTemplate2(ConnectionFactory connectionFactory, String exchange, String routingKey) { - this(new RabbitTemplate(connectionFactory)); - Assert.notNull(routingKey, "'routingKey' cannot be null"); - this.template.setExchange(exchange == null ? "" : exchange); - this.template.setRoutingKey(routingKey); - } - - /** - * Construct an instance using the provided arguments. "Direct replyTo" is used for - * replies. - * @param template a {@link RabbitTemplate} - * @since 2.0 - */ - public AsyncRabbitTemplate2(RabbitTemplate template) { - Assert.notNull(template, "'template' cannot be null"); - this.template = template; - this.container = null; - this.replyAddress = null; - this.directReplyToContainer = new DirectReplyToMessageListenerContainer(this.template.getConnectionFactory()); - JavaUtils.INSTANCE - .acceptIfNotNull(template.getAfterReceivePostProcessors(), - (value) -> this.directReplyToContainer.setAfterReceivePostProcessors( - value.toArray(new MessagePostProcessor[0]))); - this.directReplyToContainer.setMessageListener(this); - } - - /** - * @param autoStartup true for auto start. - * @see #isAutoStartup() - */ - public void setAutoStartup(boolean autoStartup) { - this.autoStartup = autoStartup; - } - - /** - * @param phase the phase. - * @see #getPhase() - */ - public void setPhase(int phase) { - this.phase = phase; - } - - /** - * Set to true to enable the receipt of returned messages that cannot be delivered - * in the form of a {@link AmqpMessageReturnedException}. - * @param mandatory true to enable returns. - */ - public void setMandatory(boolean mandatory) { - this.template.setReturnsCallback(this); - this.template.setMandatory(mandatory); - } - - /** - * @param mandatoryExpression a SpEL {@link Expression} to evaluate against each request - * message. The result of the evaluation must be a {@code boolean} value. - * @since 2.0 - */ - public void setMandatoryExpression(Expression mandatoryExpression) { - this.template.setReturnsCallback(this); - this.template.setMandatoryExpression(mandatoryExpression); - } - - /** - * @param mandatoryExpression a SpEL {@link Expression} to evaluate against each request - * message. The result of the evaluation must be a {@code boolean} value. - * @since 2.0 - */ - public void setMandatoryExpressionString(String mandatoryExpression) { - this.template.setReturnsCallback(this); - this.template.setMandatoryExpressionString(mandatoryExpression); - } - - /** - * Set to true to enable publisher confirms. When enabled, the {@link RabbitFuture} - * returned by the send and receive operation will have a - * {@code CompletableFuture} in its {@code confirm} property. - * @param enableConfirms true to enable publisher confirms. - */ - public void setEnableConfirms(boolean enableConfirms) { - this.enableConfirms = enableConfirms; - if (enableConfirms) { - this.template.setConfirmCallback(this); - } - } - - public String getBeanName() { - return this.beanName; - } - - @Override - public void setBeanName(String beanName) { - this.beanName = beanName; - } - - /** - * @return a reference to the underlying connection factory in the - * {@link RabbitTemplate}. - */ - public ConnectionFactory getConnectionFactory() { - return this.template.getConnectionFactory(); - } - - /** - * Set the receive timeout - the future returned by the send and receive - * methods will be canceled when this timeout expires. {@code <= 0} means - * futures never expire. Beware that this will cause a memory leak if a - * reply is not received. Default: 30000 (30 seconds). - * @param receiveTimeout the timeout in milliseconds. - */ - public void setReceiveTimeout(long receiveTimeout) { - this.receiveTimeout = receiveTimeout; - } - - /** - * Set the task scheduler to expire timed out futures. - * @param taskScheduler the task scheduler - * @see #setReceiveTimeout(long) - */ - public synchronized void setTaskScheduler(TaskScheduler taskScheduler) { - Assert.notNull(taskScheduler, "'taskScheduler' cannot be null"); - this.internalTaskScheduler = false; - this.taskScheduler = taskScheduler; - } - - /** - * @return a reference to the underlying {@link RabbitTemplate}'s - * {@link MessageConverter}. - */ - public MessageConverter getMessageConverter() { - return this.template.getMessageConverter(); - } - - /** - * Return the underlying {@link RabbitTemplate} used for sending. - * @return the template. - * @since 2.2 - */ - public RabbitTemplate getRabbitTemplate() { - return this.template; - } - - @Override - public 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 RabbitConverterFuture convertSendAndReceive(Object object) { - return convertSendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), object, null); - } - - @Override - public RabbitConverterFuture convertSendAndReceive(String routingKey, Object object) { - return convertSendAndReceive(this.template.getExchange(), routingKey, object, null); - } - - @Override - public RabbitConverterFuture convertSendAndReceive(String exchange, String routingKey, Object object) { - return convertSendAndReceive(exchange, routingKey, object, null); - } - - @Override - public RabbitConverterFuture convertSendAndReceive(Object object, - MessagePostProcessor messagePostProcessor) { - return convertSendAndReceive(this.template.getExchange(), this.template.getRoutingKey(), object, - messagePostProcessor); - } - - @Override - public RabbitConverterFuture convertSendAndReceive(String routingKey, Object object, - MessagePostProcessor messagePostProcessor) { - return convertSendAndReceive(this.template.getExchange(), routingKey, object, messagePostProcessor); - } - - @Override - public RabbitConverterFuture convertSendAndReceive(String exchange, String routingKey, Object object, - MessagePostProcessor messagePostProcessor) { - return convertSendAndReceive(exchange, routingKey, object, messagePostProcessor, null); - } - - @Override - public RabbitConverterFuture convertSendAndReceiveAsType(Object object, - ParameterizedTypeReference responseType) { - return convertSendAndReceiveAsType(this.template.getExchange(), this.template.getRoutingKey(), object, - null, responseType); - } - - @Override - public RabbitConverterFuture convertSendAndReceiveAsType(String routingKey, Object object, - ParameterizedTypeReference responseType) { - return convertSendAndReceiveAsType(this.template.getExchange(), routingKey, object, null, responseType); - } - - @Override - public RabbitConverterFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, - ParameterizedTypeReference responseType) { - return convertSendAndReceiveAsType(exchange, routingKey, object, null, responseType); - } - - @Override - public RabbitConverterFuture convertSendAndReceiveAsType(Object object, - MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { - return convertSendAndReceiveAsType(this.template.getExchange(), this.template.getRoutingKey(), object, - messagePostProcessor, responseType); - } - - @Override - public RabbitConverterFuture convertSendAndReceiveAsType(String routingKey, Object object, - MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { - return convertSendAndReceiveAsType(this.template.getExchange(), routingKey, object, messagePostProcessor, - responseType); - } - - @Override - public RabbitConverterFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, - MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { - Assert.state(this.template.getMessageConverter() instanceof SmartMessageConverter, - "template's message converter must be a SmartMessageConverter"); - return convertSendAndReceive(exchange, routingKey, object, messagePostProcessor, responseType); - } - - private RabbitConverterFuture convertSendAndReceive(String exchange, String routingKey, Object object, - MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { - - AsyncCorrelationData correlationData = new AsyncCorrelationData(messagePostProcessor, responseType, - this.enableConfirms); - if (this.container != null) { - this.template.convertAndSend(exchange, routingKey, object, this.messagePostProcessor, correlationData); - } - else { - MessageConverter converter = this.template.getMessageConverter(); - if (converter == null) { - throw new AmqpIllegalStateException( - "No 'messageConverter' specified. Check configuration of RabbitTemplate."); - } - Message message = converter.toMessage(object, new MessageProperties()); - this.messagePostProcessor.postProcessMessage(message, correlationData, - this.template.nullSafeExchange(exchange), this.template.nullSafeRoutingKey(routingKey)); - ChannelHolder channelHolder = this.directReplyToContainer.getChannelHolder(); - correlationData.future.setChannelHolder(channelHolder); - sendDirect(channelHolder.getChannel(), exchange, routingKey, message, correlationData); - } - RabbitConverterFuture 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 rabbitFuture = (RabbitConverterFuture) 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 implements MessagePostProcessor { - - CorrelationMessagePostProcessor() { - } - - @Override - public Message postProcessMessage(Message message) throws AmqpException { - throw new UnsupportedOperationException(); - } - - @SuppressWarnings("unchecked") - @Override - public Message postProcessMessage(Message message, Correlation correlation) throws AmqpException { - Message messageToSend = message; - AsyncCorrelationData correlationData = (AsyncCorrelationData) correlation; - if (correlationData.userPostProcessor != null) { - messageToSend = correlationData.userPostProcessor.postProcessMessage(message); - } - String correlationId = getOrSetCorrelationIdAndSetReplyTo(messageToSend, correlationData); - correlationData.future = new RabbitConverterFuture(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 extends CorrelationData { - - final MessagePostProcessor userPostProcessor; // NOSONAR - - final ParameterizedTypeReference returnType; // NOSONAR - - final boolean enableConfirms; // NOSONAR - - volatile RabbitConverterFuture future; // NOSONAR - - AsyncCorrelationData(MessagePostProcessor userPostProcessor, ParameterizedTypeReference returnType, - boolean enableConfirms) { - - this.userPostProcessor = userPostProcessor; - this.returnType = returnType; - this.enableConfirms = enableConfirms; - } - - } - -} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitConverterFuture.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitConverterFuture.java deleted file mode 100644 index fe799c57..00000000 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitConverterFuture.java +++ /dev/null @@ -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 the type. - * - * @author Gary Russell - * @since 2.4.7 - */ -public class RabbitConverterFuture extends RabbitFuture { - - private volatile ParameterizedTypeReference returnType; - - RabbitConverterFuture(String correlationId, Message requestMessage, - BiConsumer canceler, - Function, ScheduledFuture> timeoutTaskFunction) { - - super(correlationId, requestMessage, canceler, timeoutTaskFunction); - } - - public ParameterizedTypeReference getReturnType() { - return this.returnType; - } - - public void setReturnType(ParameterizedTypeReference returnType) { - this.returnType = returnType; - } - -} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitFuture.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitFuture.java deleted file mode 100644 index 6e8638e9..00000000 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitFuture.java +++ /dev/null @@ -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 the type. - * - * @author Gary Russell - * @since 2.4.7 - */ -public abstract class RabbitFuture extends CompletableFuture { - - private final String correlationId; - - private final Message requestMessage; - - private final BiConsumer canceler; - - private final Function, ScheduledFuture> timeoutTaskFunction; - - private ScheduledFuture timeoutTask; - - private volatile CompletableFuture confirm; - - private String nackCause; - - private ChannelHolder channelHolder; - - protected RabbitFuture(String correlationId, Message requestMessage, BiConsumer canceler, - Function, ScheduledFuture> timeoutTaskFunction) { - - this.correlationId = correlationId; - this.requestMessage = requestMessage; - this.canceler = canceler; - this.timeoutTaskFunction = timeoutTaskFunction; - } - - void setChannelHolder(ChannelHolder channel) { - this.channelHolder = channel; - } - - String getCorrelationId() { - return this.correlationId; - } - - ChannelHolder getChannelHolder() { - return this.channelHolder; - } - - Message getRequestMessage() { - return this.requestMessage; - } - - @Override - public boolean cancel(boolean mayInterruptIfRunning) { - if (this.timeoutTask != null) { - this.timeoutTask.cancel(true); - } - this.canceler.accept(this.correlationId, this.channelHolder); - return super.cancel(mayInterruptIfRunning); - } - - /** - * When confirms are enabled contains a {@link CompletableFuture} - * for the confirmation. - * @return the future. - */ - public CompletableFuture getConfirm() { - return this.confirm; - } - - void setConfirm(CompletableFuture confirm) { - this.confirm = confirm; - } - - /** - * When confirms are enabled and a nack is received, contains - * the cause for the nack, if any. - * @return the cause. - */ - public String getNackCause() { - return this.nackCause; - } - - void setNackCause(String nackCause) { - this.nackCause = nackCause; - } - - void startTimer() { - this.timeoutTask = this.timeoutTaskFunction.apply(this); - } - -} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitMessageFuture.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitMessageFuture.java deleted file mode 100644 index 467053a3..00000000 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/RabbitMessageFuture.java +++ /dev/null @@ -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 { - - RabbitMessageFuture(String correlationId, Message requestMessage, BiConsumer canceler, - Function, ScheduledFuture> timeoutTaskFunction) { - - super(correlationId, requestMessage, canceler, timeoutTaskFunction); - } - -} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/TimeoutTask.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/TimeoutTask.java deleted file mode 100644 index 8ca3af16..00000000 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/TimeoutTask.java +++ /dev/null @@ -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> pending; - - private final DirectReplyToMessageListenerContainer container; - - TimeoutTask(RabbitFuture future, ConcurrentMap> pending, - @Nullable DirectReplyToMessageListenerContainer container) { - - this.future = future; - this.pending = pending; - this.container = container; - } - - @Override - public void run() { - this.pending.remove(this.future.getCorrelationId()); - ChannelHolder holder = this.future.getChannelHolder(); - if (holder != null && this.container != null) { - this.container.releaseConsumerFor(holder, false, null); // NOSONAR - } - this.future.completeExceptionally( - new AmqpReplyTimeoutException("Reply timed out", this.future.getRequestMessage())); - } - -} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java index 312a523b..14162a82 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-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 future = new SettableListenableFuture<>(); - private final CompletableFuture 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 getFuture() { return this.future; } - /** - * Return a future to check the success/failure of the publish operation. - * @return the future. - * @since 2.4.7 - */ - public CompletableFuture getCompletableFuture() { - return this.completable; - } - /** * Return a returned message, if any; requires a unique * {@link #CorrelationData(String) id}. Guaranteed to be populated before the future diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java index 4c01b4d5..01f710b2 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-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 diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java index cbdd107b..1c65f593 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-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) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2Tests.java deleted file mode 100644 index 6827251e..00000000 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate2Tests.java +++ /dev/null @@ -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 latch; - - private final Message fooMessage = new SimpleMessageConverter().toMessage("foo", new MessageProperties()); - - @Test - public void testConvert1Arg() throws Exception { - final AtomicBoolean mppCalled = new AtomicBoolean(); - CompletableFuture future = this.asyncTemplate.convertSendAndReceive("foo", m -> { - mppCalled.set(true); - return m; - }); - checkConverterResult(future, "FOO"); - assertThat(mppCalled.get()).isTrue(); - } - - @Test - public void testConvert1ArgDirect() throws Exception { - this.latch.set(new CountDownLatch(1)); - CompletableFuture future1 = this.asyncDirectTemplate.convertSendAndReceive("foo"); - CompletableFuture future2 = this.asyncDirectTemplate.convertSendAndReceive("bar"); - this.latch.get().countDown(); - checkConverterResult(future1, "FOO"); - checkConverterResult(future2, "BAR"); - this.latch.set(null); - waitForZeroInUseConsumers(); - assertThat(TestUtils - .getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount", - Integer.class)).isEqualTo(2); - final String missingQueue = UUID.randomUUID().toString(); - this.asyncDirectTemplate.convertSendAndReceive("", missingQueue, "foo"); // send to nowhere - this.asyncDirectTemplate.stop(); // should clear the inUse channel map - waitForZeroInUseConsumers(); - this.asyncDirectTemplate.start(); - this.asyncDirectTemplate.setReceiveTimeout(1); - this.asyncDirectTemplate.convertSendAndReceive("", missingQueue, "foo"); // send to nowhere - waitForZeroInUseConsumers(); - - this.asyncDirectTemplate.setReceiveTimeout(10000); - this.asyncDirectTemplate.convertSendAndReceive("", missingQueue, "foo").cancel(true); - waitForZeroInUseConsumers(); - } - - @Test - public void testConvert2Args() throws Exception { - CompletableFuture future = this.asyncTemplate.convertSendAndReceive(this.requests.getName(), "foo"); - checkConverterResult(future, "FOO"); - } - - @Test - public void testConvert3Args() throws Exception { - CompletableFuture future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo"); - checkConverterResult(future, "FOO"); - } - - @Test - public void testConvert4Args() throws Exception { - CompletableFuture future = this.asyncTemplate.convertSendAndReceive("", this.requests.getName(), "foo", - message -> { - String body = new String(message.getBody()); - return new Message((body + "bar").getBytes(), message.getMessageProperties()); - }); - checkConverterResult(future, "FOOBAR"); - } - - @Test - public void testMessage1Arg() throws Exception { - CompletableFuture future = this.asyncTemplate.sendAndReceive(getFooMessage()); - checkMessageResult(future, "FOO"); - } - - @Test - public void testMessage1ArgDirect() throws Exception { - this.latch.set(new CountDownLatch(1)); - CompletableFuture future1 = this.asyncDirectTemplate.sendAndReceive(getFooMessage()); - CompletableFuture future2 = this.asyncDirectTemplate.sendAndReceive(getFooMessage()); - this.latch.get().countDown(); - Message reply1 = checkMessageResult(future1, "FOO"); - assertThat(reply1.getMessageProperties().getConsumerQueue()).isEqualTo(Address.AMQ_RABBITMQ_REPLY_TO); - Message reply2 = checkMessageResult(future2, "FOO"); - assertThat(reply2.getMessageProperties().getConsumerQueue()).isEqualTo(Address.AMQ_RABBITMQ_REPLY_TO); - this.latch.set(null); - waitForZeroInUseConsumers(); - assertThat(TestUtils - .getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount", - Integer.class)).isEqualTo(2); - this.asyncDirectTemplate.stop(); - this.asyncDirectTemplate.start(); - assertThat(TestUtils - .getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount", - Integer.class)).isEqualTo(0); - } - - private void waitForZeroInUseConsumers() throws InterruptedException { - Map inUseConsumers = TestUtils - .getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.inUseConsumerChannels", Map.class); - await().until(() -> inUseConsumers.size() == 0); - } - - @Test - public void testMessage2Args() throws Exception { - CompletableFuture future = this.asyncTemplate.sendAndReceive(this.requests.getName(), getFooMessage()); - checkMessageResult(future, "FOO"); - } - - @Test - public void testMessage3Args() throws Exception { - CompletableFuture future = this.asyncTemplate.sendAndReceive("", this.requests.getName(), - getFooMessage()); - checkMessageResult(future, "FOO"); - } - - @SuppressWarnings("unchecked") - @Test - public void testCancel() { - CompletableFuture future = this.asyncTemplate.convertSendAndReceive("foo"); - future.cancel(false); - assertThat(TestUtils.getPropertyValue(asyncTemplate, "pending", Map.class)).hasSize(0); - } - - @Test - public void testMessageCustomCorrelation() throws Exception { - Message message = getFooMessage(); - message.getMessageProperties().setCorrelationId("foo"); - CompletableFuture future = this.asyncTemplate.sendAndReceive(message); - Message result = checkMessageResult(future, "FOO"); - assertThat(result.getMessageProperties().getCorrelationId()).isEqualTo("foo"); - } - - private Message getFooMessage() { - this.fooMessage.getMessageProperties().setCorrelationId(null); - this.fooMessage.getMessageProperties().setReplyTo(null); - return this.fooMessage; - } - - @Test - @DirtiesContext - public void testReturn() throws Exception { - this.asyncTemplate.setMandatory(true); - CompletableFuture future = this.asyncTemplate.convertSendAndReceive(this.requests.getName() + "x", - "foo"); - try { - future.get(10, TimeUnit.SECONDS); - fail("Expected exception"); - } - catch (ExecutionException e) { - assertThat(e.getCause()).isInstanceOf(AmqpMessageReturnedException.class); - assertThat(((AmqpMessageReturnedException) e.getCause()).getRoutingKey()).isEqualTo(this.requests.getName() + "x"); - } - } - - @Test - @DirtiesContext - public void testReturnDirect() throws Exception { - this.asyncDirectTemplate.setMandatory(true); - CompletableFuture future = this.asyncDirectTemplate.convertSendAndReceive(this.requests.getName() + "x", - "foo"); - try { - future.get(10, TimeUnit.SECONDS); - fail("Expected exception"); - } - catch (ExecutionException e) { - assertThat(e.getCause()).isInstanceOf(AmqpMessageReturnedException.class); - assertThat(((AmqpMessageReturnedException) e.getCause()).getRoutingKey()).isEqualTo(this.requests.getName() + "x"); - } - } - - @Test - @DirtiesContext - public void testConvertWithConfirm() throws Exception { - this.asyncTemplate.setEnableConfirms(true); - RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("sleep"); - CompletableFuture confirm = future.getConfirm(); - assertThat(confirm).isNotNull(); - assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); - checkConverterResult(future, "SLEEP"); - } - - @Test - @DirtiesContext - public void testMessageWithConfirm() throws Exception { - this.asyncTemplate.setEnableConfirms(true); - RabbitMessageFuture future = this.asyncTemplate - .sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties())); - CompletableFuture confirm = future.getConfirm(); - assertThat(confirm).isNotNull(); - assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); - checkMessageResult(future, "SLEEP"); - } - - @Test - @DirtiesContext - public void testConvertWithConfirmDirect() throws Exception { - this.asyncDirectTemplate.setEnableConfirms(true); - RabbitConverterFuture future = this.asyncDirectTemplate.convertSendAndReceive("sleep"); - CompletableFuture confirm = future.getConfirm(); - assertThat(confirm).isNotNull(); - assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); - checkConverterResult(future, "SLEEP"); - } - - @Test - @DirtiesContext - public void testMessageWithConfirmDirect() throws Exception { - this.asyncDirectTemplate.setEnableConfirms(true); - RabbitMessageFuture future = this.asyncDirectTemplate - .sendAndReceive(new SimpleMessageConverter().toMessage("sleep", new MessageProperties())); - CompletableFuture confirm = future.getConfirm(); - assertThat(confirm).isNotNull(); - assertThat(confirm.get(10, TimeUnit.SECONDS)).isTrue(); - checkMessageResult(future, "SLEEP"); - } - - @SuppressWarnings("unchecked") - @Test - @DirtiesContext - public void testReceiveTimeout() throws Exception { - this.asyncTemplate.setReceiveTimeout(500); - CompletableFuture future = this.asyncTemplate.convertSendAndReceive("noReply"); - TheCallback callback = new TheCallback(); - future.whenComplete(callback); - assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1); - try { - future.get(10, TimeUnit.SECONDS); - fail("Expected ExecutionException"); - } - catch (ExecutionException e) { - assertThat(e.getCause()).isInstanceOf(AmqpReplyTimeoutException.class); - } - assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(0); - assertThat(callback.latch.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(callback.ex).isInstanceOf(AmqpReplyTimeoutException.class); - } - - @SuppressWarnings("unchecked") - @Test - @DirtiesContext - public void testReplyAfterReceiveTimeout() throws Exception { - this.asyncTemplate.setReceiveTimeout(100); - RabbitConverterFuture 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 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 future, String expected) throws InterruptedException { - final CountDownLatch cdl = new CountDownLatch(1); - final AtomicReference resultRef = new AtomicReference<>(); - future.whenComplete((result, ex) -> { - resultRef.set(result); - cdl.countDown(); - }); - assertThat(cdl.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(resultRef.get()).isEqualTo(expected); - } - - private Message checkMessageResult(CompletableFuture future, String expected) throws InterruptedException { - final CountDownLatch cdl = new CountDownLatch(1); - final AtomicReference resultRef = new AtomicReference<>(); - future.whenComplete((result, ex) -> { - resultRef.set(result); - cdl.countDown(); - }); - assertThat(cdl.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(new String(resultRef.get().getBody())).isEqualTo(expected); - return resultRef.get(); - } - - public static class TheCallback implements BiConsumer { - - private final CountDownLatch latch = new CountDownLatch(1); - - private volatile String result; - - private volatile Throwable ex; - - - @Override - public void accept(String result, Throwable ex) { - this.result = result; - this.ex = ex; - latch.countDown(); - } - - } - - @Configuration - public static class Config { - - @Bean - public AtomicReference latch() { - return new AtomicReference<>(); - } - - @Bean - public ConnectionFactory connectionFactory() { - CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost"); - connectionFactory.setPublisherConfirmType(ConfirmType.CORRELATED); - connectionFactory.setPublisherReturns(true); - return connectionFactory; - } - - @Bean - public Queue requests() { - return new AnonymousQueue(); - } - - @Bean - public Queue replies() { - return new AnonymousQueue(); - } - - @Bean - public RabbitAdmin admin(ConnectionFactory connectionFactory) { - return new RabbitAdmin(connectionFactory); - } - - @Bean - public GZipPostProcessor gZipPostProcessor() { - GZipPostProcessor gZipPostProcessor = new GZipPostProcessor(); - gZipPostProcessor.setCopyProperties(true); - return gZipPostProcessor; - } - - @Bean - public RabbitTemplate template(ConnectionFactory connectionFactory) { - RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); - rabbitTemplate.setRoutingKey(requests().getName()); - rabbitTemplate.addBeforePublishPostProcessors(gZipPostProcessor()); - rabbitTemplate.addAfterReceivePostProcessors(new GUnzipPostProcessor()); - return rabbitTemplate; - } - - @Bean - public RabbitTemplate templateForDirect(ConnectionFactory connectionFactory) { - RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); - rabbitTemplate.setRoutingKey(requests().getName()); - rabbitTemplate.addBeforePublishPostProcessors(gZipPostProcessor()); - rabbitTemplate.addAfterReceivePostProcessors(new GUnzipPostProcessor()); - return rabbitTemplate; - } - - @Bean - @Primary - public SimpleMessageListenerContainer replyContainer(ConnectionFactory connectionFactory) { - SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); - container.setAfterReceivePostProcessors(new GUnzipPostProcessor()); - container.setQueueNames(replies().getName()); - return container; - } - - @Bean - public AsyncRabbitTemplate2 asyncTemplate(RabbitTemplate template, SimpleMessageListenerContainer container) { - return new AsyncRabbitTemplate2(template, container); - } - - @Bean - public AsyncRabbitTemplate2 asyncDirectTemplate(RabbitTemplate templateForDirect) { - return new AsyncRabbitTemplate2(templateForDirect); - } - - @Bean - public SimpleMessageListenerContainer remoteContainer(ConnectionFactory connectionFactory) { - SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); - container.setQueueNames(requests().getName()); - container.setAfterReceivePostProcessors(new GUnzipPostProcessor()); - MessageListenerAdapter messageListener = - new MessageListenerAdapter((ReplyingMessageListener) - message -> { - CountDownLatch countDownLatch = latch().get(); - if (countDownLatch != null) { - try { - countDownLatch.await(10, TimeUnit.SECONDS); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } - if ("sleep".equals(message)) { - try { - Thread.sleep(500); // time for confirm to be delivered, or timeout to occur - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } - else if ("noReply".equals(message)) { - return null; - } - return message.toUpperCase(); - }); - - messageListener.setBeforeSendReplyPostProcessors(gZipPostProcessor()); - container.setMessageListener(messageListener); - return container; - } - - } - -} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java index ff684b1f..75eddf50 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-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 future = this.asyncTemplate.convertSendAndReceive("sleep"); + RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("sleep"); ListenableFuture 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 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 future = this.asyncDirectTemplate.convertSendAndReceive("sleep"); + RabbitConverterFuture future = this.asyncDirectTemplate.convertSendAndReceive("sleep"); ListenableFuture 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 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 future = this.asyncTemplate.convertSendAndReceive("sleep"); + RabbitConverterFuture 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 future = this.asyncTemplate.convertSendAndReceive("noReply"); + RabbitConverterFuture future = this.asyncTemplate.convertSendAndReceive("noReply"); TheCallback callback = new TheCallback(); future.addCallback(callback); assertThat(TestUtils.getPropertyValue(this.asyncTemplate, "pending", Map.class)).hasSize(1); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AsyncListenerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AsyncListenerTests.java index b7d1ad83..50195f34 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AsyncListenerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AsyncListenerTests.java @@ -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 diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java index 34b36e3c..1c2a504f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-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> makeAFoo() { Foo> 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 diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java index 62e8871b..e48dc3b2 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-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(); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index c8d89a47..2c9808fe 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -839,14 +839,14 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { admin.declareQueue(queue); CorrelationData cd1 = new CorrelationData(); this.templateWithConfirmsEnabled.convertAndSend("", queue.getName(), "foo", cd1); - assertThat(cd1.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 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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java index 2774bdd3..8d46fb24 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-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(); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateRoutingConnectionFactoryIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateRoutingConnectionFactoryIntegrationTests.java index 7084ef31..b2ebbbb7 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateRoutingConnectionFactoryIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateRoutingConnectionFactoryIntegrationTests.java @@ -109,7 +109,7 @@ class RabbitTemplateRoutingConnectionFactoryIntegrationTests { rabbitTemplate.send(ROUTE, message, correlationData); assertThat(rabbitTemplate.getUnconfirmedCount()).isEqualTo(1); - final CorrelationData.Confirm confirm = correlationData.getCompletableFuture().get(10, TimeUnit.SECONDS); + final CorrelationData.Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS); assertThat(confirm.isAck()).isTrue(); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index f35f18ae..33028a9d 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -4678,8 +4678,7 @@ Version 2.0 introduced variants of these methods (`convertSendAndReceiveAsType`) You must configure the underlying `RabbitTemplate` with a `SmartMessageConverter`. See <> for more information. -Starting with version 2.4.7, the `AsyncRabbitTemplate` is deprecated in favor of `AsyncRabbitTemplate2` which returns `CompletableFuture` s instead of `ListenableFuture` s. -In 3.0, `AsyncRabbitTemplate2` will be renamed to `AsyncRabbitTemplate`. +IMPORTANT: In version 3.0, `AsyncRabbitTemplate` method return types will be `CompletableFuture` instead of `ListenableFuture`. [[remoting]] ===== Spring Remoting with AMQP diff --git a/src/reference/asciidoc/stream.adoc b/src/reference/asciidoc/stream.adoc index 5849eca2..84fdc8f0 100644 --- a/src/reference/asciidoc/stream.adoc +++ b/src/reference/asciidoc/stream.adoc @@ -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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 4b8bb210..671cd01d 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -31,13 +31,3 @@ See <> for more information. The `Jackson2JsonMessageConverter` can now determine the charset from the `contentEncoding` header. See <> for more information. - -==== AsyncRabbitTemplate - -The `AsyncRabbitTemplate` is deprecated in favor of `AsyncRabbitTemplate2` which returns `CompletableFuture` s instead of `ListenableFuture` s. -See <> for more information. - -==== Stream Support Changes - -`RabbitStreamOperations` and `RabbitStreamTemplate` have been deprecated in favor of `RabbitStreamOperations2` and `RabbitStreamTemplate2` respectively; they return `CompletableFuture` instead of `ListenableFuture`. -See <> for more information.