From b64bf03fceee1f58901dd2b123b197802a12eb9b Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 29 Aug 2014 22:16:51 +0300 Subject: [PATCH] INT-3506 Async Gateway Improvements 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. --- .../annotation/AnnotationConstants.java | 37 ++++ .../annotation/MessagingGateway.java | 2 + .../config/MessagingGatewayRegistrar.java | 8 +- .../integration/config/xml/GatewayParser.java | 31 +++- .../gateway/GatewayProxyFactoryBean.java | 76 +++++++-- .../config/xml/spring-integration-4.1.xsd | 12 +- .../config/xml/GatewayParserTests-context.xml | 6 + .../config/xml/GatewayParserTests.java | 22 ++- .../gateway/AsyncGatewayTests.java | 161 +++++++++++++++++- .../gateway/GatewayInterfaceTests.java | 130 +++++++++++++- src/reference/docbook/amqp.xml | 2 +- src/reference/docbook/gateway.xml | 83 ++++++++- src/reference/docbook/whats-new.xml | 24 ++- 13 files changed, 547 insertions(+), 47 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/annotation/AnnotationConstants.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/AnnotationConstants.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/AnnotationConstants.java new file mode 100644 index 0000000000..6de6c54fee --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/AnnotationConstants.java @@ -0,0 +1,37 @@ +/* + * Copyright 2014 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.annotation; + +/** + * Common value constants for annotation attributes. + * + * @author Gary Russell + * @since 4.1 + * + */ +public final class AnnotationConstants { + + /** + * Constant defining a value as a replacement for {@code null} which + * we cannot use in annotation attributes. + */ + public static final String NULL = "__NULL__"; + + private AnnotationConstants() { + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java index 5b60e1b8bc..1fa3169c74 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java @@ -83,6 +83,8 @@ public @interface MessagingGateway { * to use for any of the interface methods that have a {@link java.util.concurrent.Future} return type. * This {@code Executor} will only be used for those async methods; the sync methods * will be invoked in the caller's thread. + * Use {@link AnnotationConstants#NULL} to specify no async executor - for example + * if your downstream flow returns a {@link java.util.concurrent.Future}. * @return the suggested executor bean name, if any */ String asyncExecutor() default ""; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessagingGatewayRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessagingGatewayRegistrar.java index b920a1e5cd..0162c9a69e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessagingGatewayRegistrar.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessagingGatewayRegistrar.java @@ -37,10 +37,11 @@ import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.annotation.ImportBeanDefinitionRegistrar; import org.springframework.core.type.AnnotationMetadata; import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.annotation.AnnotationConstants; import org.springframework.integration.annotation.MessagingGateway; -import org.springframework.integration.util.MessagingAnnotationUtils; import org.springframework.integration.gateway.GatewayMethodMetadata; import org.springframework.integration.gateway.GatewayProxyFactoryBean; +import org.springframework.integration.util.MessagingAnnotationUtils; import org.springframework.util.Assert; import org.springframework.util.MultiValueMap; import org.springframework.util.ObjectUtils; @@ -133,7 +134,10 @@ public class MessagingGatewayRegistrar implements ImportBeanDefinitionRegistrar if (StringUtils.hasText(errorChannel)) { gatewayProxyBuilder.addPropertyReference("errorChannel", errorChannel); } - if (StringUtils.hasText(asyncExecutor)) { + if (asyncExecutor == null || AnnotationConstants.NULL.equals(asyncExecutor)) { + gatewayProxyBuilder.addPropertyValue("asyncExecutor", null); + } + else if (StringUtils.hasText(asyncExecutor)) { gatewayProxyBuilder.addPropertyReference("asyncExecutor", asyncExecutor); } if (StringUtils.hasText(reactorEnvironment)) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java index de705b1e0a..1b4337c457 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java @@ -58,14 +58,26 @@ public class GatewayParser implements BeanDefinitionParser { final Map gatewayAttributes = new HashMap(); gatewayAttributes.put("name", element.getAttribute(AbstractBeanDefinitionParser.ID_ATTRIBUTE)); gatewayAttributes.put("defaultPayloadExpression", element.getAttribute("default-payload-expression")); - gatewayAttributes.put("defaultRequestChannel", element.getAttribute(isNested ? "request-channel" : "default-request-channel")); - gatewayAttributes.put("defaultReplyChannel", element.getAttribute(isNested ? "reply-channel" : "default-reply-channel")); + gatewayAttributes.put("defaultRequestChannel", + element.getAttribute(isNested ? "request-channel" : "default-request-channel")); + gatewayAttributes.put("defaultReplyChannel", + element.getAttribute(isNested ? "reply-channel" : "default-reply-channel")); gatewayAttributes.put("errorChannel", element.getAttribute("error-channel")); - gatewayAttributes.put("asyncExecutor", element.getAttribute("async-executor")); + + String asyncExecutor = element.getAttribute("async-executor"); + if (!element.hasAttribute("async-executor") || StringUtils.hasLength(asyncExecutor)) { + gatewayAttributes.put("asyncExecutor", asyncExecutor); + } + else { + gatewayAttributes.put("asyncExecutor", null); + } + gatewayAttributes.put("mapper", element.getAttribute("mapper")); gatewayAttributes.put("reactorEnvironment", element.getAttribute("reactor-environment")); - gatewayAttributes.put("defaultReplyTimeout", element.getAttribute(isNested ? "reply-timeout" : "default-reply-timeout")); - gatewayAttributes.put("defaultRequestTimeout", element.getAttribute(isNested ? "request-timeout" : "default-request-timeout")); + gatewayAttributes.put("defaultReplyTimeout", + element.getAttribute(isNested ? "reply-timeout" : "default-reply-timeout")); + gatewayAttributes.put("defaultRequestTimeout", + element.getAttribute(isNested ? "request-timeout" : "default-request-timeout")); List headerElements = DomUtils.getChildElementsByTagName(element, "default-header"); @@ -88,7 +100,8 @@ public class GatewayParser implements BeanDefinitionParser { String methodName = methodElement.getAttribute("name"); BeanDefinitionBuilder methodMetadataBuilder = BeanDefinitionBuilder.genericBeanDefinition( GatewayMethodMetadata.class); - methodMetadataBuilder.addPropertyValue("requestChannelName", methodElement.getAttribute("request-channel")); + methodMetadataBuilder.addPropertyValue("requestChannelName", + methodElement.getAttribute("request-channel")); methodMetadataBuilder.addPropertyValue("replyChannelName", methodElement.getAttribute("reply-channel")); methodMetadataBuilder.addPropertyValue("requestTimeout", methodElement.getAttribute("request-timeout")); methodMetadataBuilder.addPropertyValue("replyTimeout", methodElement.getAttribute("reply-timeout")); @@ -97,7 +110,8 @@ public class GatewayParser implements BeanDefinitionParser { Assert.state(!hasMapper || !StringUtils.hasText(element.getAttribute("payload-expression")), "'payload-expression' is not allowed when a 'mapper' is provided"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(methodMetadataBuilder, methodElement, "payload-expression"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(methodMetadataBuilder, methodElement, + "payload-expression"); List invocationHeaders = DomUtils.getChildElementsByTagName(methodElement, "header"); if (!CollectionUtils.isEmpty(invocationHeaders)) { @@ -106,7 +120,8 @@ public class GatewayParser implements BeanDefinitionParser { Map headerExpressions = new ManagedMap(); for (Element headerElement : invocationHeaders) { BeanDefinition expressionDef = IntegrationNamespaceUtils - .createExpressionDefinitionFromValueOrExpression("value", "expression", parserContext, headerElement, true); + .createExpressionDefinitionFromValueOrExpression("value", "expression", parserContext, + headerElement, true); headerExpressions.put(headerElement.getAttribute("name"), expressionDef); } 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 ce15063c78..925d70e832 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 @@ -29,6 +29,11 @@ import java.util.concurrent.Future; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; +import reactor.core.Environment; +import reactor.core.composable.Promise; +import reactor.core.composable.spec.Promises; +import reactor.function.Functions; + import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.AopUtils; import org.springframework.beans.SimpleTypeConverter; @@ -38,6 +43,7 @@ import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.FactoryBean; import org.springframework.core.convert.ConversionService; +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; @@ -62,11 +68,6 @@ import org.springframework.util.ObjectUtils; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; -import reactor.core.Environment; -import reactor.core.composable.Promise; -import reactor.core.composable.spec.Promises; -import reactor.function.Functions; - /** * Generates a proxy for the provided service interface to enable interaction * with messaging components without application code being aware of them allowing @@ -111,6 +112,10 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private volatile AsyncTaskExecutor asyncExecutor = new SimpleAsyncTaskExecutor(); + private volatile Class asyncSubmitType; + + private volatile Class asyncSubmitListenableType; + private volatile Environment reactorEnvironment; private volatile boolean initialized; @@ -214,9 +219,19 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } } + /** + * Set the executor for use when the gateway method returns + * {@link java.util.concurrent.Future} or {@link org.springframework.util.concurrent.ListenableFuture}. + * Set it to null to disable the async processing, and any + * {@link java.util.concurrent.Future} return types must be returned by the downstream flow. + * @param executor The executor. + */ public void setAsyncExecutor(Executor executor) { - Assert.notNull(executor, "executor must not be null"); - this.asyncExecutor = (executor instanceof AsyncTaskExecutor) ? (AsyncTaskExecutor) executor + if (executor == null && logger.isInfoEnabled()) { + logger.info("A null executor disables the async gateway; " + + "methods returning Future will run on the calling thread"); + } + this.asyncExecutor = (executor instanceof AsyncTaskExecutor || executor == null) ? (AsyncTaskExecutor) executor : new TaskExecutorAdapter(executor); } @@ -275,6 +290,21 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint this.gatewayMap.put(method, gateway); } this.serviceProxy = new ProxyFactory(proxyInterface, this).getProxy(this.beanClassLoader); + if (this.asyncExecutor != null) { + Callable task = new Callable() { + + @Override + public String call() throws Exception { + return null; + } + }; + Future 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.start(); this.initialized = true; } @@ -309,8 +339,20 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint @Override public Object invoke(final MethodInvocation invocation) throws Throwable { final Class returnType = invocation.getMethod().getReturnType(); - if (Future.class.isAssignableFrom(returnType)) { - return this.asyncExecutor.submit(new AsyncInvocationTask(invocation)); + if (this.asyncExecutor != null && !Object.class.equals(returnType)) { + if (returnType.isAssignableFrom(this.asyncSubmitType)) { + return this.asyncExecutor.submit(new AsyncInvocationTask(invocation)); + } + else if (returnType.isAssignableFrom(asyncSubmitListenableType)) { + return ((AsyncListenableTaskExecutor) this.asyncExecutor).submitListenable(new AsyncInvocationTask(invocation)); + } + else if (Future.class.isAssignableFrom(returnType)) { + if (logger.isDebugEnabled()) { + 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: " + + returnType.getSimpleName()); + } + } } if (Promise.class.isAssignableFrom(returnType)) { if (this.reactorEnvironment == null) { @@ -442,9 +484,12 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint boolean hasValue = StringUtils.hasText(value); if (!(hasValue ^ StringUtils.hasText(expression))) { - throw new BeanDefinitionStoreException("exactly one of 'value' or 'expression' is required on a gateway's header."); + throw new BeanDefinitionStoreException("exactly one of 'value' or 'expression' " + + "is required on a gateway's header."); } - headerExpressions.put(name, hasValue ? new LiteralExpression(value): PARSER.parseExpression(expression)); + headerExpressions.put(name, hasValue + ? new LiteralExpression(value) + : PARSER.parseExpression(expression)); } } @@ -476,7 +521,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } } } - GatewayMethodInboundMessageMapper messageMapper = new GatewayMethodInboundMessageMapper(method, headerExpressions, + GatewayMethodInboundMessageMapper messageMapper = new GatewayMethodInboundMessageMapper(method, + headerExpressions, this.globalMethodMetadata != null ? this.globalMethodMetadata.getHeaderExpressions() : null, this.argsMapper, this.getMessageBuilderFactory()); if (StringUtils.hasText(payloadExpression)) { @@ -506,9 +552,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint if (this.getBeanFactory() != null) { gateway.setBeanFactory(this.getBeanFactory()); } - if (this.shouldTrack) { - gateway.setShouldTrack(this.shouldTrack); - } + gateway.setShouldTrack(this.shouldTrack); gateway.afterPropertiesSet(); return gateway; } @@ -577,6 +621,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private MethodInvocationGateway(GatewayMethodInboundMessageMapper messageMapper) { this.setRequestMapper(messageMapper); } + } @@ -600,6 +645,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint throw new MessagingException("asynchronous gateway invocation failed", t); } } + } } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd index 185f33d817..9c8a863bd7 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd @@ -755,9 +755,17 @@ diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml index f0ca02288f..5d5249f2c1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml @@ -36,6 +36,12 @@ default-reply-channel="replyChannel" async-executor="testExecutor"/> + + 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> 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) { 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 index 7e12830953..e4237f79c0 100644 --- 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 @@ -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> f = service.returnMessageListenable("foo"); + long start = System.currentTimeMillis(); + final AtomicReference> result = new AtomicReference>(); + final CountDownLatch latch = new CountDownLatch(1); + f.addCallback(new ListenableFutureCallback>() { + + @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 reply = new GenericMessage(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> 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 returnStringPromise(String s); Promise> returnMessagePromise(String s); @@ -271,4 +389,43 @@ public class AsyncGatewayTests { } + private static class CustomFuture implements Future { + + 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; + } + + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java index 935e19714e..beff7d1c0c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java @@ -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 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 result2 = this.execGateway.test2(Thread.currentThread()); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference thread = new AtomicReference(); + result2.addCallback(new ListenableFutureCallback() { + + @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.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 map); + @Gateway(requestChannel = "gatewayThreadChannel") + Future test3(Thread caller); + + } + + @MessagingGateway(asyncExecutor = "exec") + public interface ExecGateway { + + @Gateway(requestChannel = "gatewayThreadChannel") + Future test1(Thread caller); + + @Gateway(requestChannel = "gatewayThreadChannel") + ListenableFuture test2(Thread caller); + + } + + @MessagingGateway(asyncExecutor = AnnotationConstants.NULL) + public interface NoExecGateway { + + @Gateway(requestChannel = "gatewayThreadChannel") + Future test1(Thread caller); + } } diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index 052219fbc5..c84bcf1fe4 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -749,7 +749,7 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann generic *, to avoid mapping of request headers to the reply. - Class AmqpHeaders + Class org.springframework.amqp.support.AmqpHeaders identifies the default headers that will be used by the DefaultAmqpHeaderMapper: diff --git a/src/reference/docbook/gateway.xml b/src/reference/docbook/gateway.xml index ae80e11759..737ddda6ed 100644 --- a/src/reference/docbook/gateway.xml +++ b/src/reference/docbook/gateway.xml @@ -442,13 +442,13 @@ of this chapter.
Asynchronous Gateway - As a pattern the Messaging Gateway is a very nice way to hide messaging-specific code while still exposing the full capabilities of the + As a pattern, the Messaging Gateway is a very nice way to hide messaging-specific code while still exposing the full capabilities of the messaging system. As you've seen, the GatewayProxyFactoryBean provides a convenient way to expose a Proxy over a service-interface thus giving you POJO-based access to a messaging system (based on objects in your own domain, or primitives/Strings, etc).  But when a gateway is exposed via simple POJO methods which return values it does imply that for each Request message (generated when the method is invoked) there must be a Reply message (generated when the method has returned). Since Messaging systems naturally are asynchronous you may not always be able to guarantee the contract where "for each request there will always be be a reply".  - With Spring Integration 2.0 we are introducing support for an Asynchronous Gateway which is a convenient way to initiate + With Spring Integration 2.0 we introduced support for an Asynchronous Gateway which is a convenient way to initiate flows where you may not know if a reply is expected or how long will it take for replies to arrive. @@ -462,14 +462,16 @@ of this chapter. service-interface="org.springframework.integration.sample.gateway.futures.MathServiceGateway" default-request-channel="requestChannel"/>]]> - However the Gateway Interface (service-interface) is a bit different. + However the Gateway Interface (service-interface) is a little different: public interface MathServiceGateway { + Future<Integer> multiplyByTwo(int i); + } - As you can see from the example above the return type for the gateway method is a Future. When + As you can see from the example above, the return type for the gateway method is a Future. When GatewayProxyFactoryBean sees that the return type of the gateway method is a Future, it immediately switches to the async mode by utilizing an AsyncTaskExecutor. That is all. The call to such a method always returns immediately with a Future instance. @@ -483,18 +485,85 @@ int finalResult =  result.get(1000, TimeUnit.SECONDS); async-gateway sample distributed within the Spring Integration samples. + + ListenableFuture + + + Starting with version 4.1, async gateway methods can also return + ListenableFuture (introduced in Spring Framework 4.0). These + return types allow you to provide a callback which is invoked when the result is available + (or an exception occurs). When the gateway detects this return type, and the task executor + (see below) is an AsyncListenableTaskExecutor, the executor's + submitListenable() method is invoked. + + result = this.asyncGateway.async("foo"); +result.addCallback(new ListenableFutureCallback() { + @Override + public void onSuccess(Thread result) { + ... + } + + @Override + public void onFailure(Throwable t) { + ... + } +});]]> Asynchronous Gateway and AsyncTaskExecutor - By default GatewayProxyFactoryBean uses org.springframework.core.task.SimpleAsyncTaskExecutor + By default, the GatewayProxyFactoryBean uses org.springframework.core.task.SimpleAsyncTaskExecutor when submitting internal AsyncInvocationTask instances for any gateway method whose - return type is Future.class. However the async-executor attribute in the + return type is Future. However the async-executor attribute in the <gateway/> element's configuration allows you to 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. 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): + + doAsync(String foo); + +}]]> + + If you wish to return a different Future + implementation, you can provide a custom executor, or disable the executor altogether and + return the Future in the reply message payload from the downstream flow. + To disable the executor, simply set it to null in the + GatewayProxyFactoryBean (setAsyncTaskExecutor(null)). When configuring + the gateway with XML, use async-executor=""; when configuring using the + @MessagingGateway annotation, use: + + doAsync(String foo); + +}]]> + + If the return type is a specific concrete Future implementation + or some other subinterface that is not supported by the configured executor, the flow will + run on the caller's thread and the flow must return the required type in the reply message + payload. + Asynchronous Gateway and Reactor Promise - Starting with version 4.1, the GatewayProxyFactoryBean allows the + Also starting with version 4.1, the GatewayProxyFactoryBean allows the use of a Reactor with gateway interface methods, utilizing a Promise<?> return type. The internal AsyncInvocationTask is wrapped in a diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 51e5265446..414569e320 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -11,13 +11,13 @@
New Components -
- Promise<?> Gateway - - A Reactor Promise return type is now supported for Messaging Gateway methods. - See . - -
+
+ Promise<?> Gateway + + A Reactor Promise return type is now supported for Messaging Gateway methods. + See . + +
General Changes @@ -177,5 +177,15 @@ See for more information.
+
+ Async Gateway + + In addition to the Promise return type mentioned above, + gateway methods may now return a ListenableFuture, introduced + in Spring Framework 4.0. You can also disable the async processing in the gateway, + allowing a downstream flow to directly return a Future. + See . + +