diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java index fd3bf18124..b55f77c870 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java @@ -21,6 +21,7 @@ import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledFuture; import org.springframework.amqp.core.MessageDeliveryMode; @@ -51,7 +52,6 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; import org.springframework.util.StringUtils; -import org.springframework.util.concurrent.SettableListenableFuture; /** * A base {@link AbstractReplyProducingMessageHandler} extension for AMQP message handlers. @@ -703,7 +703,7 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin } @Override - public SettableListenableFuture getFuture() { + public CompletableFuture getFuture() { if (this.userData instanceof CorrelationData) { return ((CorrelationData) this.userData).getFuture(); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java index f05e6cfa31..d86665980c 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,11 +16,13 @@ package org.springframework.integration.amqp.outbound; +import java.util.function.BiConsumer; + import org.springframework.amqp.core.AmqpMessageReturnedException; import org.springframework.amqp.core.AmqpReplyTimeoutException; import org.springframework.amqp.core.ReturnedMessage; import org.springframework.amqp.rabbit.AsyncRabbitTemplate; -import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture; +import org.springframework.amqp.rabbit.RabbitMessageFuture; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm; import org.springframework.amqp.rabbit.core.RabbitTemplate; @@ -33,7 +35,6 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; -import org.springframework.util.concurrent.ListenableFutureCallback; /** * An outbound gateway where the sending thread is released immediately and the reply @@ -91,13 +92,13 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint { generateRoutingKey(requestMessage), amqpMessage); CorrelationData correlationData = generateCorrelationData(requestMessage); if (correlationData != null && future.getConfirm() != null) { - future.getConfirm().addCallback(new CorrelationCallback(correlationData, future)); + future.getConfirm().whenComplete(new CorrelationCallback(correlationData, future)); } - future.addCallback(new FutureCallback(requestMessage, correlationData)); + future.whenComplete(new FutureCallback(requestMessage, correlationData)); return null; } - private final class FutureCallback implements ListenableFutureCallback { + private final class FutureCallback implements BiConsumer { private final Message requestMessage; @@ -109,65 +110,65 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint { } @Override - public void onSuccess(org.springframework.amqp.core.Message result) { - AbstractIntegrationMessageBuilder replyMessageBuilder = null; - try { - replyMessageBuilder = buildReply(AsyncAmqpOutboundGateway.this.messageConverter, result); - sendOutputs(replyMessageBuilder, this.requestMessage); - } - catch (Exception ex) { - Exception exceptionToLogAndSend = ex; - if (!(ex instanceof MessagingException)) { // NOSONAR - exceptionToLogAndSend = new MessageHandlingException(this.requestMessage, - "failed to handle a message in the [" + AsyncAmqpOutboundGateway.this + ']', ex); - if (replyMessageBuilder != null) { - exceptionToLogAndSend = - new MessagingException(replyMessageBuilder.build(), exceptionToLogAndSend); + public void accept(org.springframework.amqp.core.Message message, Throwable throwable) { + if (throwable == null) { + AbstractIntegrationMessageBuilder replyMessageBuilder = null; + try { + replyMessageBuilder = buildReply(AsyncAmqpOutboundGateway.this.messageConverter, message); + sendOutputs(replyMessageBuilder, this.requestMessage); + } + catch (Exception ex) { + Exception exceptionToLogAndSend = ex; + if (!(ex instanceof MessagingException)) { // NOSONAR + exceptionToLogAndSend = new MessageHandlingException(this.requestMessage, + "failed to handle a message in the [" + AsyncAmqpOutboundGateway.this + ']', ex); + if (replyMessageBuilder != null) { + exceptionToLogAndSend = + new MessagingException(replyMessageBuilder.build(), exceptionToLogAndSend); + } } + logger.error(exceptionToLogAndSend, () -> "Failed to send async reply: " + message.toString()); + sendErrorMessage(this.requestMessage, exceptionToLogAndSend); } - logger.error(exceptionToLogAndSend, () -> "Failed to send async reply: " + result.toString()); - sendErrorMessage(this.requestMessage, exceptionToLogAndSend); - } - } - - @Override - public void onFailure(Throwable ex) { - Throwable exceptionToSend = ex; - if (ex instanceof AmqpReplyTimeoutException) { - if (getRequiresReply()) { - exceptionToSend = - new ReplyRequiredException(this.requestMessage, "Timeout on async request/reply", ex); - } - else { - logger.debug(() -> "Reply not required and async timeout for " + this.requestMessage); - return; - } - } - if (ex instanceof AmqpMessageReturnedException) { - AmqpMessageReturnedException amre = (AmqpMessageReturnedException) ex; - MessageChannel returnChannel = getReturnChannel(); - if (returnChannel != null) { - Message returnedMessage = buildReturnedMessage( - new ReturnedMessage(amre.getReturnedMessage(), amre.getReplyCode(), amre.getReplyText(), - amre.getExchange(), amre.getRoutingKey()), - AsyncAmqpOutboundGateway.this.messageConverter); - sendOutput(returnedMessage, returnChannel, true); - } - this.correlationData.setReturned(amre.getReturned()); - /* - * Complete the user's future (if present) since the async template will only complete - * once, successfully, or with a failure. - */ - this.correlationData.getFuture().set(new Confirm(true, null)); } else { - sendErrorMessage(this.requestMessage, exceptionToSend); + Throwable exceptionToSend = throwable; + if (throwable instanceof AmqpReplyTimeoutException) { + if (getRequiresReply()) { + exceptionToSend = + new ReplyRequiredException(this.requestMessage, "Timeout on async request/reply", + throwable); + } + else { + logger.debug(() -> "Reply not required and async timeout for " + this.requestMessage); + return; + } + } + if (throwable instanceof AmqpMessageReturnedException amre) { + MessageChannel returnChannel = getReturnChannel(); + if (returnChannel != null) { + Message returnedMessage = buildReturnedMessage( + new ReturnedMessage(amre.getReturnedMessage(), amre.getReplyCode(), amre.getReplyText(), + amre.getExchange(), amre.getRoutingKey()), + AsyncAmqpOutboundGateway.this.messageConverter); + sendOutput(returnedMessage, returnChannel, true); + } + this.correlationData.setReturned(amre.getReturned()); + /* + * Complete the user's future (if present) since the async template will only complete + * once, successfully, or with a failure. + */ + this.correlationData.getFuture().complete(new Confirm(true, null)); + } + else { + sendErrorMessage(this.requestMessage, exceptionToSend); + } } } } - private final class CorrelationCallback implements ListenableFutureCallback { + private final class CorrelationCallback implements BiConsumer { private final CorrelationData correlationData; @@ -179,17 +180,15 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint { } @Override - public void onSuccess(Boolean result) { - try { - handleConfirm(this.correlationData, result, this.replyFuture.getNackCause()); + public void accept(Boolean result, Throwable throwable) { + if (result != null) { + try { + handleConfirm(this.correlationData, result, this.replyFuture.getNackCause()); + } + catch (Exception e) { + logger.error("Failed to send publisher confirm"); + } } - catch (Exception e) { - logger.error("Failed to send publisher confirm"); - } - } - - @Override - public void onFailure(Throwable ex) { } } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java index e41decb14a..c3e4eef713 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,6 +26,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; @@ -35,7 +36,7 @@ import org.mockito.ArgumentMatchers; import org.springframework.amqp.core.AmqpReplyTimeoutException; import org.springframework.amqp.rabbit.AsyncRabbitTemplate; -import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture; +import org.springframework.amqp.rabbit.RabbitMessageFuture; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.core.RabbitTemplate; @@ -58,7 +59,6 @@ import org.springframework.integration.test.condition.LogLevels; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ErrorMessage; -import org.springframework.util.concurrent.SettableListenableFuture; /** * @author Gary Russell @@ -198,15 +198,13 @@ class AsyncAmqpGatewayTests { ackChannel.receive(10000); ackChannel.purge(null); + RabbitMessageFuture future = mock(RabbitMessageFuture.class); + willReturn("nacknack").given(future).getNackCause(); + willReturn(CompletableFuture.completedFuture(false)).given(future).getConfirm(); + asyncTemplate = mock(AsyncRabbitTemplate.class); - RabbitMessageFuture future = asyncTemplate.new RabbitMessageFuture(null, null); willReturn(future).given(asyncTemplate).sendAndReceive(anyString(), anyString(), any(org.springframework.amqp.core.Message.class)); - DirectFieldAccessor dfa = new DirectFieldAccessor(future); - dfa.setPropertyValue("nackCause", "nacknack"); - SettableListenableFuture confirmFuture = new SettableListenableFuture<>(); - confirmFuture.set(false); - dfa.setPropertyValue("confirm", confirmFuture); new DirectFieldAccessor(gateway).setPropertyValue("template", asyncTemplate); message = MessageBuilder.withPayload("buz").setErrorChannel(errorChannel).build(); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java index 038d50d43a..bf2800c2ad 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,6 +23,7 @@ import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.BDDMockito.willAnswer; import static org.mockito.BDDMockito.willDoNothing; +import static org.mockito.BDDMockito.willReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; @@ -35,6 +36,7 @@ import org.mockito.ArgumentCaptor; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.AsyncRabbitTemplate; +import org.springframework.amqp.rabbit.RabbitMessageFuture; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.core.RabbitTemplate; @@ -99,7 +101,7 @@ public class OutboundEndpointTests { new SimpleMessageListenerContainer(connectionFactory), "replyTo")); amqpTemplate.setTaskScheduler(mock(TaskScheduler.class)); AsyncAmqpOutboundGateway gateway = new AsyncAmqpOutboundGateway(amqpTemplate); - willAnswer(invocation -> amqpTemplate.new RabbitMessageFuture("foo", invocation.getArgument(2))) + willReturn(mock(RabbitMessageFuture.class)) .given(amqpTemplate) .sendAndReceive(anyString(), anyString(), any(Message.class)); gateway.setExchangeName("foo"); @@ -121,8 +123,7 @@ public class OutboundEndpointTests { RabbitTemplate amqpTemplate = spy(new RabbitTemplate(connectionFactory)); AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate); endpoint.setHeadersMappedLast(true); - final AtomicReference amqpMessage = - new AtomicReference(); + final AtomicReference amqpMessage = new AtomicReference<>(); willAnswer(invocation -> { amqpMessage.set(invocation.getArgument(2)); return null; @@ -146,8 +147,7 @@ public class OutboundEndpointTests { DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper(); mapper.setRequestHeaderNames("*"); endpoint.setHeaderMapper(mapper); - final AtomicReference amqpMessage = - new AtomicReference(); + final AtomicReference amqpMessage = new AtomicReference<>(); willAnswer(invocation -> { amqpMessage.set(invocation.getArgument(2)); return null;