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

GH-3635: Add Future<Void> & Mono<Void> to gateway (#3899)

Browse Source 
* GH-3635: Add Future<Void> & Mono<Void> to gateway

Fixes https://github.com/spring-projects/spring-integration/issues/3635

When `Future<Void>` & `Mono<Void>` is used as a messaging gateway return type,
the application hangs out on this barrier which may lead to the out of memory eventually

* Add support for the `Future<Void>` & `Mono<Void>` messaging gateway return type
and ensure an asynchronous call for the `gateway.send(Message)` operation and
its exception handling.
In case of successful call, the `Future` is fulfilled with `null` and `Mono` is completed as empty

* * Check for `void.class` as well in the `GatewayProxyFactoryBean.isVoidReturnType`

* * Allow `Future<Void>` as a reply type of the gateway request-reply operation

* * Fix  Checkstyle violations

* * Resolve `System.err.println()` in the test code

* Add `Thread.currentThread.interrupt()` to the `InterruptedException` block in the test
This commit is contained in:
Artem Bilan
2022-10-04 16:03:04 -04:00
committed by GitHub
parent 0eb6ae172e
commit 5fadaf533d
4 changed files with 125 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -564,15 +564,16 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
}
boolean shouldReturnMessage =
Message.class.isAssignableFrom(gateway.returnType) || (!runningOnCallerThread && gateway.expectMessage);
boolean shouldReply = gateway.returnType != void.class;
boolean oneWay =
void.class.isAssignableFrom(gateway.returnType) || (gateway.isVoidReturn && !runningOnCallerThread);
int paramCount = method.getParameterTypes().length;
Object response;
boolean hasPayloadExpression = findPayloadExpression(method);
if (paramCount == 0 && !hasPayloadExpression) {
response = receive(gateway, method, shouldReply, shouldReturnMessage);
response = receive(gateway, method, !oneWay, shouldReturnMessage);
}
else {
response = sendOrSendAndReceive(invocation, gateway, shouldReturnMessage, shouldReply);
response = sendOrSendAndReceive(invocation, gateway, shouldReturnMessage, !oneWay);
}
return response(gateway.returnType, shouldReturnMessage, response);
}
@@ -640,7 +641,12 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
}
}
else {
gateway.send(args);
if (gateway.isMonoReturn) {
return Mono.fromRunnable(() -> gateway.send(args));
}
else {
gateway.send(args);
}
}
return null;
}

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

