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

INT-943 added support for Future return types on Gateway interface methods

Browse Source
This commit is contained in:
Mark Fisher
2010-09-03 03:00:09 +00:00
parent b816ac7d44
commit 8a6ba97c1f
2 changed files with 200 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,10 +17,15 @@
package org.springframework.integration.gateway;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
@@ -33,9 +38,13 @@ import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.support.TaskExecutorAdapter;
import org.springframework.expression.Expression;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.AbstractEndpoint;
@@ -85,6 +94,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
private final Map<Method, MethodInvocationGateway> gatewayMap = new HashMap<Method, MethodInvocationGateway>();
private volatile AsyncTaskExecutor asyncExecutor = new SimpleAsyncTaskExecutor();
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
@@ -116,7 +127,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
this.serviceInterface = serviceInterface;
}
/**
* Set the default request channel.
*
@@ -168,6 +178,12 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
}
}
public void setAsyncExecutor(Executor executor) {
Assert.notNull(executor, "executor must not be null");
this.asyncExecutor = (executor instanceof AsyncTaskExecutor) ? (AsyncTaskExecutor) executor
: new TaskExecutorAdapter(executor);
}
public void setTypeConverter(TypeConverter typeConverter) {
Assert.notNull(typeConverter, "typeConverter must not be null");
this.typeConverter = typeConverter;
@@ -222,7 +238,14 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
return true;
}
public Object invoke(MethodInvocation invocation) throws Throwable {
public Object invoke(final MethodInvocation invocation) throws Throwable {
if (Future.class.isAssignableFrom(invocation.getMethod().getReturnType())) {
return this.asyncExecutor.submit(new AsyncInvocationTask(invocation));
}
return this.doInvoke(invocation);
}
private Object doInvoke(MethodInvocation invocation) throws Throwable {
Method method = invocation.getMethod();
if (AopUtils.isToStringMethod(method)) {
return "gateway proxy for service interface [" + this.serviceInterface + "]";
@@ -245,13 +268,14 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
Method method = invocation.getMethod();
MethodInvocationGateway gateway = this.gatewayMap.get(method);
Class<?> returnType = method.getReturnType();
boolean isReturnTypeMessage = Message.class.isAssignableFrom(returnType);
boolean shouldReturnMessage = Message.class.isAssignableFrom(returnType)
|| hasFutureParameterizedWithMessage(method);
boolean shouldReply = returnType != void.class;
int paramCount = method.getParameterTypes().length;
Object response = null;
if (paramCount == 0) {
if (shouldReply) {
if (isReturnTypeMessage) {
if (shouldReturnMessage) {
return gateway.receive();
}
response = gateway.receive();
@@ -260,7 +284,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
else {
Object[] args = invocation.getArguments();
if (shouldReply) {
response = isReturnTypeMessage ? gateway.sendAndReceiveMessage(args) : gateway.sendAndReceive(args);
response = shouldReturnMessage ? gateway.sendAndReceiveMessage(args) : gateway.sendAndReceive(args);
}
else {
gateway.send(args);
@@ -389,7 +413,11 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
this.exceptionMapper = exceptionMapper;
}
@SuppressWarnings("unchecked")
private <T> T convert(Object source, Class<T> expectedReturnType) {
if (Future.class.isAssignableFrom(expectedReturnType)) {
return (T) source;
}
if (this.getConversionService() != null) {
return this.getConversionService().convert(source, expectedReturnType);
}
@@ -398,6 +426,25 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
}
}
private static boolean hasFutureParameterizedWithMessage(Method method) {
if (Future.class.isAssignableFrom(method.getReturnType())) {
Type returnType = method.getGenericReturnType();
if (returnType instanceof ParameterizedType) {
Type[] typeArgs = ((ParameterizedType) returnType).getActualTypeArguments();
if (typeArgs != null && typeArgs.length == 1) {
Type parameterizedType = typeArgs[0];
if (parameterizedType instanceof ParameterizedType) {
Type rawType = ((ParameterizedType) parameterizedType).getRawType();
if (rawType instanceof Class) {
return Message.class.isAssignableFrom((Class<?>) rawType);
}
}
}
}
}
return false;
}
private static class MethodInvocationGateway extends AbstractMessagingGateway {
@@ -406,4 +453,26 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
}
}
private class AsyncInvocationTask implements Callable<Object> {
private final MethodInvocation invocation;
private AsyncInvocationTask(MethodInvocation invocation) {
this.invocation = invocation;
}
public Object call() throws Exception {
try {
return doInvoke(this.invocation);
}
catch (Throwable t) {
if (t instanceof RuntimeException) {
throw (RuntimeException) t;
}
throw new MessagingException("asynchronous gateway invocation failed", t);
}
}
}
}

126
spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java Normal file
View File

@@ -0,0 +1,126 @@
/*
* Copyright 2002-2010 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.gateway;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.message.GenericMessage;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.0
*/
public class AsyncGatewayTests {
@Test
public void futureWithMessageReturned() throws Exception {
QueueChannel requestChannel = new QueueChannel();
startResponder(requestChannel);
GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
proxyFactory.setDefaultRequestChannel(requestChannel);
proxyFactory.setServiceInterface(TestEchoService.class);
proxyFactory.setBeanName("testGateway");
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Future<Message<?>> f = service.returnMessage("foo");
long start = System.currentTimeMillis();
Object result = f.get(1000, TimeUnit.MILLISECONDS);
long elapsed = System.currentTimeMillis() - start;
assertTrue(elapsed >= 200);
assertTrue(result instanceof Message<?>);
assertEquals("foobar", ((Message<?>) result).getPayload());
}
@Test
public void futureWithPayloadReturned() throws Exception {
QueueChannel requestChannel = new QueueChannel();
startResponder(requestChannel);
GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
proxyFactory.setDefaultRequestChannel(requestChannel);
proxyFactory.setServiceInterface(TestEchoService.class);
proxyFactory.setBeanName("testGateway");
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Future<String> f = service.returnString("foo");
long start = System.currentTimeMillis();
Object result = f.get(1000, TimeUnit.MILLISECONDS);
long elapsed = System.currentTimeMillis() - start;
assertTrue(elapsed >= 200);
assertTrue(result instanceof String);
assertEquals("foobar", result);
}
@Test
public void futureWithWildcardReturned() throws Exception {
QueueChannel requestChannel = new QueueChannel();
startResponder(requestChannel);
GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean();
proxyFactory.setDefaultRequestChannel(requestChannel);
proxyFactory.setServiceInterface(TestEchoService.class);
proxyFactory.setBeanName("testGateway");
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Future<?> f = service.returnSomething("foo");
long start = System.currentTimeMillis();
Object result = f.get(1000, TimeUnit.MILLISECONDS);
long elapsed = System.currentTimeMillis() - start;
assertTrue(elapsed >= 200);
assertTrue(result instanceof String);
assertEquals("foobar", result);
}
private static void startResponder(final PollableChannel requestChannel) {
new Thread(new Runnable() {
public void run() {
Message<?> input = requestChannel.receive();
GenericMessage<String> reply = new GenericMessage<String>(input.getPayload() + "bar");
try {
Thread.sleep(200);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
((MessageChannel) input.getHeaders().getReplyChannel()).send(reply);
}
}).start();
}
static interface TestEchoService {
Future<String> returnString(String s);
Future<Message<?>> returnMessage(String s);
Future<?> returnSomething(String s);
}
}