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

Fix deprecations around ListenableFuture (#3865)

Browse Source 
* 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
This commit is contained in:
Artem Bilan
2022-07-28 09:32:01 -04:00
committed by GitHub
parent 2e9beada0b
commit 5572c2161d
23 changed files with 196 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -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<String, GatewayMethodMetadata> 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<String> task = () -> null;
Future<String> 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: "

15
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java
View File

@@ -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;

29
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java
View File

@@ -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<String> process(String message) {
SettableListenableFuture<String> future = new SettableListenableFuture<>();
future.set(message);
public CompletableFuture<String> process(String message) {
CompletableFuture<String> future = new CompletableFuture<>();
future.complete(message);
return future;
}

10
spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java
View File

@@ -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<Message<?>>(reply);
payload = CompletableFuture.completedFuture(reply);
}
else if (request.getPayload().equals("flowCompletable")) {
payload = CompletableFuture.<String>completedFuture("SYNC_COMPLETABLE");
@@ -448,7 +447,7 @@ public class GatewayParserTests {
}
@Override
@SuppressWarnings({ "rawtypes", "unchecked" })
@SuppressWarnings("unchecked")
public <T> Future<T> submit(Callable<T> 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<T>) CompletableFuture.completedFuture(modifiedMessage);
}
catch (Exception e) {
throw new IllegalStateException("unexpected exception in testExecutor", e);

24
spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java
View File

@@ -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<Message<?>> f = service.returnMessageListenable("foo");
CompletableFuture<Message<?>> f = service.returnMessageListenable("foo");
long start = System.currentTimeMillis();
final AtomicReference<Message<?>> result = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);
f.addCallback(new ListenableFutureCallback<Message<?>>() {
@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<Message<?>> returnMessageListenable(String s);
CompletableFuture<Message<?>> returnMessageListenable(String s);
@Gateway(headers = @GatewayHeader(name = "method", expression = "#gatewayMethod.name"))
CustomFuture returnCustomFuture(String s);

38
spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java
View File

@@ -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<Thread> result2 = this.execGateway.test2(Thread.currentThread());
CompletableFuture<Thread> result2 = this.execGateway.test2(Thread.currentThread());
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<Thread> 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<Thread> 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<Thread> test1(Thread caller);
@Gateway(requestChannel = "gatewayThreadChannel")
ListenableFuture<Thread> test2(Thread caller);
CompletableFuture<Thread> test2(Thread caller);
@Gateway(requestChannel = "gatewayThreadChannel")
@SuppressWarnings("deprecation")
org.springframework.util.concurrent.ListenableFuture<Thread> test3(Thread caller);
}

17
spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java
View File

@@ -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<String> future = new SettableListenableFuture<>();
CompletableFuture<String> 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) {

17
spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java
View File

@@ -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<String> future;
private volatile CompletableFuture<String> future;
private volatile String payload;
@SuppressWarnings("unused")
public ListenableFuture<String> process(String payload) {
this.future = new SettableListenableFuture<>();
public CompletableFuture<String> process(String payload) {
this.future = new CompletableFuture<>();
this.payload = payload;
return this.future;
}

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
View File

@@ -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<Message<?>> future = new SettableListenableFuture<>();
private final CompletableFuture<Message<?>> 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<Message<?>> getFuture() {
CompletableFuture<Message<?>> getFuture() {
return this.future;
}

20
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

@@ -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<String, TimedReply> earlyOrLateReplies = new ConcurrentHashMap<>();
private final Map<String, SettableListenableFuture<AbstractIntegrationMessageBuilder<?>>> futures =
private final Map<String, CompletableFuture<AbstractIntegrationMessageBuilder<?>>> futures =
new ConcurrentHashMap<>();
private final Object lifeCycleMonitor = new Object();
@@ -1075,7 +1075,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler
LinkedBlockingQueue<jakarta.jms.Message> replyQueue = null;
String correlationToLog = correlation;
logger.debug(() -> getComponentName() + " Sending message with correlationId " + correlationToLog);
SettableListenableFuture<AbstractIntegrationMessageBuilder<?>> future = null;
CompletableFuture<AbstractIntegrationMessageBuilder<?>> 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<jakarta.jms.Message> 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<AbstractIntegrationMessageBuilder<?>> createFuture(final String correlationId) {
SettableListenableFuture<AbstractIntegrationMessageBuilder<?>> future = new SettableListenableFuture<>();
private CompletableFuture<AbstractIntegrationMessageBuilder<?>> createFuture(final String correlationId) {
CompletableFuture<AbstractIntegrationMessageBuilder<?>> 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<AbstractIntegrationMessageBuilder<?>> future = this.futures.remove(correlationId);
CompletableFuture<AbstractIntegrationMessageBuilder<?>> 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<AbstractIntegrationMessageBuilder<?>> future = this.futures.remove(correlationId);
CompletableFuture<AbstractIntegrationMessageBuilder<?>> 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);

10
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java
View File

@@ -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<K, V>
* 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<K, V> consumerTaskExecutor(
AsyncListenableTaskExecutor consumerTaskExecutor) {
this.target.getContainerProperties().setConsumerTaskExecutor(consumerTaskExecutor);
public KafkaMessageListenerContainerSpec<K, V> listenerTaskExecutor(AsyncTaskExecutor consumerTaskExecutor) {
this.target.getContainerProperties().setListenerTaskExecutor(consumerTaskExecutor);
return this;
}

25
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
View File

@@ -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);

8
spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
View File

@@ -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<String, SettableListenableFuture<String>> notifyMap = new ConcurrentHashMap<>();
private final Map<String, CompletableFuture<String>> notifyMap = new ConcurrentHashMap<>();
@Override
public void onMessage(Message message, byte[] pattern) {
@@ -652,7 +652,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
}
public Future<String> 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;
});
}

80
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java
View File

@@ -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<StompSession> stompSessionListenableFuture;
private volatile CompletableFuture<StompSession> 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<StompSession, Throwable>() {
@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<StompSession> doConnect(StompSessionHandler handler);
protected abstract CompletableFuture<StompSession> doConnect(StompSessionHandler handler);
private class CompositeStompSessionHandler extends StompSessionHandlerAdapter {

9
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/ReactorNettyTcpStompSessionManager.java
View File

@@ -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<StompSession> doConnect(StompSessionHandler handler) {
return ((ReactorNettyTcpStompClient) this.stompClient).connect(getConnectHeaders(), handler);
protected CompletableFuture<StompSession> doConnect(StompSessionHandler handler) {
return ((ReactorNettyTcpStompClient) this.stompClient).connectAsync(getConnectHeaders(), handler);
}
}

8
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java
View File

@@ -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<StompSession> doConnect(StompSessionHandler handler) {
protected CompletableFuture<StompSession> 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);
}
}

15
spring-integration-stomp/src/test/java/org/springframework/integration/stomp/StompSessionManagerTests.java
View File

@@ -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<StompSession> doConnect(StompSessionHandler handler) {
protected CompletableFuture<StompSession> doConnect(StompSessionHandler handler) {
if (!this.thrown.getAndSet(true)) {
throw new RuntimeException("intentional");
}
else {
SettableListenableFuture<StompSession> future = new SettableListenableFuture<>();
CompletableFuture<StompSession> 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<StompSession> stompSessionFuture = new SettableListenableFuture<>();
final CompletableFuture<StompSession> stompSessionFuture = new CompletableFuture<>();
sessionManager.connect(new StompSessionHandlerAdapter() {
@Override
public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
stompSessionFuture.set(session);
stompSessionFuture.complete(session);
}
});

29
spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java
View File

@@ -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<WebSocketSession> future =
this.client.doHandshake(ClientWebSocketContainer.this.webSocketHandler,
CompletableFuture<WebSocketSession> future =
this.client.execute(ClientWebSocketContainer.this.webSocketHandler,
ClientWebSocketContainer.this.headers, getUri());
future.addCallback(new ListenableFutureCallback<WebSocketSession>() {
@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();
});
}

10
spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java
View File

@@ -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<WebSocketSession> doHandshakeInternal(WebSocketHandler webSocketHandler,
protected CompletableFuture<WebSocketSession> executeInternal(WebSocketHandler webSocketHandler,
HttpHeaders headers, URI uri, List<String> protocols, List<WebSocketExtension> extensions,
Map<String, Object> attributes) {
ListenableFuture<WebSocketSession> future =
super.doHandshakeInternal(webSocketHandler, headers, uri, protocols, extensions,
CompletableFuture<WebSocketSession> future =
super.executeInternal(webSocketHandler, headers, uri, protocols, extensions,
attributes);
if (failure.get()) {
future.cancel(true);

12
spring-integration-ws/src/test/java/org/springframework/integration/ws/SimpleWebServiceOutboundGatewayTests.java
View File

@@ -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<WebServiceMessage> requestFuture = new SettableListenableFuture<>();
final CompletableFuture<WebServiceMessage> 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<WebServiceMessage> requestFuture = new SettableListenableFuture<>();
final CompletableFuture<WebServiceMessage> 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);
}

33
src/reference/asciidoc/gateway.adoc
View File

@@ -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 <<gateway-asynctaskexecutor,task executor>> is an `AsyncListenableTaskExecutor`, the executor's `submitListenable()` method is invoked.
The following example shows how to use a `ListenableFuture`:
====
[source,java]
----
ListenableFuture<String> result = this.asyncGateway.async("something");
result.addCallback(new ListenableFutureCallback<String>() {
@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 `<gateway/>` 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 <<gw-completable-future>>.
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<Invoice>`, 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 <<gateway-asynctaskexecutor>> 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

4
src/reference/asciidoc/service-activator.adoc
View File

@@ -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.

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

@@ -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