@@ -30,18 +30,24 @@ import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.GatewayHeader;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.ReflectionUtils;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
/**
* @author Mark Fisher
@@ -70,7 +76,7 @@ public class AsyncGatewayTests {
}
@Test
public void futureWithError() throws Exception {
public void futureWithError() {
final Error error = new Error("error");
DirectChannel channel = new DirectChannel() {
@@ -140,7 +146,7 @@ public class AsyncGatewayTests {
}
@Test
public void nonAsyncFutureReturned() throws Exception {
public void nonAsyncFutureReturned() {
QueueChannel requestChannel = new QueueChannel();
addThreadEnricher(requestChannel);
startResponder(requestChannel);
@@ -204,6 +210,60 @@ public class AsyncGatewayTests {
assertThat(result).isEqualTo("foobar");
}
@Test
public void futureVoid() throws Exception {
GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
proxyFactory.setDefaultRequestChannel(new NullChannel());
proxyFactory.setBeanName("testGateway");
proxyFactory.setBeanFactory(mock(BeanFactory.class));
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Future<Void> f = service.asyncSendAndForget("test1");
Object result = f.get(10, TimeUnit.SECONDS);
assertThat(result).isNull();
new DirectFieldAccessor(proxyFactory).setPropertyValue("initialized", false);
proxyFactory.setDefaultRequestChannel((message, timeout) -> {
throw new MessageDispatchingException(message, "intentional dispatcher error");
});
proxyFactory.afterPropertiesSet();
Future<Void> futureError = service.asyncSendAndForget("test2");
assertThatExceptionOfType(ExecutionException.class)
.isThrownBy(() -> futureError.get(10, TimeUnit.SECONDS))
.withCauseInstanceOf(MessageDispatchingException.class)
.withMessageContaining("intentional dispatcher error");
}
@Test
public void futureVoidReply() throws Exception {
QueueChannel requestChannel = new QueueChannel();
CountDownLatch readyForReplyLatch = new CountDownLatch(1);
new Thread(() -> {
try {
Message<?> input = requestChannel.receive();
CompletableFuture<Void> reply = new CompletableFuture<>();
((MessageChannel) input.getHeaders().getReplyChannel()).send(new GenericMessage<>(reply));
readyForReplyLatch.await(10, TimeUnit.SECONDS);
reply.complete(null);
}
catch (InterruptedException ex) {
Thread.currentThread.interrupt();
ReflectionUtils.rethrowRuntimeException(ex);
}
}).start();
GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
proxyFactory.setDefaultRequestChannel(requestChannel);
proxyFactory.setBeanName("testGateway");
proxyFactory.setBeanFactory(mock(BeanFactory.class));
proxyFactory.setAsyncExecutor(null);
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Future<Void> f = service.sendAndReceiveFutureVoid("test");
readyForReplyLatch.countDown();
Object result = f.get(10, TimeUnit.SECONDS);
assertThat(result).isNull();
}
@Test
public void monoWithMessageReturned() {
@@ -252,7 +312,7 @@ public class AsyncGatewayTests {
}
@Test
public void monoWithConsumer() throws Exception {
public void monoWithConsumer() {
QueueChannel requestChannel = new QueueChannel();
startResponder(requestChannel);
GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
@@ -263,16 +323,38 @@ public class AsyncGatewayTests {
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Mono<String> mono = service.returnStringPromise("foo");
final AtomicReference<String> result = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);
StepVerifier.create(mono)
.expectNext("foobar")
.verifyComplete();
}
mono.subscribe(s -> {
result.set(s);
latch.countDown();
@Test
public void monoVoid() throws InterruptedException {
GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class);
proxyFactory.setDefaultRequestChannel(new NullChannel());
proxyFactory.setBeanName("testGateway");
proxyFactory.setBeanFactory(mock(BeanFactory.class));
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Mono<Void> mono = service.monoVoid("test1");
CountDownLatch emptyMonoLatch = new CountDownLatch(1);
mono.switchIfEmpty(Mono.empty().doOnSuccess(v -> emptyMonoLatch.countDown()).then()).subscribe();
assertThat(emptyMonoLatch.await(10, TimeUnit.SECONDS)).isTrue();
new DirectFieldAccessor(proxyFactory).setPropertyValue("initialized", false);
proxyFactory.setDefaultRequestChannel((message, timeout) -> {
throw new MessageDispatchingException(message, "intentional dispatcher error");
});
proxyFactory.afterPropertiesSet();
latch.await(10, TimeUnit.SECONDS);
assertThat(result.get()).isEqualTo("foobar");
Mono<Void> monoError = service.monoVoid("test2");
StepVerifier.create(monoError)
.expectSubscription()
.expectError(MessageDispatchingException.class)
.verify(Duration.ofSeconds(10));
}
private static void startResponder(final PollableChannel requestChannel) {
@@ -323,18 +405,15 @@ public class AsyncGatewayTests {
Mono<?> returnSomethingPromise(String s);
Future<Void> asyncSendAndForget(String s);
Future<Void> sendAndReceiveFutureVoid(String s);
Mono<Void> monoVoid(String s);
}
private static class CustomFuture implements Future<String> {
private final String result;
private final Thread thread;
private CustomFuture(String result, Thread thread) {
this.result = result;
this.thread = thread;
}
private record CustomFuture(String result, Thread thread) implements Future<String> {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {