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

INT-3506 Async Gateway Improvements

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3506
JIRA: https://jira.spring.io/browse/INT-3428

Support flows downstream of the gateway that support
returning a `Future<?>` payload.

Currently, any method that returns a type that is assignable
to `Future<?>` runs async and returns a `FutureTask<?>`.

This prevents a service-interface method that returns a
custom `Future<?>` object from being invoked without wrapping
that `Future<?>` in a `FutureTask<?>`.

Allow the async-executor to be set to `null` causing any method
returning `Future<?>` to run on the calling thread.

Add support for `ListenableFuture<?>`.

If the return type is a `RunnableFuture`, `ListenableFuture` or
`Future`, or exactly a `FutureTask` or `ListenableFutureTask`, run the flow
on the executor (if present); otherwise run on the calling thread.

INT-3506 Fix Object returnType

INT-3506 Polishing - PR Comments

Perform dummy invocations of `submit` and `submitListenable` to
determine the actual return types so that we can determine at
runtime whether the executor will return a type that is compatible
with the method return type. If not, run on the caller's thread.

Add DEBUG Log If Incompatible Future<?>

INT-3506 Add Support for MessagingGateway

Add a constant to indicate no executor.

Add tests.

INT-3506 Polishing and Docs

- Docbook
- XSD
- Change test to send calling thread in payload so we can determine whether
   we need to return a Future or not; previously relied on the thread name
   which was brittle.

Polishing JavaDocs.
Change `amqp.xml` to use `org.springframework.amqp.support.AmqpHeaders` instead of an old one.
This commit is contained in:
Gary Russell
2014-08-29 22:16:51 +03:00
committed by Artem Bilan
parent e4eac27b12
commit b64bf03fce
13 changed files with 547 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -36,6 +36,12 @@
default-reply-channel="replyChannel"
async-executor="testExecutor"/>
<gateway id="asyncOff"
service-interface="org.springframework.integration.gateway.TestService"
default-request-channel="requestChannel"
default-reply-channel="replyChannel"
async-executor=""/>
<gateway id="promise"
service-interface="org.springframework.integration.gateway.TestService"
default-request-channel="requestChannel"

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

@@ -17,6 +17,8 @@
package org.springframework.integration.config.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
@@ -32,6 +34,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.gateway.TestService;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
@@ -93,6 +96,13 @@ public class GatewayParserTests {
Message<?> reply = result.get(1, TimeUnit.SECONDS);
assertEquals("foo", reply.getPayload());
assertEquals("testExecutor", reply.getHeaders().get("executor"));
assertNotNull(TestUtils.getPropertyValue(context.getBean("&async"), "asyncExecutor"));
}
@Test
public void testAsyncDisabledGateway() throws Exception {
Object service = context.getBean("&asyncOff");
assertNull(TestUtils.getPropertyValue(service, "asyncExecutor"));
}
@Test
@@ -104,10 +114,12 @@ public class GatewayParserTests {
Promise<Message<?>> result = service.promise("foo");
Message<?> reply = result.await(1, TimeUnit.SECONDS);
assertEquals("foo", reply.getPayload());
assertNotNull(TestUtils.getPropertyValue(context.getBean("&promise"), "asyncExecutor"));
}
private void startResponder(final PollableChannel requestChannel, final MessageChannel replyChannel) {
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Message<?> request = requestChannel.receive();
Message<?> reply = MessageBuilder.fromMessage(request)
@@ -125,6 +137,7 @@ public class GatewayParserTests {
private volatile String beanName;
@Override
public void setBeanName(String beanName) {
this.beanName = beanName;
}
@@ -135,8 +148,15 @@ public class GatewayParserTests {
try {
Future<?> result = super.submit(task);
Message<?> message = (Message<?>) result.get(1, TimeUnit.SECONDS);
Message<?> modifiedMessage = MessageBuilder.fromMessage(message)
Message<?> modifiedMessage;
if (message == null) {
modifiedMessage = MessageBuilder.withPayload("foo")
.setHeader("executor", this.beanName).build();
}
else {
modifiedMessage = MessageBuilder.fromMessage(message)
.setHeader("executor", this.beanName).build();
}
return new AsyncResult(modifiedMessage);
}
catch (Exception e) {

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

@@ -17,25 +17,33 @@
package org.springframework.integration.gateway;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.GatewayHeader;
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.GenericMessage;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import reactor.core.Environment;
import reactor.core.composable.Promise;
@@ -79,6 +87,97 @@ public class AsyncGatewayTests {
assertEquals("foobar", ((Message<?>) result).getPayload());
}
@Test
public void listenableFutureWithMessageReturned() throws Exception {
QueueChannel requestChannel = new QueueChannel();
addThreadEnricher(requestChannel);
startResponder(requestChannel);
GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
proxyFactory.setDefaultRequestChannel(requestChannel);
proxyFactory.setServiceInterface(TestEchoService.class);
proxyFactory.setBeanName("testGateway");
proxyFactory.setBeanFactory(mock(BeanFactory.class));
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
ListenableFuture<Message<?>> f = service.returnMessageListenable("foo");
long start = System.currentTimeMillis();
final AtomicReference<Message<?>> result = new AtomicReference<Message<?>>();
final CountDownLatch latch = new CountDownLatch(1);
f.addCallback(new ListenableFutureCallback<Message<?>>() {
@Override
public void onSuccess(Message<?> msg) {
result.set(msg);
latch.countDown();
}
@Override
public void onFailure(Throwable t) {
}
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
long elapsed = System.currentTimeMillis() - start;
assertTrue(elapsed >= 200);
assertEquals("foobar", result.get().getPayload());
Object thread = result.get().getHeaders().get("thread");
assertNotEquals(Thread.currentThread(), thread);
}
@Test
public void customFutureReturned() throws Exception {
QueueChannel requestChannel = new QueueChannel();
addThreadEnricher(requestChannel);
startResponder(requestChannel);
GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
proxyFactory.setDefaultRequestChannel(requestChannel);
proxyFactory.setServiceInterface(TestEchoService.class);
proxyFactory.setBeanName("testGateway");
proxyFactory.setBeanFactory(mock(BeanFactory.class));
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
CustomFuture f = service.returnCustomFuture("foo");
String result = f.get(1000, TimeUnit.MILLISECONDS);
assertEquals("foobar", result);
assertEquals(Thread.currentThread(), f.thread);
}
@Test
public void nonAsyncFutureReturned() throws Exception {
QueueChannel requestChannel = new QueueChannel();
addThreadEnricher(requestChannel);
startResponder(requestChannel);
GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
proxyFactory.setDefaultRequestChannel(requestChannel);
proxyFactory.setServiceInterface(TestEchoService.class);
proxyFactory.setBeanName("testGateway");
proxyFactory.setBeanFactory(mock(BeanFactory.class));
proxyFactory.setAsyncExecutor(null); // Not async - user flow returns Future<?>
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
CustomFuture f = (CustomFuture) service.returnCustomFutureWithTypeFuture("foo");
String result = f.get(1000, TimeUnit.MILLISECONDS);
assertEquals("foobar", result);
assertEquals(Thread.currentThread(), f.thread);
}
protected void addThreadEnricher(QueueChannel requestChannel) {
requestChannel.addInterceptor(new ChannelInterceptorAdapter() {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
return MessageBuilder.fromMessage(message)
.setHeader("thread", Thread.currentThread())
.build();
}
});
}
@Test
public void futureWithPayloadReturned() throws Exception {
QueueChannel requestChannel = new QueueChannel();
@@ -239,9 +338,13 @@ public class AsyncGatewayTests {
private static void startResponder(final PollableChannel requestChannel) {
new Thread(new Runnable() {
@Override
public void run() {
Message<?> input = requestChannel.receive();
GenericMessage<String> reply = new GenericMessage<String>(input.getPayload() + "bar");
String payload = input.getPayload() + "bar";
Message<?> reply = MessageBuilder.withPayload(payload)
.copyHeaders(input.getHeaders())
.build();
try {
Thread.sleep(200);
}
@@ -249,6 +352,13 @@ public class AsyncGatewayTests {
Thread.currentThread().interrupt();
return;
}
String header = (String) input.getHeaders().get("method");
if (header != null && header.startsWith("returnCustomFuture")) {
reply = MessageBuilder.withPayload(new CustomFuture(payload,
(Thread) input.getHeaders().get("thread")))
.copyHeaders(input.getHeaders())
.build();
}
((MessageChannel) input.getHeaders().getReplyChannel()).send(reply);
}
}).start();
@@ -263,6 +373,14 @@ public class AsyncGatewayTests {
Future<?> returnSomething(String s);
ListenableFuture<Message<?>> returnMessageListenable(String s);
@Gateway(headers=@GatewayHeader(name="method", expression="#gatewayMethod.name"))
CustomFuture returnCustomFuture(String s);
@Gateway(headers=@GatewayHeader(name="method", expression="#gatewayMethod.name"))
Future<?> returnCustomFutureWithTypeFuture(String s);
Promise<String> returnStringPromise(String s);
Promise<Message<?>> returnMessagePromise(String s);
@@ -271,4 +389,43 @@ public class AsyncGatewayTests {
}
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;
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return false;
}
@Override
public boolean isCancelled() {
return false;
}
@Override
public boolean isDone() {
return true;
}
@Override
public String get() throws InterruptedException, ExecutionException {
return result;
}
@Override
public String get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException,
TimeoutException {
return result;
}
}
}

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

@@ -17,8 +17,10 @@
package org.springframework.integration.gateway;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
@@ -31,7 +33,11 @@ import static org.mockito.Mockito.verify;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.Map;
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.hamcrest.Matchers;
import org.junit.Test;
@@ -39,17 +45,19 @@ import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.annotation.AnnotationConstants;
import org.springframework.integration.annotation.BridgeTo;
import org.springframework.integration.annotation.Gateway;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.support.MessageBuilder;
@@ -60,9 +68,15 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
/**
* @author Oleg Zhurakousky
@@ -77,6 +91,24 @@ public class GatewayInterfaceTests {
@Autowired
private Int2634Gateway int2634Gateway;
@Autowired
private ExecGateway execGateway;
@Autowired
private NoExecGateway noExecGateway;
@Autowired
@Qualifier("&gatewayInterfaceTests$ExecGateway")
private GatewayProxyFactoryBean execGatewayFB;
@Autowired
@Qualifier("&gatewayInterfaceTests$NoExecGateway")
private GatewayProxyFactoryBean noExecGatewayFB;
@Autowired
private SimpleAsyncTaskExecutor exec;
@Test
public void testWithServiceSuperclassAnnotatedMethod() throws Exception {
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
@@ -326,6 +358,45 @@ public class GatewayInterfaceTests {
assertEquals(param, result);
}
/*
* Tests use current thread in payload and reply has the thread that actually
* performed the send() on gatewayThreadChannel.
*/
@Test
public void testExecs() throws Exception {
assertSame(exec, TestUtils.getPropertyValue(execGatewayFB, "asyncExecutor"));
assertNull(TestUtils.getPropertyValue(noExecGatewayFB, "asyncExecutor"));
Future<Thread> result = this.int2634Gateway.test3(Thread.currentThread());
assertNotEquals(Thread.currentThread(), result.get());
assertThat(result.get().getName(), startsWith("SimpleAsync"));
result = this.execGateway.test1(Thread.currentThread());
assertNotEquals(Thread.currentThread(), result.get());
assertThat(result.get().getName(), startsWith("exec-"));
result = this.noExecGateway.test1(Thread.currentThread());
assertEquals(Thread.currentThread(), result.get());
ListenableFuture<Thread> result2 = this.execGateway.test2(Thread.currentThread());
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<Thread> thread = new AtomicReference<Thread>();
result2.addCallback(new ListenableFutureCallback<Thread>() {
@Override
public void onSuccess(Thread result) {
thread.set(result);
latch.countDown();
}
@Override
public void onFailure(Throwable t) {
}
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertThat(result2.get().getName(), startsWith("exec-"));
}
public interface Foo {
@@ -373,6 +444,39 @@ public class GatewayInterfaceTests {
public MessageChannel gatewayChannel() {
return new DirectChannel();
}
@Bean
@BridgeTo
public MessageChannel gatewayThreadChannel() {
DirectChannel channel = new DirectChannel();
channel.addInterceptor(new ChannelInterceptorAdapter() {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
Object payload;
if (Thread.currentThread().equals(message.getPayload())) {
// running on calling thread - need to return a Future.
payload = new AsyncResult<Thread>(Thread.currentThread());
}
else {
payload = Thread.currentThread();
}
return MessageBuilder.withPayload(payload)
.copyHeaders(message.getHeaders())
.build();
}
});
return channel;
}
@Bean
public AsyncTaskExecutor exec() {
SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
simpleAsyncTaskExecutor.setThreadNamePrefix("exec-");
return simpleAsyncTaskExecutor;
}
}
@MessagingGateway
@@ -384,6 +488,28 @@ public class GatewayInterfaceTests {
@Gateway(requestChannel = "gatewayChannel")
Object test2(@Payload Map<Object, ?> map);
@Gateway(requestChannel = "gatewayThreadChannel")
Future<Thread> test3(Thread caller);
}
@MessagingGateway(asyncExecutor = "exec")
public interface ExecGateway {
@Gateway(requestChannel = "gatewayThreadChannel")
Future<Thread> test1(Thread caller);
@Gateway(requestChannel = "gatewayThreadChannel")
ListenableFuture<Thread> test2(Thread caller);
}
@MessagingGateway(asyncExecutor = AnnotationConstants.NULL)
public interface NoExecGateway {
@Gateway(requestChannel = "gatewayThreadChannel")
Future<Thread> test1(Thread caller);
}
}