From 8ac4bfbd2c20d3378ec66c653f3b679c2e428822 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 30 Jul 2013 17:29:20 +0300 Subject: [PATCH] INT-3477: Add Reactor Promise for Gateway JIRA: https://jira.spring.io/browse/INT-3477 INT-3477: Fix `reactorEnvironment` propagation * `MessagingGatewayRegistrar` parser the value for the `reactorEnvironment` * Provide more interest test-case * Polishing docs INT-3477 Require `Environment` in case of Promise * Do not use Reactor `Environment` as default instance. * Require `Environment` reference, when is `Promise` method * Polishing for tests * Apply Gary's polishing for docs INT-3477: Assert.notNull -> Assert.state INT-3477: PR review --- build.gradle | 4 +- .../annotation/MessagingGateway.java | 21 +-- .../config/MessagingGatewayRegistrar.java | 4 + .../integration/config/xml/GatewayParser.java | 1 + .../gateway/GatewayProxyFactoryBean.java | 41 ++++- .../config/xml/spring-integration-4.1.xsd | 16 +- ...sts.xml => GatewayParserTests-context.xml} | 34 ++-- .../config/xml/GatewayParserTests.java | 39 +++-- .../configuration/EnableIntegrationTests.java | 67 +++++++- .../gateway/AsyncGatewayTests.java | 145 +++++++++++++++++- .../integration/gateway/TestService.java | 11 +- src/reference/docbook/gateway.xml | 68 +++++++- src/reference/docbook/whats-new.xml | 10 ++ 13 files changed, 413 insertions(+), 48 deletions(-) rename spring-integration-core/src/test/java/org/springframework/integration/config/xml/{gatewayParserTests.xml => GatewayParserTests-context.xml} (50%) diff --git a/build.gradle b/build.gradle index e542a1b147..409eb79138 100644 --- a/build.gradle +++ b/build.gradle @@ -101,7 +101,8 @@ subprojects { subproject -> openJpaVersion = '2.3.0' pahoMqttClientVersion = '0.4.0' postgresVersion = '9.1-901-1.jdbc4' - reactorVersion = '1.1.2.RELEASE' + reactorVersion = '1.1.4.RELEASE' + reactorSpringVersion = '1.1.3.RELEASE' romeToolsVersion = '1.5.0' romeVersion = '1.0.0' saajApiVersion = '1.3.5' @@ -243,6 +244,7 @@ project('spring-integration-core') { compile "org.springframework:spring-tx:$springVersion" compile "org.springframework.retry:spring-retry:$springRetryVersion" compile "org.projectreactor:reactor-core:$reactorVersion" + compile("org.projectreactor.spring:reactor-spring-context:$reactorSpringVersion", optional) compile("org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion", optional) compile("com.fasterxml.jackson.core:jackson-databind:$jackson2Version", optional) compile("com.jayway.jsonpath:json-path:$jsonpathVersion", optional) 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 3287d89c0e..5b60e1b8bc 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 @@ -36,14 +36,12 @@ public @interface MessagingGateway { /** * The value may indicate a suggestion for a logical component name, * to be turned into a Spring bean in case of an autodetected component. - * * @return the suggested component name, if any */ String name() default ""; /** * Identifies default channel the messages will be sent to upon invocation of methods of the gateway proxy. - * * @return the suggested channel name, if any */ String defaultRequestChannel() default ""; @@ -51,7 +49,6 @@ public @interface MessagingGateway { /** * Identifies default channel the gateway proxy will subscribe to to receive reply {@code Message}s, which will then be * converted to the return type of the method signature. - * * @return the suggested channel name, if any */ String defaultReplyChannel() default ""; @@ -61,7 +58,6 @@ public @interface MessagingGateway { * gateway's proxy invocation. If no {@code errorChannel} reference is provided, the gateway will * propagate {@code Exception}s to the caller. To completely suppress {@code Exception}s, provide a * reference to the {@code nullChannel} here. - * * @return the suggested channel name, if any */ String errorChannel() default ""; @@ -70,7 +66,6 @@ public @interface MessagingGateway { * Provides the amount of time dispatcher would wait to send a {@code Message}. * This timeout would only apply if there is a potential to block in the send call. * For example if this gateway is hooked up to a {@code QueueChannel}.  - * * @return the suggested timeout in milliseconds, if any */ long defaultRequestTimeout() default Long.MIN_VALUE; @@ -79,7 +74,6 @@ public @interface MessagingGateway { * Allows to specify how long this gateway will wait for the reply {@code Message} * before returning. By default it will wait indefinitely. {@code null} is returned if the gateway times out. - * * @return the suggested timeout in milliseconds, if any */ long defaultReplyTimeout() default Long.MIN_VALUE; @@ -89,7 +83,6 @@ 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. - * * @return the suggested executor bean name, if any */ String asyncExecutor() default ""; @@ -99,7 +92,6 @@ public @interface MessagingGateway { * unless explicitly overridden by a method declaration. Variables include {@code #args}, {@code #methodName}, * {@code #methodString} and {@code #methodObject}; * a bean resolver is also available, enabling expressions like {@code @someBean(#args)}. - * * @return the suggested payload expression, if any */ String defaultPayloadExpression() default ""; @@ -107,7 +99,6 @@ public @interface MessagingGateway { /** * Provides custom message headers. These default headers are created for * all methods on the service-interface (unless overridden by a specific method). - * * @return the suggested payload expression, if any */ GatewayHeader[] defaultHeaders() default {}; @@ -117,9 +108,19 @@ public @interface MessagingGateway { * to map the method arguments to a {@link org.springframework.messaging.Message}. When this * is provided, no {@code payload-expression}s or {@code header}s are allowed; the custom mapper is * responsible for creating the message. - * * @return the suggested mapper bean name, if any */ String mapper() default ""; + /** + * Provide a reference to an {@link reactor.core.Environment} + * to use for any of the interface methods that have a {@link reactor.core.composable.Promise} return type. + * This {@code reactor.core.Environment} will only be used for those async methods; the sync methods + * will be invoked in the caller's thread. + *

