diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 921d2d495b..ac2da02562 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -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 gatewayMap = new HashMap(); + 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 convert(Object source, Class 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 { + + 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); + } + } + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java new file mode 100644 index 0000000000..32d74a8b1a --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -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> 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 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 reply = new GenericMessage(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 returnString(String s); + + Future> returnMessage(String s); + + Future returnSomething(String s); + + } + +}