From 5572c2161d97ffe830ce9743c1cb1cd95bb5782c Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 28 Jul 2022 09:32:01 -0400 Subject: [PATCH] Fix deprecations around ListenableFuture (#3865) * Fix deprecations around ListenableFuture SF has deprecated a `ListenableFuture` and API around it * Migrate to `CompletableFuture` everywhere a `ListenableFuture` has been used * Suppress a deprecation for `ListenableFuture` keeping the functionality until the next version * Resolve deprecations nad removals from the latest Spring for Apache Kafka * Fix documentation for the `ListenableFuture` in favor of `CompletableFuture` NOTE: the AMQP module is left as is until `ListenableFuture` deprecation is resolved in Spring AMQP * * Restore some `ListenableFuture` test for messaging gateway --- .../gateway/GatewayProxyFactoryBean.java | 27 ++----- .../AbstractMessageProducingHandler.java | 15 ++-- .../AnnotatedEndpointActivationTests.java | 29 +++---- .../config/xml/GatewayParserTests.java | 10 +-- .../gateway/AsyncGatewayTests.java | 24 ++---- .../gateway/GatewayInterfaceTests.java | 38 ++++----- .../handler/AsyncHandlerTests.java | 17 ++-- ...eActivatorDefaultFrameworkMethodTests.java | 17 ++-- .../ip/tcp/TcpOutboundGateway.java | 10 +-- .../integration/jms/JmsOutboundGateway.java | 20 ++--- .../KafkaMessageListenerContainerSpec.java | 10 +-- .../KafkaProducerMessageHandlerTests.java | 25 +++--- .../redis/util/RedisLockRegistry.java | 8 +- .../stomp/AbstractStompSessionManager.java | 80 +++++++++---------- .../ReactorNettyTcpStompSessionManager.java | 9 ++- .../stomp/WebSocketStompSessionManager.java | 8 +- .../stomp/StompSessionManagerTests.java | 15 ++-- .../websocket/ClientWebSocketContainer.java | 29 +++---- .../ClientWebSocketContainerTests.java | 10 +-- .../SimpleWebServiceOutboundGatewayTests.java | 12 +-- src/reference/asciidoc/gateway.adoc | 33 ++------ src/reference/asciidoc/service-activator.adoc | 4 +- src/reference/asciidoc/whats-new.adoc | 5 ++ 23 files changed, 196 insertions(+), 259 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 2efdaeae94..9631d56c1b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,7 +26,6 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.Callable; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executor; import java.util.concurrent.Future; @@ -49,7 +48,6 @@ import org.springframework.beans.factory.BeanInitializationException; import org.springframework.beans.factory.FactoryBean; import org.springframework.core.MethodParameter; import org.springframework.core.ResolvableType; -import org.springframework.core.task.AsyncListenableTaskExecutor; import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.support.TaskExecutorAdapter; @@ -145,10 +143,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private boolean asyncExecutorExplicitlySet; - private Class asyncSubmitType; - - private Class asyncSubmitListenableType; - private volatile boolean initialized; private Map methodMetadataMap; @@ -468,15 +462,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint new ProxyFactory(this.serviceInterface, this); gatewayProxyFactory.addAdvice(new DefaultMethodInvokingMethodInterceptor()); this.serviceProxy = gatewayProxyFactory.getProxy(this.beanClassLoader); - if (this.asyncExecutor != null) { - Callable task = () -> null; - Future submitType = this.asyncExecutor.submit(task); - this.asyncSubmitType = submitType.getClass(); - if (this.asyncExecutor instanceof AsyncListenableTaskExecutor) { - submitType = ((AsyncListenableTaskExecutor) this.asyncExecutor).submitListenable(task); - this.asyncSubmitListenableType = submitType.getClass(); - } - } this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(beanFactory); this.initialized = true; } @@ -511,6 +496,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint @Override @Nullable + @SuppressWarnings("deprecation") public Object invoke(final MethodInvocation invocation) throws Throwable { // NOSONAR final Class returnType; MethodInvocationGateway gateway = this.gatewayMap.get(invocation.getMethod()); @@ -522,15 +508,16 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } if (this.asyncExecutor != null && !Object.class.equals(returnType)) { Invoker invoker = new Invoker(invocation); - if (returnType.isAssignableFrom(this.asyncSubmitType)) { + if (Future.class.equals(returnType)) { return this.asyncExecutor.submit(invoker::get); } - else if (returnType.isAssignableFrom(this.asyncSubmitListenableType)) { - return ((AsyncListenableTaskExecutor) this.asyncExecutor).submitListenable(invoker::get); - } else if (CompletableFuture.class.equals(returnType)) { // exact return CompletableFuture.supplyAsync(invoker, this.asyncExecutor); } + else if (org.springframework.util.concurrent.ListenableFuture.class.equals(returnType)) { + return ((org.springframework.core.task.AsyncListenableTaskExecutor) this.asyncExecutor) + .submitListenable(invoker::get); + } else if (Future.class.isAssignableFrom(returnType)) { logger.debug(() -> "AsyncTaskExecutor submit*() return types are incompatible with the method return " + "type; running on calling thread; the downstream flow must return the required Future: " diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index af2416642d..4a2989e1a6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -51,7 +51,6 @@ import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; -import org.springframework.util.concurrent.ListenableFuture; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -106,9 +105,9 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } /** - * Allow async replies. If the handler reply is a {@link ListenableFuture}, send - * the output when it is satisfied rather than sending the future as the result. - * Ignored for return types other than {@link ListenableFuture}. + * Allow async replies. If the handler reply is a {@link CompletableFuture} or {@link Publisher}, + * send the output when it is satisfied rather than sending the future as the result. + * Ignored for return types other than {@link CompletableFuture} or {@link Publisher}. * @param async true to allow. * @since 4.3 */ @@ -299,6 +298,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan return replyChannel; } + @SuppressWarnings("deprecation") private void doProduceOutput(Message requestMessage, MessageHeaders requestHeaders, Object reply, @Nullable Object replyChannelArg) { @@ -307,7 +307,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan replyChannel = getOutputChannel(); } - if (this.async && (reply instanceof ListenableFuture + if (this.async && (reply instanceof org.springframework.util.concurrent.ListenableFuture || reply instanceof CompletableFuture || reply instanceof Publisher)) { @@ -351,13 +351,14 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan return builder; } + @SuppressWarnings("deprecation") private void asyncNonReactiveReply(Message requestMessage, Object reply, @Nullable Object replyChannel) { CompletableFuture future; if (reply instanceof CompletableFuture) { future = (CompletableFuture) reply; } - else if (reply instanceof ListenableFuture) { - future = ((ListenableFuture) reply).completable(); + else if (reply instanceof org.springframework.util.concurrent.ListenableFuture) { + future = ((org.springframework.util.concurrent.ListenableFuture) reply).completable(); } else { Mono reactiveReply; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java index 25b3542246..b020333289 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,10 +17,12 @@ package org.springframework.integration.config.annotation; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; +import java.util.concurrent.CompletableFuture; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -33,9 +35,7 @@ import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SettableListenableFuture; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Dave Syer @@ -44,7 +44,7 @@ import org.springframework.util.concurrent.SettableListenableFuture; * @author Artem Bilan * @author Yilin Wei */ -@RunWith(SpringRunner.class) +@SpringJUnitConfig @DirtiesContext public class AnnotatedEndpointActivationTests { @@ -72,7 +72,7 @@ public class AnnotatedEndpointActivationTests { // them will get the message. private static volatile int count = 0; - @Before + @BeforeEach public void resetCount() { count = 0; } @@ -108,11 +108,12 @@ public class AnnotatedEndpointActivationTests { assertThat(count).isEqualTo(1); } - @Test(expected = MessageDeliveryException.class) + @Test @DirtiesContext public void stopContext() { applicationContext.stop(); - this.input.send(new GenericMessage<>("foo")); + assertThatExceptionOfType(MessageDeliveryException.class) + .isThrownBy(() -> this.input.send(new GenericMessage<>("foo"))); } @Test @@ -159,9 +160,9 @@ public class AnnotatedEndpointActivationTests { private static class AnnotatedEndpoint3 { @ServiceActivator(inputChannel = "inputAsync", outputChannel = "outputAsync", async = "true") - public ListenableFuture process(String message) { - SettableListenableFuture future = new SettableListenableFuture<>(); - future.set(message); + public CompletableFuture process(String message) { + CompletableFuture future = new CompletableFuture<>(); + future.complete(message); return future; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java index ae29dbd04f..c3dfae34ca 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -59,7 +59,6 @@ import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.GenericMessage; -import org.springframework.scheduling.annotation.AsyncResult; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @@ -403,7 +402,7 @@ public class GatewayParserTests { .setCorrelationId(request.getHeaders().getId()).build(); Object payload = null; if (request.getPayload().equals("futureSync")) { - payload = new AsyncResult>(reply); + payload = CompletableFuture.completedFuture(reply); } else if (request.getPayload().equals("flowCompletable")) { payload = CompletableFuture.completedFuture("SYNC_COMPLETABLE"); @@ -448,7 +447,7 @@ public class GatewayParserTests { } @Override - @SuppressWarnings({ "rawtypes", "unchecked" }) + @SuppressWarnings("unchecked") public Future submit(Callable task) { try { Future result = super.submit(task); @@ -462,7 +461,8 @@ public class GatewayParserTests { modifiedMessage = MessageBuilder.fromMessage(message) .setHeader("executor", this.beanName).build(); } - return new AsyncResult(modifiedMessage); + + return (Future) CompletableFuture.completedFuture(modifiedMessage); } catch (Exception e) { throw new IllegalStateException("unexpected exception in testExecutor", e); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index 800e9354b0..d93dfb9c8b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.mockito.Mockito.mock; import java.time.Duration; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @@ -39,8 +40,6 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.ListenableFutureCallback; import reactor.core.publisher.Mono; @@ -105,22 +104,15 @@ public class AsyncGatewayTests { proxyFactory.setBeanFactory(mock(BeanFactory.class)); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - ListenableFuture> f = service.returnMessageListenable("foo"); + CompletableFuture> f = service.returnMessageListenable("foo"); long start = System.currentTimeMillis(); final AtomicReference> result = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(1); - f.addCallback(new ListenableFutureCallback>() { - - @Override - public void onSuccess(Message msg) { - result.set(msg); + f.whenComplete((message, throwable) -> { + if (throwable == null) { + result.set(message); latch.countDown(); } - - @Override - public void onFailure(Throwable t) { - } - }); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); long elapsed = System.currentTimeMillis() - start; @@ -300,7 +292,7 @@ public class AsyncGatewayTests { String header = (String) input.getHeaders().get("method"); if (header != null && header.startsWith("returnCustomFuture")) { reply = MessageBuilder.withPayload(new CustomFuture(payload, - (Thread) input.getHeaders().get("thread"))) + (Thread) input.getHeaders().get("thread"))) .copyHeaders(input.getHeaders()) .build(); } @@ -317,7 +309,7 @@ public class AsyncGatewayTests { Future returnSomething(String s); - ListenableFuture> returnMessageListenable(String s); + CompletableFuture> returnMessageListenable(String s); @Gateway(headers = @GatewayHeader(name = "method", expression = "#gatewayMethod.name")) CustomFuture returnCustomFuture(String s); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java index 258e97dba7..7f32068594 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -30,11 +30,11 @@ import java.lang.reflect.Method; import java.util.Collection; import java.util.Collections; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; @@ -81,13 +81,10 @@ import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.MessageHeaderAccessor; -import org.springframework.scheduling.annotation.AsyncResult; import org.springframework.stereotype.Component; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ActiveProfiles; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.ListenableFutureCallback; /** * @author Oleg Zhurakousky @@ -399,6 +396,7 @@ public class GatewayInterfaceTests { * performed the send() on gatewayThreadChannel. */ @Test + @SuppressWarnings("deprecation") public void testExecs() throws Exception { assertThat(TestUtils.getPropertyValue(execGatewayFB, "asyncExecutor")).isSameAs(exec); assertThat(TestUtils.getPropertyValue(noExecGatewayFB, "asyncExecutor")).isNull(); @@ -414,25 +412,23 @@ public class GatewayInterfaceTests { result = this.noExecGateway.test1(Thread.currentThread()); assertThat(result.get()).isEqualTo(Thread.currentThread()); - ListenableFuture result2 = this.execGateway.test2(Thread.currentThread()); + CompletableFuture result2 = this.execGateway.test2(Thread.currentThread()); final CountDownLatch latch = new CountDownLatch(1); - final AtomicReference thread = new AtomicReference<>(); - result2.addCallback(new ListenableFutureCallback<>() { - - @Override - public void onSuccess(Thread result) { - thread.set(result); + result2.whenComplete((currentThread, throwable) -> { + if (throwable == null) { latch.countDown(); } - - @Override - public void onFailure(Throwable t) { - } - }); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(result2.get().getName()).startsWith("exec-"); + org.springframework.util.concurrent.ListenableFuture result3 = + this.execGateway.test3(Thread.currentThread()); + final CountDownLatch latch1 = new CountDownLatch(1); + result3.addCallback(data -> latch1.countDown(), ex -> { }); + assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(result3.get().getName()).startsWith("exec-"); + /* @IntegrationComponentScan(useDefaultFilters = false, includeFilters = @ComponentScan.Filter(TestMessagingGateway.class)) @@ -615,7 +611,7 @@ public class GatewayInterfaceTests { Object payload; if (Thread.currentThread().equals(message.getPayload())) { // running on calling thread - need to return a Future. - payload = new AsyncResult<>(Thread.currentThread()); + payload = CompletableFuture.completedFuture(Thread.currentThread()); } else { payload = Thread.currentThread(); @@ -683,7 +679,11 @@ public class GatewayInterfaceTests { Future test1(Thread caller); @Gateway(requestChannel = "gatewayThreadChannel") - ListenableFuture test2(Thread caller); + CompletableFuture test2(Thread caller); + + @Gateway(requestChannel = "gatewayThreadChannel") + @SuppressWarnings("deprecation") + org.springframework.util.concurrent.ListenableFuture test3(Thread caller); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java index 89742dcb0f..697e1302f2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,6 +22,7 @@ import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -46,7 +47,6 @@ import org.springframework.messaging.MessagingException; import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.util.concurrent.SettableListenableFuture; /** * @author Gary Russell @@ -80,19 +80,14 @@ public class AsyncHandlerTests { @Override protected Object handleRequestMessage(Message requestMessage) { - final SettableListenableFuture future = new SettableListenableFuture<>(); + CompletableFuture future = new CompletableFuture<>(); AsyncHandlerTests.this.executor.execute(() -> { try { latch.await(10, TimeUnit.SECONDS); switch (whichTest) { - case 0: - future.set("reply"); - break; - case 1: - future.setException(new RuntimeException("foo")); - break; - case 2: - future.setException(new MessagingException(requestMessage)); + case 0 -> future.complete("reply"); + case 1 -> future.completeExceptionally(new RuntimeException("foo")); + case 2 -> future.completeExceptionally(new MessagingException(requestMessage)); } } catch (InterruptedException e) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java index 9489e89cdf..e6988fcb76 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java @@ -19,6 +19,7 @@ package org.springframework.integration.handler; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; @@ -43,8 +44,6 @@ import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ErrorMessage; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SettableListenableFuture; /** * See INT-1688 for background. @@ -210,7 +209,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { this.asyncIn.send(message); Message reply = replyChannel.receive(0); assertThat(reply).isNull(); - this.asyncService.future.set(this.asyncService.payload.toUpperCase()); + this.asyncService.future.complete(this.asyncService.payload.toUpperCase()); reply = replyChannel.receive(0); assertThat(reply).isNotNull(); assertThat(reply.getPayload()).isEqualTo("TEST"); @@ -225,7 +224,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { Message message = MessageBuilder.withPayload("testing").setReplyChannel(replyChannel).build(); this.asyncIn.send(message); assertThat(reply.get()).isNull(); - this.asyncService.future.set(this.asyncService.payload.toUpperCase()); + this.asyncService.future.complete(this.asyncService.payload.toUpperCase()); assertThat(reply.get()).isNotNull(); assertThat(reply.get().getPayload()).isEqualTo("TESTING"); } @@ -235,7 +234,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { QueueChannel errorChannel = new QueueChannel(); Message message = MessageBuilder.withPayload("test").setErrorChannel(errorChannel).build(); this.asyncIn.send(message); - this.asyncService.future.setException(new RuntimeException("intended")); + this.asyncService.future.completeExceptionally(new RuntimeException("intended")); Message error = errorChannel.receive(0); assertThat(error).isNotNull(); assertThat(error).isInstanceOf(ErrorMessage.class); @@ -249,7 +248,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { public void testAsyncErrorNoHeader() { Message message = MessageBuilder.withPayload("test").build(); this.asyncIn.send(message); - this.asyncService.future.setException(new RuntimeException("intended")); + this.asyncService.future.completeExceptionally(new RuntimeException("intended")); Message error = this.errorChannel.receive(0); assertThat(error).isNotNull(); assertThat(error).isInstanceOf(ErrorMessage.class); @@ -326,13 +325,13 @@ public class ServiceActivatorDefaultFrameworkMethodTests { private static class AsyncService { - private volatile SettableListenableFuture future; + private volatile CompletableFuture future; private volatile String payload; @SuppressWarnings("unused") - public ListenableFuture process(String payload) { - this.future = new SettableListenableFuture<>(); + public CompletableFuture process(String payload) { + this.future = new CompletableFuture<>(); this.payload = payload; return this.future; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index cccd670015..13c14c1449 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; import java.time.Instant; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Semaphore; @@ -48,7 +49,6 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; -import org.springframework.util.concurrent.SettableListenableFuture; /** * TCP outbound gateway that uses a client connection factory. If the factory is configured @@ -342,7 +342,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler } } if (isAsync()) { - reply.getFuture().set(message); + reply.getFuture().complete(message); cleanUp(reply.isHaveSemaphore(), reply.getConnection(), connectionId); } else { @@ -427,7 +427,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler private final boolean haveSemaphore; - private final SettableListenableFuture> future = new SettableListenableFuture<>(); + private final CompletableFuture> future = new CompletableFuture<>(); private volatile Message reply; @@ -443,7 +443,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler getTaskScheduler() .schedule(() -> { TcpOutboundGateway.this.pendingReplies.remove(connection.getConnectionId()); - this.future.setException( + this.future.completeExceptionally( new MessageTimeoutException(requestMessage, "Timed out waiting for response")); }, Instant.now().plusMillis(remoteTimeout)); } @@ -495,7 +495,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler return this.reply; } - SettableListenableFuture> getFuture() { + CompletableFuture> getFuture() { return this.future; } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 94012aaf7c..b17769b7f3 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -22,6 +22,7 @@ import java.util.Iterator; import java.util.Map; import java.util.Map.Entry; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executor; import java.util.concurrent.LinkedBlockingQueue; @@ -71,7 +72,6 @@ import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; -import org.springframework.util.concurrent.SettableListenableFuture; /** * An outbound Messaging Gateway for request/reply JMS. @@ -101,7 +101,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler private final ConcurrentHashMap earlyOrLateReplies = new ConcurrentHashMap<>(); - private final Map>> futures = + private final Map>> futures = new ConcurrentHashMap<>(); private final Object lifeCycleMonitor = new Object(); @@ -1075,7 +1075,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler LinkedBlockingQueue replyQueue = null; String correlationToLog = correlation; logger.debug(() -> getComponentName() + " Sending message with correlationId " + correlationToLog); - SettableListenableFuture> future = null; + CompletableFuture> future = null; boolean async = isAsync(); if (!async) { replyQueue = new LinkedBlockingQueue<>(1); @@ -1112,7 +1112,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler messageProducer = session.createProducer(reqDestination); LinkedBlockingQueue replyQueue = new LinkedBlockingQueue<>(1); - this.sendRequestMessage(jmsRequest, messageProducer, priority); + sendRequestMessage(jmsRequest, messageProducer, priority); correlation = jmsRequest.getJMSMessageID(); String correlationToLog = correlation; @@ -1169,8 +1169,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler return reply; } - private SettableListenableFuture> createFuture(final String correlationId) { - SettableListenableFuture> future = new SettableListenableFuture<>(); + private CompletableFuture> createFuture(final String correlationId) { + CompletableFuture> future = new CompletableFuture<>(); this.futures.put(correlationId, future); if (this.receiveTimeout > 0) { getTaskScheduler().schedule(() -> expire(correlationId), Instant.now().plusMillis(this.receiveTimeout)); @@ -1179,11 +1179,11 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler } private void expire(String correlationId) { - SettableListenableFuture> future = this.futures.remove(correlationId); + CompletableFuture> future = this.futures.remove(correlationId); if (future != null) { try { if (getRequiresReply()) { - future.setException(new JmsTimeoutException("No reply in " + this.receiveTimeout + " ms")); + future.completeExceptionally(new JmsTimeoutException("No reply in " + this.receiveTimeout + " ms")); } else { logger.debug(() -> "Reply expired and reply not required for " + correlationId); @@ -1276,10 +1276,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler } private void onMessageAsync(jakarta.jms.Message message, String correlationId) throws JMSException { - SettableListenableFuture> future = this.futures.remove(correlationId); + CompletableFuture> future = this.futures.remove(correlationId); if (future != null) { message.setJMSCorrelationID(null); - future.set(buildReply(message)); + future.complete(buildReply(message)); } else { logger.warn(() -> "Late reply for " + correlationId); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java index 34b98d79c3..4d839d02f3 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java @@ -21,7 +21,7 @@ import java.util.regex.Pattern; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.OffsetCommitCallback; -import org.springframework.core.task.AsyncListenableTaskExecutor; +import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.integration.dsl.IntegrationComponentSpec; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.listener.CommonErrorHandler; @@ -157,12 +157,10 @@ public class KafkaMessageListenerContainerSpec * Set the executor for threads that poll the consumer. * @param consumerTaskExecutor the executor * @return the spec. - * @see ContainerProperties#setConsumerTaskExecutor(AsyncListenableTaskExecutor) + * @see ContainerProperties#setListenerTaskExecutor(AsyncTaskExecutor) */ - public KafkaMessageListenerContainerSpec consumerTaskExecutor( - AsyncListenableTaskExecutor consumerTaskExecutor) { - - this.target.getContainerProperties().setConsumerTaskExecutor(consumerTaskExecutor); + public KafkaMessageListenerContainerSpec listenerTaskExecutor(AsyncTaskExecutor consumerTaskExecutor) { + this.target.getContainerProperties().setListenerTaskExecutor(consumerTaskExecutor); return this; } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index 75caaa55ce..9521c02c6d 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -41,6 +41,7 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -105,8 +106,6 @@ import org.springframework.transaction.TransactionException; import org.springframework.transaction.support.AbstractPlatformTransactionManager; import org.springframework.transaction.support.DefaultTransactionStatus; import org.springframework.transaction.support.TransactionTemplate; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SettableListenableFuture; /** * @author Gary Russell @@ -474,8 +473,7 @@ class KafkaProducerMessageHandlerTests { given(pf.transactionCapable()).willReturn(true); Producer producer = mock(Producer.class); given(pf.createProducer(isNull())).willReturn(producer); - ListenableFuture future = mock(ListenableFuture.class); - willReturn(future).given(producer).send(any(ProducerRecord.class), any(Callback.class)); + willReturn(mock(Future.class)).given(producer).send(any(ProducerRecord.class), any(Callback.class)); KafkaTemplate template = new KafkaTemplate(pf); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(template); handler.setTopicExpression(new LiteralExpression("bar")); @@ -517,7 +515,7 @@ class KafkaProducerMessageHandlerTests { ConsumerFactory cf = mock(ConsumerFactory.class); willReturn(mockConsumer).given(cf).createConsumer("group", "", null, KafkaTestUtils.defaultPropertyOverrides()); Producer producer = mock(Producer.class); - given(producer.send(any(), any())).willReturn(new SettableListenableFuture<>()); + given(producer.send(any(), any())).willReturn(mock(Future.class)); final CountDownLatch closeLatch = new CountDownLatch(2); willAnswer(i -> { closeLatch.countDown(); @@ -579,8 +577,7 @@ class KafkaProducerMessageHandlerTests { }; pf.setTransactionIdPrefix("default.tx.id."); - ListenableFuture future = mock(ListenableFuture.class); - willReturn(future).given(producer).send(any(ProducerRecord.class), any(Callback.class)); + willReturn(mock(Future.class)).given(producer).send(any(ProducerRecord.class), any(Callback.class)); KafkaTemplate template = new KafkaTemplate(pf); template.setTransactionIdPrefix("overridden.tx.id."); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(template); @@ -604,8 +601,7 @@ class KafkaProducerMessageHandlerTests { ProducerFactory pf = mock(ProducerFactory.class); given(pf.transactionCapable()).willReturn(true); given(pf.createProducer(isNull())).willReturn(producer); - ListenableFuture future = mock(ListenableFuture.class); - willReturn(future).given(producer).send(any(ProducerRecord.class), any(Callback.class)); + willReturn(mock(Future.class)).given(producer).send(any(ProducerRecord.class), any(Callback.class)); KafkaTemplate template = new KafkaTemplate(pf); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(template); handler.setTopicExpression(new LiteralExpression("bar")); @@ -655,7 +651,7 @@ class KafkaProducerMessageHandlerTests { ConsumerFactory cf = mock(ConsumerFactory.class); willReturn(mockConsumer).given(cf).createConsumer("group", "", null, KafkaTestUtils.defaultPropertyOverrides()); Producer producer = mock(Producer.class); - given(producer.send(any(), any())).willReturn(new SettableListenableFuture<>()); + given(producer.send(any(), any())).willReturn(mock(Future.class)); final CountDownLatch closeLatch = new CountDownLatch(2); willAnswer(i -> { closeLatch.countDown(); @@ -732,8 +728,7 @@ class KafkaProducerMessageHandlerTests { ProducerFactory pf = mock(ProducerFactory.class); Producer producer = mock(Producer.class); given(pf.createProducer()).willReturn(producer); - ListenableFuture future = mock(ListenableFuture.class); - willReturn(future).given(producer).send(any(ProducerRecord.class), any(Callback.class)); + willReturn(mock(Future.class)).given(producer).send(any(ProducerRecord.class), any(Callback.class)); KafkaTemplate template = new KafkaTemplate(pf); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(template); handler.setTopicExpression(new LiteralExpression("bar")); @@ -756,8 +751,7 @@ class KafkaProducerMessageHandlerTests { ProducerFactory pf = mock(ProducerFactory.class); Producer producer = mock(Producer.class); given(pf.createProducer()).willReturn(producer); - ListenableFuture future = mock(ListenableFuture.class); - willReturn(future).given(producer).send(any(ProducerRecord.class), any(Callback.class)); + willReturn(mock(Future.class)).given(producer).send(any(ProducerRecord.class), any(Callback.class)); KafkaTemplate template = new KafkaTemplate(pf); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(template); handler.setTopicExpression(new LiteralExpression("bar")); @@ -777,8 +771,7 @@ class KafkaProducerMessageHandlerTests { ProducerFactory pf = mock(ProducerFactory.class); Producer producer = mock(Producer.class); given(pf.createProducer()).willReturn(producer); - ListenableFuture future = mock(ListenableFuture.class); - willReturn(future).given(producer).send(any(ProducerRecord.class), any(Callback.class)); + willReturn(mock(Future.class)).given(producer).send(any(ProducerRecord.class), any(Callback.class)); KafkaTemplate template = new KafkaTemplate(pf); RecordMessageConverter converter = mock(RecordMessageConverter.class); ProducerRecord recordFromConverter = mock(ProducerRecord.class); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java index a2b4ff95b1..7ef33da691 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java @@ -23,6 +23,7 @@ import java.util.LinkedHashMap; import java.util.Map; import java.util.Map.Entry; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; @@ -54,7 +55,6 @@ import org.springframework.integration.support.locks.ExpirableLockRegistry; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; -import org.springframework.util.concurrent.SettableListenableFuture; /** * Implementation of {@link ExpirableLockRegistry} providing a distributed lock using Redis. @@ -643,7 +643,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl private static final class RedisUnLockNotifyMessageListener implements MessageListener { - private final Map> notifyMap = new ConcurrentHashMap<>(); + private final Map> notifyMap = new ConcurrentHashMap<>(); @Override public void onMessage(Message message, byte[] pattern) { @@ -652,7 +652,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl } public Future subscribeLock(String lockKey) { - return this.notifyMap.computeIfAbsent(lockKey, key -> new SettableListenableFuture<>()); + return this.notifyMap.computeIfAbsent(lockKey, key -> new CompletableFuture<>()); } public void unSubscribeLock(String localLock) { @@ -661,7 +661,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl private void unlockNotify(String lockKey) { this.notifyMap.computeIfPresent(lockKey, (key, lockFuture) -> { - lockFuture.set(key); + lockFuture.complete(key); return lockFuture; }); } diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java index b6d66ac3fa..bacd4982a7 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java @@ -20,10 +20,12 @@ import java.time.Instant; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -46,7 +48,6 @@ import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.ListenableFutureCallback; /** * Base {@link StompSessionManager} implementation to manage a single {@link StompSession} @@ -103,7 +104,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager private volatile boolean connected; - private volatile ListenableFuture stompSessionListenableFuture; + private volatile CompletableFuture stompSessionFuture; private volatile ScheduledFuture reconnectFuture; @@ -187,7 +188,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager this.logger.debug("Connecting " + this); } try { - this.stompSessionListenableFuture = doConnect(this.compositeStompSessionHandler); + this.stompSessionFuture = doConnect(this.compositeStompSessionHandler); } catch (Exception e) { if (currentEpoch == this.epoch.get()) { @@ -216,29 +217,30 @@ public abstract class AbstractStompSessionManager implements StompSessionManager private CountDownLatch addStompSessionCallback(int currentEpoch) { CountDownLatch connectLatch = new CountDownLatch(1); - this.stompSessionListenableFuture.addCallback( - stompSession -> { - AbstractStompSessionManager.this.logger.debug("onSuccess"); - AbstractStompSessionManager.this.connected = true; - AbstractStompSessionManager.this.connecting = false; - if (stompSession != null) { - stompSession.setAutoReceipt(isAutoReceiptEnabled()); + this.stompSessionFuture.whenComplete((stompSession, throwable) -> { + if (throwable == null) { + AbstractStompSessionManager.this.logger.debug("onSuccess"); + AbstractStompSessionManager.this.connected = true; + AbstractStompSessionManager.this.connecting = false; + if (stompSession != null) { + stompSession.setAutoReceipt(isAutoReceiptEnabled()); + } + if (AbstractStompSessionManager.this.applicationEventPublisher != null) { + AbstractStompSessionManager.this.applicationEventPublisher.publishEvent( + new StompSessionConnectedEvent(this)); + } + AbstractStompSessionManager.this.reconnectFuture = null; + connectLatch.countDown(); } - if (AbstractStompSessionManager.this.applicationEventPublisher != null) { - AbstractStompSessionManager.this.applicationEventPublisher.publishEvent( - new StompSessionConnectedEvent(this)); + else { + AbstractStompSessionManager.this.logger.debug("onFailure", throwable); + connectLatch.countDown(); + if (currentEpoch == AbstractStompSessionManager.this.epoch.get()) { + scheduleReconnect(throwable); + } } - AbstractStompSessionManager.this.reconnectFuture = null; - connectLatch.countDown(); - - }, - e -> { - AbstractStompSessionManager.this.logger.debug("onFailure", e); - connectLatch.countDown(); - if (currentEpoch == AbstractStompSessionManager.this.epoch.get()) { - scheduleReconnect(e); - } - }); + } + ); return connectLatch; } @@ -271,29 +273,21 @@ public abstract class AbstractStompSessionManager implements StompSessionManager @Override public void destroy() { - if (this.stompSessionListenableFuture != null) { + if (this.stompSessionFuture != null) { if (this.reconnectFuture != null) { this.reconnectFuture.cancel(false); this.reconnectFuture = null; } - this.stompSessionListenableFuture.addCallback( - new ListenableFutureCallback<>() { + this.stompSessionFuture.whenComplete(new BiConsumer() { - @Override - public void onFailure(Throwable ex) { - AbstractStompSessionManager.this.connected = false; - } - - @Override - public void onSuccess(StompSession session) { - if (session != null) { - session.disconnect(); - } - AbstractStompSessionManager.this.connected = false; - } - - }); - this.stompSessionListenableFuture = null; + @Override public void accept(StompSession session, Throwable throwable) { + if (session != null) { + session.disconnect(); + } + AbstractStompSessionManager.this.connected = false; + } + }); + this.stompSessionFuture = null; } } @@ -353,7 +347,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager '}'; } - protected abstract ListenableFuture doConnect(StompSessionHandler handler); + protected abstract CompletableFuture doConnect(StompSessionHandler handler); private class CompositeStompSessionHandler extends StompSessionHandlerAdapter { diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/ReactorNettyTcpStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/ReactorNettyTcpStompSessionManager.java index 8c420a5e28..8aa1b691d0 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/ReactorNettyTcpStompSessionManager.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/ReactorNettyTcpStompSessionManager.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,10 +16,11 @@ package org.springframework.integration.stomp; +import java.util.concurrent.CompletableFuture; + import org.springframework.messaging.simp.stomp.ReactorNettyTcpStompClient; import org.springframework.messaging.simp.stomp.StompSession; import org.springframework.messaging.simp.stomp.StompSessionHandler; -import org.springframework.util.concurrent.ListenableFuture; /** * The {@link ReactorNettyTcpStompClient} based {@link AbstractStompSessionManager} implementation. @@ -37,8 +38,8 @@ public class ReactorNettyTcpStompSessionManager extends AbstractStompSessionMana } @Override - protected ListenableFuture doConnect(StompSessionHandler handler) { - return ((ReactorNettyTcpStompClient) this.stompClient).connect(getConnectHeaders(), handler); + protected CompletableFuture doConnect(StompSessionHandler handler) { + return ((ReactorNettyTcpStompClient) this.stompClient).connectAsync(getConnectHeaders(), handler); } } diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java index 88c709a6a8..38c3f7c4c2 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,11 +17,11 @@ package org.springframework.integration.stomp; import java.util.Arrays; +import java.util.concurrent.CompletableFuture; import org.springframework.messaging.simp.stomp.StompSession; import org.springframework.messaging.simp.stomp.StompSessionHandler; import org.springframework.util.Assert; -import org.springframework.util.concurrent.ListenableFuture; import org.springframework.web.socket.WebSocketHttpHeaders; import org.springframework.web.socket.messaging.WebSocketStompClient; @@ -55,9 +55,9 @@ public class WebSocketStompSessionManager extends AbstractStompSessionManager { } @Override - protected ListenableFuture doConnect(StompSessionHandler handler) { + protected CompletableFuture doConnect(StompSessionHandler handler) { return ((WebSocketStompClient) this.stompClient) - .connect(this.url, this.handshakeHeaders, getConnectHeaders(), handler, this.uriVariables); + .connectAsync(this.url, this.handshakeHeaders, getConnectHeaders(), handler, this.uriVariables); } } diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/StompSessionManagerTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/StompSessionManagerTests.java index 89d3893967..1d19e2d90e 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/StompSessionManagerTests.java +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/StompSessionManagerTests.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. @@ -19,6 +19,7 @@ package org.springframework.integration.stomp; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -30,8 +31,6 @@ import org.springframework.messaging.simp.stomp.StompSession; import org.springframework.messaging.simp.stomp.StompSessionHandler; import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter; import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SettableListenableFuture; /** * @author Artem Bilan @@ -49,14 +48,14 @@ public class StompSessionManagerTests { private final AtomicBoolean thrown = new AtomicBoolean(); @Override - protected ListenableFuture doConnect(StompSessionHandler handler) { + protected CompletableFuture doConnect(StompSessionHandler handler) { if (!this.thrown.getAndSet(true)) { throw new RuntimeException("intentional"); } else { - SettableListenableFuture future = new SettableListenableFuture<>(); + CompletableFuture future = new CompletableFuture<>(); StompSession stompSession = mock(StompSession.class); - future.set(stompSession); + future.complete(stompSession); handler.afterConnected(stompSession, getConnectHeaders()); return future; } @@ -66,12 +65,12 @@ public class StompSessionManagerTests { sessionManager.start(); - final SettableListenableFuture stompSessionFuture = new SettableListenableFuture<>(); + final CompletableFuture stompSessionFuture = new CompletableFuture<>(); sessionManager.connect(new StompSessionHandlerAdapter() { @Override public void afterConnected(StompSession session, StompHeaders connectedHeaders) { - stompSessionFuture.set(session); + stompSessionFuture.complete(session); } }); diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java index 025d5240ef..d8022b49c2 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package org.springframework.integration.websocket; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -25,8 +26,6 @@ import org.springframework.context.SmartLifecycle; import org.springframework.http.HttpHeaders; import org.springframework.util.Assert; import org.springframework.util.StringUtils; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.ListenableFutureCallback; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.WebSocketHttpHeaders; import org.springframework.web.socket.WebSocketSession; @@ -241,26 +240,22 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine logger.info("Connecting to WebSocket at " + getUri()); } ClientWebSocketContainer.this.headers.setSecWebSocketProtocol(getSubProtocols()); - ListenableFuture future = - this.client.doHandshake(ClientWebSocketContainer.this.webSocketHandler, + CompletableFuture future = + this.client.execute(ClientWebSocketContainer.this.webSocketHandler, ClientWebSocketContainer.this.headers, getUri()); - future.addCallback(new ListenableFutureCallback() { - - @Override - public void onSuccess(WebSocketSession session) { + future.whenComplete((session, throwable) -> { + if (throwable == null) { ClientWebSocketContainer.this.clientSession = session; logger.info("Successfully connected"); - ClientWebSocketContainer.this.connectionLatch.countDown(); } - - @Override - public void onFailure(Throwable t) { - logger.error("Failed to connect", t); - ClientWebSocketContainer.this.openConnectionException = t; - ClientWebSocketContainer.this.connectionLatch.countDown(); + else { + Throwable cause = throwable.getCause(); + cause = cause != null ? cause : throwable; + logger.error("Failed to connect", cause); + ClientWebSocketContainer.this.openConnectionException = cause; } - + ClientWebSocketContainer.this.connectionLatch.countDown(); }); } diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java index 33719181c4..64c8300f24 100644 --- a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2021 the original author or authors. + * Copyright 2014-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,6 +26,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -36,7 +37,6 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.http.HttpHeaders; -import org.springframework.util.concurrent.ListenableFuture; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.PingMessage; import org.springframework.web.socket.PongMessage; @@ -72,12 +72,12 @@ public class ClientWebSocketContainerTests { StandardWebSocketClient webSocketClient = new StandardWebSocketClient() { @Override - protected ListenableFuture doHandshakeInternal(WebSocketHandler webSocketHandler, + protected CompletableFuture executeInternal(WebSocketHandler webSocketHandler, HttpHeaders headers, URI uri, List protocols, List extensions, Map attributes) { - ListenableFuture future = - super.doHandshakeInternal(webSocketHandler, headers, uri, protocols, extensions, + CompletableFuture future = + super.executeInternal(webSocketHandler, headers, uri, protocols, extensions, attributes); if (failure.get()) { future.cancel(true); diff --git a/spring-integration-ws/src/test/java/org/springframework/integration/ws/SimpleWebServiceOutboundGatewayTests.java b/spring-integration-ws/src/test/java/org/springframework/integration/ws/SimpleWebServiceOutboundGatewayTests.java index 694654712a..36b5b5120e 100644 --- a/spring-integration-ws/src/test/java/org/springframework/integration/ws/SimpleWebServiceOutboundGatewayTests.java +++ b/spring-integration-ws/src/test/java/org/springframework/integration/ws/SimpleWebServiceOutboundGatewayTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,6 +23,7 @@ import static org.mockito.Mockito.mock; import java.io.ByteArrayInputStream; import java.io.InputStreamReader; import java.net.URI; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -46,7 +47,6 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.FileCopyUtils; -import org.springframework.util.concurrent.SettableListenableFuture; import org.springframework.ws.WebServiceMessage; import org.springframework.ws.WebServiceMessageFactory; import org.springframework.ws.client.WebServiceClientException; @@ -142,13 +142,13 @@ public class SimpleWebServiceOutboundGatewayTests { SimpleWebServiceOutboundGateway gateway = new SimpleWebServiceOutboundGateway(uri); gateway.setBeanFactory(mock(BeanFactory.class)); - final SettableListenableFuture requestFuture = new SettableListenableFuture<>(); + final CompletableFuture requestFuture = new CompletableFuture<>(); ClientInterceptorAdapter interceptorAdapter = new ClientInterceptorAdapter() { @Override public boolean handleRequest(MessageContext messageContext) throws WebServiceClientException { - requestFuture.set(messageContext.getRequest()); + requestFuture.complete(messageContext.getRequest()); return super.handleRequest(messageContext); } @@ -202,13 +202,13 @@ public class SimpleWebServiceOutboundGatewayTests { SimpleWebServiceOutboundGateway gateway = new SimpleWebServiceOutboundGateway(uri); gateway.setBeanFactory(mock(BeanFactory.class)); - final SettableListenableFuture requestFuture = new SettableListenableFuture<>(); + final CompletableFuture requestFuture = new CompletableFuture<>(); ClientInterceptorAdapter interceptorAdapter = new ClientInterceptorAdapter() { @Override public boolean handleRequest(MessageContext messageContext) throws WebServiceClientException { - requestFuture.set(messageContext.getRequest()); + requestFuture.complete(messageContext.getRequest()); return super.handleRequest(messageContext); } diff --git a/src/reference/asciidoc/gateway.adoc b/src/reference/asciidoc/gateway.adoc index 9023a77a6a..5856707d12 100644 --- a/src/reference/asciidoc/gateway.adoc +++ b/src/reference/asciidoc/gateway.adoc @@ -583,39 +583,13 @@ int finalResult = result.get(1000, TimeUnit.SECONDS); For a more detailed example, see the https://github.com/spring-projects/spring-integration-samples/tree/main/intermediate/async-gateway[async-gateway] sample in the Spring Integration samples. -===== `ListenableFuture` - -Starting with version 4.1, asynchronous gateway methods can also return `ListenableFuture` (introduced in Spring Framework 4.0). -These return types let you provide a callback, which is invoked when the result is available (or an exception occurs). -When the gateway detects this return type and the <> is an `AsyncListenableTaskExecutor`, the executor's `submitListenable()` method is invoked. -The following example shows how to use a `ListenableFuture`: - -==== -[source,java] ----- -ListenableFuture result = this.asyncGateway.async("something"); -result.addCallback(new ListenableFutureCallback() { - - @Override - public void onSuccess(String result) { - ... - } - - @Override - public void onFailure(Throwable t) { - ... - } -}); ----- -==== - [[gateway-asynctaskexecutor]] ===== `AsyncTaskExecutor` By default, the `GatewayProxyFactoryBean` uses `org.springframework.core.task.SimpleAsyncTaskExecutor` when submitting internal `AsyncInvocationTask` instances for any gateway method whose return type is a `Future`. However, the `async-executor` attribute in the `` element's configuration lets you provide a reference to any implementation of `java.util.concurrent.Executor` available within the Spring application context. -The (default) `SimpleAsyncTaskExecutor` supports both `Future` and `ListenableFuture` return types, returning `FutureTask` or `ListenableFutureTask` respectively. +The (default) `SimpleAsyncTaskExecutor` supports both `Future` and `CompletableFuture` return types. See <>. Even though there is a default executor, it is often useful to provide an external one so that you can identify its threads in logs (when using XML, the thread name is based on the executor's bean name), as the following example shows: @@ -671,6 +645,9 @@ There are two modes of operation when returning this type: * When the async executor is explicitly set to `null` and the return type is `CompletableFuture` or the return type is a subclass of `CompletableFuture`, the flow is invoked on the caller's thread. In this scenario, the downstream flow is expected to return a `CompletableFuture` of the appropriate type. +NOTE: The `org.springframework.util.concurrent.ListenableFuture` has been deprecated starting with Spring Framework `6.0`. +It is recommended now to migrate to the `CompletableFuture` which provides similar processing functionality. + ====== Usage Scenarios In the following scenario, the caller thread returns immediately with a `CompletableFuture`, which is completed when the downstream flow replies to the gateway (with an `Invoice` object). @@ -802,7 +779,7 @@ The calling thread continues, with `handleInvoice()` being called when the flow ===== Downstream Flows Returning an Asynchronous Type -As mentioned in the `ListenableFuture` section above, if you wish some downstream component to return a message with an async payload (`Future`, `Mono`, and others), you must explicitly set the async executor to `null` (or `""` when using XML configuration). +As mentioned in the <> section above, if you wish some downstream component to return a message with an async payload (`Future`, `Mono`, and others), you must explicitly set the async executor to `null` (or `""` when using XML configuration). The flow is then invoked on the caller thread and the result can be retrieved later. ===== `void` Return Type diff --git a/src/reference/asciidoc/service-activator.adoc b/src/reference/asciidoc/service-activator.adoc index 975a586ec0..551b2d3944 100644 --- a/src/reference/asciidoc/service-activator.adoc +++ b/src/reference/asciidoc/service-activator.adoc @@ -156,9 +156,9 @@ See <<./dsl.adoc#java-dsl-handle,Service Activators and the `.handle()` method>> The service activator is invoked by the calling thread. This is an upstream thread if the input channel is a `SubscribableChannel` or a poller thread for a `PollableChannel`. -If the service returns a `ListenableFuture`, the default action is to send that as the payload of the message sent to the output (or reply) channel. +If the service returns a `CompletableFuture`, the default action is to send that as the payload of the message sent to the output (or reply) channel. Starting with version 4.3, you can now set the `async` attribute to `true` (by using `setAsync(true)` when using Java configuration). -If the service returns a `ListenableFuture` when this the `async` attribute is set to `true`, the calling thread is released immediately and the reply message is sent on the thread (from within your service) that completes the future. +If the service returns a `CompletableFuture` when this the `async` attribute is set to `true`, the calling thread is released immediately and the reply message is sent on the thread (from within your service) that completes the future. This is particularly advantageous for long-running services that use a `PollableChannel`, because the poller thread is released to perform other services within the framework. If the service completes the future with an `Exception`, normal error processing occurs. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 2fb9dd6092..fbf4f9ea24 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -52,6 +52,11 @@ The factory class will be removed in the future releases. See <<./dsl.adoc#java-dsl,Java DSL>> for more information. +The `org.springframework.util.concurrent.ListenableFuture` has been deprecated starting with Spring Framework `6.0`. +All Spring Integration async API has been migrated to the `CompletableFuture`. + +See <<./gateway.adoc#gw-completable-future, CompletableFuture support>> for more information. + [[x6.0-http]] === HTTP Changes