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

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

@@ -1,8 +1,8 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans:beans xmlns="http://www.springframework.org/schema/integration"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:beans="http://www.springframework.org/schema/beans"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+			 xmlns:beans="http://www.springframework.org/schema/beans"
+			 xsi:schemaLocation="http://www.springframework.org/schema/beans
 			http://www.springframework.org/schema/beans/spring-beans.xsd
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration.xsd">
@@ -16,29 +16,37 @@
 	</channel>
 
 	<gateway id="oneWay"
-	         service-interface="org.springframework.integration.gateway.TestService"
-	         default-request-channel="requestChannel"/>
+			 service-interface="org.springframework.integration.gateway.TestService"
+			 default-request-channel="requestChannel"/>
 
 	<gateway id="solicitResponse"
-	         service-interface="org.springframework.integration.gateway.TestService"
-	         default-reply-channel="replyChannel"
-	         default-reply-timeout="3000"/>
+			 service-interface="org.springframework.integration.gateway.TestService"
+			 default-reply-channel="replyChannel"
+			 default-reply-timeout="3000"/>
 
 	<gateway id="requestReply"
-	         service-interface="org.springframework.integration.gateway.TestService"
-	         default-request-channel="requestChannel"
-	         default-reply-channel="replyChannel"
-	         default-reply-timeout="5000"/>
+			 service-interface="org.springframework.integration.gateway.TestService"
+			 default-request-channel="requestChannel"
+			 default-reply-channel="replyChannel"
+			 default-reply-timeout="5000"/>
 
 	<gateway id="async"
 			 service-interface="org.springframework.integration.gateway.TestService"
 			 default-request-channel="requestChannel"
 			 default-reply-channel="replyChannel"
 			 async-executor="testExecutor"/>
-	
+
+	<gateway id="promise"
+			 service-interface="org.springframework.integration.gateway.TestService"
+			 default-request-channel="requestChannel"
+			 default-reply-channel="replyChannel"
+			 reactor-environment="reactorEnvironment"/>
+
 	<!-- no assertions for this. The fact that this config does not result in error is sufficient -->
 	<gateway default-request-channel="nullChannel"/>
 
 	<beans:bean id="testExecutor" class="org.springframework.integration.config.xml.GatewayParserTests$TestExecutor"/>
 
+	<beans:bean id="reactorEnvironment" class="reactor.core.Environment" destroy-method="shutdown"/>
+
 </beans:beans>

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<String>("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<Message<?>> 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);
 			}
 		}
+
 	}
 
 }

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<List<Integer>> ref = new AtomicReference<List<Integer>>();
+		final CountDownLatch consumeLatch = new CountDownLatch(1);
+
+		Streams.defer(Arrays.asList("1", "2", "3", "4", "5"))
+				.env(this.environment)
+				.get()
+				.map(new Function<String, Integer>() {
+					@Override
+					public Integer apply(String s) {
+						return Integer.parseInt(s);
+					}
+				})
+				.mapMany(new Function<Integer, Composable<Integer>>() {
+					@Override
+					public Composable<Integer> apply(Integer integer) {
+						return testGateway.multiply(integer);
+					}
+				})
+				.collect()
+				.consume(new Consumer<List<Integer>>() {
+					@Override
+					public void accept(List<Integer> integers) {
+						ref.set(integers);
+						consumeLatch.countDown();
+					}
+				})
+				.flush();
+
+
+		assertTrue(consumeLatch.await(2, TimeUnit.SECONDS));
+
+		List<Integer> integers = ref.get();
+		assertEquals(5, integers.size());
+
+		assertThat(integers, Matchers.<Integer>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<Integer> 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 {
 

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<Message<?>> 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<String> 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<String> promise = service.returnStringPromise("foo");
+		long start = System.currentTimeMillis();
+
+		final AtomicReference<String> result = new AtomicReference<String>();
+		final CountDownLatch latch = new CountDownLatch(1);
+
+		promise.consume(new Consumer<String>() {
+			@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<String> returnStringPromise(String s);
+
+		Promise<Message<?>> returnMessagePromise(String s);
+
+		Promise<?> returnSomethingPromise(String s);
+
 	}
 
 }

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<Message<?>> async(String s);
 
+	Promise<Message<?>> promise(String s);
+
 }