This attribute is required in case of {@link reactor.core.composable.Promise} usage. + * @return the suggested reactor Environment bean name. + * @since 4.1 + */ + String reactorEnvironment() 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 15479975e8..0e645efa20 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 @@ -83,6 +83,7 @@ public class MessagingGatewayRegistrar implements ImportBeanDefinitionRegistrar String defaultReplyChannel = (String) gatewayAttributes.get("defaultReplyChannel"); String errorChannel = (String) gatewayAttributes.get("errorChannel"); String asyncExecutor = (String) gatewayAttributes.get("asyncExecutor"); + String reactorEnvironment = (String) gatewayAttributes.get("reactorEnvironment"); String mapper = (String) gatewayAttributes.get("mapper"); boolean hasMapper = StringUtils.hasText(mapper); @@ -135,6 +136,9 @@ public class MessagingGatewayRegistrar implements ImportBeanDefinitionRegistrar if (StringUtils.hasText(asyncExecutor)) { gatewayProxyBuilder.addPropertyReference("asyncExecutor", asyncExecutor); } + if (StringUtils.hasText(reactorEnvironment)) { + gatewayProxyBuilder.addPropertyReference("reactorEnvironment", reactorEnvironment); + } if (StringUtils.hasText(mapper)) { gatewayProxyBuilder.addPropertyReference("mapper", mapper); } 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 8656978ace..de705b1e0a 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 @@ -63,6 +63,7 @@ public class GatewayParser implements BeanDefinitionParser { gatewayAttributes.put("errorChannel", element.getAttribute("error-channel")); gatewayAttributes.put("asyncExecutor", element.getAttribute("async-executor")); 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")); 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 44c971eb1a..9414c2426b 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 @@ -62,6 +62,11 @@ 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 @@ -75,7 +80,8 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Artem Bilan */ -public class GatewayProxyFactoryBean extends AbstractEndpoint implements TrackableComponent, FactoryBean, MethodInterceptor, BeanClassLoaderAware { +public class GatewayProxyFactoryBean extends AbstractEndpoint + implements TrackableComponent, FactoryBean, MethodInterceptor, BeanClassLoaderAware { private static final SpelExpressionParser PARSER = new SpelExpressionParser(); @@ -105,6 +111,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab private volatile AsyncTaskExecutor asyncExecutor = new SimpleAsyncTaskExecutor(); + private volatile Environment reactorEnvironment; + private volatile boolean initialized; private final Object initializationMonitor = new Object(); @@ -225,6 +233,17 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab this.globalMethodMetadata = globalMethodMetadata; } + /** + * Set the Reactor {@link Environment} to be used for processing methods with a + * {@link Promise} return type. (Required when any such methods are declared on the + * service interface). + * @param reactorEnvironment the Reactor Environment. + * @since 4.1 + */ + public void setReactorEnvironment(Environment reactorEnvironment) { + this.reactorEnvironment = reactorEnvironment; + } + @Override public void setBeanClassLoader(ClassLoader beanClassLoader) { this.beanClassLoader = beanClassLoader; @@ -289,9 +308,18 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab @Override public Object invoke(final MethodInvocation invocation) throws Throwable { - if (Future.class.isAssignableFrom(invocation.getMethod().getReturnType())) { + final Class returnType = invocation.getMethod().getReturnType(); + if (Future.class.isAssignableFrom(returnType)) { return this.asyncExecutor.submit(new AsyncInvocationTask(invocation)); } + if (Promise.class.isAssignableFrom(returnType)) { + if (this.reactorEnvironment == null) { + throw new IllegalStateException("'reactorEnvironment' is required in case of 'Promise' return type."); + } + return Promises.task(Functions.supplier(new AsyncInvocationTask(invocation))) + .env(this.reactorEnvironment) + .get(); + } return this.doInvoke(invocation); } @@ -317,7 +345,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab MethodInvocationGateway gateway = this.gatewayMap.get(method); Class returnType = method.getReturnType(); boolean shouldReturnMessage = Message.class.isAssignableFrom(returnType) - || hasFutureParameterizedWithMessage(method); + || hasReturnParameterizedWithMessage(method); boolean shouldReply = returnType != void.class; int paramCount = method.getParameterTypes().length; Object response = null; @@ -510,6 +538,9 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab if (Future.class.isAssignableFrom(expectedReturnType)) { return (T) source; } + if (Promise.class.isAssignableFrom(expectedReturnType)) { + return (T) source; + } if (this.getConversionService() != null) { return this.getConversionService().convert(source, expectedReturnType); } @@ -518,8 +549,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab } } - private static boolean hasFutureParameterizedWithMessage(Method method) { - if (Future.class.isAssignableFrom(method.getReturnType())) { + private static boolean hasReturnParameterizedWithMessage(Method method) { + if (Future.class.isAssignableFrom(method.getReturnType()) || Promise.class.isAssignableFrom(method.getReturnType())) { Type returnType = method.getGenericReturnType(); if (returnType instanceof ParameterizedType) { Type[] typeArgs = ((ParameterizedType) returnType).getActualTypeArguments(); 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 45a2434abc..d27d5a10dc 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 @@ -778,6 +778,20 @@ + + + + + + + @@ -1462,7 +1476,7 @@ diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/gatewayParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml similarity index 50% rename from spring-integration-core/src/test/java/org/springframework/integration/config/xml/gatewayParserTests.xml rename to spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml index 2b4c5e5cb9..f0ca02288f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/gatewayParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml @@ -1,8 +1,8 @@ @@ -16,29 +16,37 @@ + service-interface="org.springframework.integration.gateway.TestService" + default-request-channel="requestChannel"/> + service-interface="org.springframework.integration.gateway.TestService" + default-reply-channel="replyChannel" + default-reply-timeout="3000"/> + service-interface="org.springframework.integration.gateway.TestService" + default-request-channel="requestChannel" + default-reply-channel="replyChannel" + default-reply-timeout="5000"/> - + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java index ee48b647e1..17888f45c4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-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. @@ -24,26 +24,39 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import org.junit.Test; +import org.junit.runner.RunWith; + import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; -import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.integration.gateway.TestService; +import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; -import org.springframework.integration.gateway.TestService; import org.springframework.messaging.support.GenericMessage; -import org.springframework.integration.support.MessageBuilder; 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 reactor.core.composable.Promise; /** * @author Mark Fisher + * @author Artem Bilan */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD) public class GatewayParserTests { + @Autowired + private ApplicationContext context; + @Test public void testOneWay() { - ApplicationContext context = new ClassPathXmlApplicationContext("gatewayParserTests.xml", this.getClass()); TestService service = (TestService) context.getBean("oneWay"); service.oneWay("foo"); PollableChannel channel = (PollableChannel) context.getBean("requestChannel"); @@ -53,7 +66,6 @@ public class GatewayParserTests { @Test public void testSolicitResponse() { - ApplicationContext context = new ClassPathXmlApplicationContext("gatewayParserTests.xml", this.getClass()); PollableChannel channel = (PollableChannel) context.getBean("replyChannel"); channel.send(new GenericMessage("foo")); TestService service = (TestService) context.getBean("solicitResponse"); @@ -63,18 +75,16 @@ public class GatewayParserTests { @Test public void testRequestReply() { - ApplicationContext context = new ClassPathXmlApplicationContext("gatewayParserTests.xml", this.getClass()); PollableChannel requestChannel = (PollableChannel) context.getBean("requestChannel"); MessageChannel replyChannel = (MessageChannel) context.getBean("replyChannel"); this.startResponder(requestChannel, replyChannel); TestService service = (TestService) context.getBean("requestReply"); String result = service.requestReply("foo"); - assertEquals("foo", result); + assertEquals("foo", result); } @Test public void testAsyncGateway() throws Exception { - ApplicationContext context = new ClassPathXmlApplicationContext("gatewayParserTests.xml", this.getClass()); PollableChannel requestChannel = (PollableChannel) context.getBean("requestChannel"); MessageChannel replyChannel = (MessageChannel) context.getBean("replyChannel"); this.startResponder(requestChannel, replyChannel); @@ -85,6 +95,16 @@ public class GatewayParserTests { assertEquals("testExecutor", reply.getHeaders().get("executor")); } + @Test + public void testPromiseGateway() throws Exception { + PollableChannel requestChannel = context.getBean("requestChannel", PollableChannel.class); + MessageChannel replyChannel = context.getBean("replyChannel", MessageChannel.class); + this.startResponder(requestChannel, replyChannel); + TestService service = context.getBean("promise", TestService.class); + Promise> result = service.promise("foo"); + Message reply = result.await(1, TimeUnit.SECONDS); + assertEquals("foo", reply.getPayload()); + } private void startResponder(final PollableChannel requestChannel, final MessageChannel replyChannel) { Executors.newSingleThreadExecutor().execute(new Runnable() { @@ -123,6 +143,7 @@ public class GatewayParserTests { throw new IllegalStateException("unexpected exception in testExecutor", e); } } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java index 7a3c8af121..701abb9351 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java @@ -32,6 +32,7 @@ import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; +import java.util.Arrays; import java.util.Date; import java.util.List; import java.util.concurrent.CountDownLatch; @@ -116,6 +117,14 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.support.AnnotationConfigContextLoader; +import reactor.core.Environment; +import reactor.core.composable.Composable; +import reactor.core.composable.Promise; +import reactor.core.composable.spec.Streams; +import reactor.function.Consumer; +import reactor.function.Function; +import reactor.spring.context.config.EnableReactor; + /** * @author Artem Bilan * @since 4.0 @@ -541,6 +550,49 @@ public class EnableIntegrationTests { assertNull(replyChannel.receive(10)); } + @Autowired + private Environment environment; + + @Test + public void testPromiseGateway() throws Exception { + + final AtomicReference> ref = new AtomicReference>(); + final CountDownLatch consumeLatch = new CountDownLatch(1); + + Streams.defer(Arrays.asList("1", "2", "3", "4", "5")) + .env(this.environment) + .get() + .map(new Function() { + @Override + public Integer apply(String s) { + return Integer.parseInt(s); + } + }) + .mapMany(new Function>() { + @Override + public Composable apply(Integer integer) { + return testGateway.multiply(integer); + } + }) + .collect() + .consume(new Consumer>() { + @Override + public void accept(List integers) { + ref.set(integers); + consumeLatch.countDown(); + } + }) + .flush(); + + + assertTrue(consumeLatch.await(2, TimeUnit.SECONDS)); + + List integers = ref.get(); + assertEquals(5, integers.size()); + + assertThat(integers, Matchers.contains(2, 4, 6, 8, 10)); + } + @Configuration @ComponentScan @IntegrationComponentScan @@ -775,6 +827,7 @@ public class EnableIntegrationTests { @EnableMessageHistory("${message.history.tracked.components}") @EnablePublisher("publishedChannel") @EnableAsync + @EnableReactor public static class ContextConfiguration2 { @Bean @@ -873,6 +926,11 @@ public class EnableIntegrationTests { return new SerializingConverter(); } + @Bean + public DirectChannel promiseChannel() { + return new DirectChannel(); + } + } @Configuration @@ -1038,6 +1096,10 @@ public class EnableIntegrationTests { /*@BridgeFrom("") public void invalidBridgeAnnotationMethod(Object payload) {}*/ + @ServiceActivator(inputChannel = "promiseChannel") + public Integer multiply(Integer value) { + return value * 2; + } } @TestMessagingGateway @@ -1050,6 +1112,9 @@ public class EnableIntegrationTests { @Async void sendAsync(String payload); + @Gateway(requestChannel = "promiseChannel") + Promise multiply(Integer value); + } @TestMessagingGateway2 @@ -1062,7 +1127,7 @@ public class EnableIntegrationTests { @Target({ElementType.TYPE, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) - @MessagingGateway(defaultRequestChannel = "gatewayChannel", + @MessagingGateway(defaultRequestChannel = "gatewayChannel", reactorEnvironment = "reactorEnv", defaultHeaders = @GatewayHeader(name = "foo", value = "FOO")) public static @interface TestMessagingGateway { 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 75703ce031..7e12830953 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-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. @@ -17,32 +17,48 @@ package org.springframework.integration.gateway; import static org.junit.Assert.assertEquals; +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.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.hamcrest.Matchers; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.channel.QueueChannel; -import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; + +import reactor.core.Environment; +import reactor.core.composable.Promise; +import reactor.function.Consumer; /** * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan * @since 2.0 */ public class AsyncGatewayTests { + private final Environment reactorEnvironment = new Environment(); + // TODO: changed from 0 because of recurrent failure: is this right? private final long safety = 100; + public void tearDown() { + this.reactorEnvironment.shutdown(); + } + @Test public void futureWithMessageReturned() throws Exception { QueueChannel requestChannel = new QueueChannel(); @@ -79,7 +95,7 @@ public class AsyncGatewayTests { Object result = f.get(1000, TimeUnit.MILLISECONDS); long elapsed = System.currentTimeMillis() - start; - assertTrue(elapsed >= 200-safety); + assertTrue(elapsed >= 200 - safety); assertTrue(result instanceof String); assertEquals("foobar", result); } @@ -100,12 +116,127 @@ public class AsyncGatewayTests { Object result = f.get(1000, TimeUnit.MILLISECONDS); long elapsed = System.currentTimeMillis() - start; - assertTrue(elapsed >= 200-safety); + assertTrue(elapsed >= 200 - safety); assertTrue(result instanceof String); assertEquals("foobar", result); } + @Test + public void promiseWithMessageReturned() throws Exception { + QueueChannel requestChannel = new QueueChannel(); + startResponder(requestChannel); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setDefaultRequestChannel(requestChannel); + proxyFactory.setServiceInterface(TestEchoService.class); + proxyFactory.setBeanFactory(mock(BeanFactory.class)); + proxyFactory.setBeanName("testGateway"); + proxyFactory.setReactorEnvironment(this.reactorEnvironment); + proxyFactory.afterPropertiesSet(); + TestEchoService service = (TestEchoService) proxyFactory.getObject(); + Promise> promise = service.returnMessagePromise("foo"); + long start = System.currentTimeMillis(); + Object result = promise.await(1, TimeUnit.SECONDS); + long elapsed = System.currentTimeMillis() - start; + assertTrue(elapsed >= 200); + assertEquals("foobar", ((Message) result).getPayload()); + } + + @Test + public void promiseWithPayloadReturned() throws Exception { + QueueChannel requestChannel = new QueueChannel(); + startResponder(requestChannel); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setDefaultRequestChannel(requestChannel); + proxyFactory.setServiceInterface(TestEchoService.class); + proxyFactory.setBeanFactory(mock(BeanFactory.class)); + proxyFactory.setReactorEnvironment(this.reactorEnvironment); + proxyFactory.setBeanName("testGateway"); + proxyFactory.afterPropertiesSet(); + TestEchoService service = (TestEchoService) proxyFactory.getObject(); + Promise promise = service.returnStringPromise("foo"); + long start = System.currentTimeMillis(); + Object result = promise.await(1, TimeUnit.SECONDS); + long elapsed = System.currentTimeMillis() - start; + + assertTrue(elapsed >= 200 - safety); + assertEquals("foobar", result); + } + + @Test + public void promiseWithWildcardReturned() throws Exception { + QueueChannel requestChannel = new QueueChannel(); + startResponder(requestChannel); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setDefaultRequestChannel(requestChannel); + proxyFactory.setServiceInterface(TestEchoService.class); + proxyFactory.setBeanFactory(mock(BeanFactory.class)); + proxyFactory.setBeanName("testGateway"); + proxyFactory.setReactorEnvironment(this.reactorEnvironment); + proxyFactory.afterPropertiesSet(); + TestEchoService service = (TestEchoService) proxyFactory.getObject(); + Promise promise = service.returnSomethingPromise("foo"); + long start = System.currentTimeMillis(); + Object result = promise.await(1, TimeUnit.SECONDS); + long elapsed = System.currentTimeMillis() - start; + + assertTrue(elapsed >= 200 - safety); + assertTrue(result instanceof String); + assertEquals("foobar", result); + } + + @Test + public void promiseWithConsumer() throws Exception { + QueueChannel requestChannel = new QueueChannel(); + startResponder(requestChannel); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setDefaultRequestChannel(requestChannel); + proxyFactory.setServiceInterface(TestEchoService.class); + proxyFactory.setBeanFactory(mock(BeanFactory.class)); + proxyFactory.setBeanName("testGateway"); + proxyFactory.setReactorEnvironment(this.reactorEnvironment); + proxyFactory.afterPropertiesSet(); + TestEchoService service = (TestEchoService) proxyFactory.getObject(); + Promise promise = service.returnStringPromise("foo"); + long start = System.currentTimeMillis(); + + final AtomicReference result = new AtomicReference(); + final CountDownLatch latch = new CountDownLatch(1); + + promise.consume(new Consumer() { + @Override + public void accept(String s) { + result.set(s); + latch.countDown(); + } + }) + .flush(); + + latch.await(1, TimeUnit.SECONDS); + long elapsed = System.currentTimeMillis() - start; + + assertTrue(elapsed >= 200 - safety); + assertEquals("foobar", result.get()); + } + + @Test + public void promiseMethodWithoutEnvironment() throws Exception { + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setServiceInterface(TestEchoService.class); + proxyFactory.setBeanFactory(mock(BeanFactory.class)); + proxyFactory.setBeanName("testGateway"); + proxyFactory.afterPropertiesSet(); + TestEchoService service = (TestEchoService) proxyFactory.getObject(); + try { + service.returnStringPromise("foo"); + fail("IllegalStateException expected"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(IllegalStateException.class)); + assertEquals(e.getMessage(), "'reactorEnvironment' is required in case of 'Promise' return type."); + } + } + private static void startResponder(final PollableChannel requestChannel) { new Thread(new Runnable() { public void run() { @@ -132,6 +263,12 @@ public class AsyncGatewayTests { Future returnSomething(String s); + Promise returnStringPromise(String s); + + Promise> returnMessagePromise(String s); + + Promise returnSomethingPromise(String s); + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java index c0b84d4232..d3ff8fdac3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-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. @@ -18,17 +18,20 @@ package org.springframework.integration.gateway; import java.util.concurrent.Future; -import org.springframework.messaging.Message; import org.springframework.integration.annotation.Payload; +import org.springframework.messaging.Message; + +import reactor.core.composable.Promise; /** * @author Mark Fisher * @author Oleg Zhurakousky + * @author Artem Bilan */ public interface TestService { String requestReply(String input); - + byte[] requestReplyInBytes(String input); void oneWay(String input); @@ -46,4 +49,6 @@ public interface TestService { Future> async(String s); + Promise> promise(String s); + } diff --git a/src/reference/docbook/gateway.xml b/src/reference/docbook/gateway.xml index 7148110d93..ae80e11759 100644 --- a/src/reference/docbook/gateway.xml +++ b/src/reference/docbook/gateway.xml @@ -484,7 +484,7 @@ int finalResult =  result.get(1000, TimeUnit.SECONDS); async-gateway sample distributed within the Spring Integration samples. - Asynchronous Gateway and AsyncTaskExecutor + Asynchronous Gateway and AsyncTaskExecutor By default GatewayProxyFactoryBean uses org.springframework.core.task.SimpleAsyncTaskExecutor when submitting internal AsyncInvocationTask instances for any gateway method whose @@ -492,6 +492,72 @@ int finalResult =  result.get(1000, TimeUnit.SECONDS); <gateway/> element's configuration allows you to provide a reference to any implementation of java.util.concurrent.Executor available within the Spring application context. + Asynchronous Gateway and Reactor Promise + + 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 + reactor.function.Supplier with the provided reactorEnvironment, using + a default RingBufferDispatcher for the Promise + consumption. Note, a reactorEnvironment reference is required whenever a service interface + has at least one method with a Promise<?> return type. (Only those methods run + on the reactor's dispatcher). + + + A Promise can be used to retrieve the result later (similar to a + Future<?>) or you can consume from it with the dispatcher invoking + your Consumer when the result is returned to the gateway. + + + + The Promise isn't flushed immediately by the framework. + Hence the underlying message flow won't be started before the gateway method returns (as it is with + Future<?> Executor task). + The flow will be started when the Promise is flushed or via + Promise.await(). Alternatively, the Promise (being a + Composable) might be a part of Reactor Stream<?>, + when the flush() is related to the entire Stream. For example: + + @MessagingGateway(reactorEnvironment = "reactorEnv") +public static interface TestGateway { + + @Gateway(requestChannel = "promiseChannel") + Promise<Integer> multiply(Integer value); + + } + + ... + + @ServiceActivator(inputChannel = "promiseChannel") + public Integer multiply(Integer value) { + return value * 2; + } + + ... + + Streams.defer(Arrays.asList("1", "2", "3", "4", "5")) + .env(this.environment) + .get() + .map(Integer::parseInt) + .mapMany(integer -> testGateway.multiply(integer)) + .collect() + .consume(integers -> ...) + .flush(); + + + Another example is a simple callback scenario: + promise = service.process(myOrder); + +promise.consume(new Consumer() { + @Override + public void accept(Invoice invoice) { + handleInvoice(invoice); + } +}) +.flush();]]> + The calling thread continues, with handleInvoice() being called when the flow completes. +
Gateway behavior when no response arrives diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 6ed8e5a15c..b3ed054d4b 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -9,6 +9,16 @@ in more details, please see the Issue Tracker tickets that were resolved as part of the 4.1 development process. +
+ New Components +
+ Promise<?> Gateway + + A Reactor Promise return type is now supported for Messaging Gateway methods. + See . + +
+
General Changes