INT-3937: Deprecate Reactor Environment usage

JIRA: https://jira.spring.io/browse/INT-3937

build.gradle

@@ -317,7 +317,6 @@ project('spring-integration-core') {
 
 		testCompile ("org.aspectj:aspectjweaver:$aspectjVersion")
 		testCompile ("net.openhft:chronicle:$chronicleVersion")
-		testCompile ("io.projectreactor.spring:reactor-spring-context:$reactorVersion")
 	}
 }
 

spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java

@@ -135,7 +135,9 @@ public @interface MessagingGateway {
 	 * <p> This attribute is required in case of {@link reactor.rx.Promise} usage.
 	 * @return the suggested reactor Environment bean name.
 	 * @since 4.1
+	 * @deprecated with no-op in favor of global JVM-wide Reactor configuration.
 	 */
+	@Deprecated
 	String reactorEnvironment() default "";
 
 }

spring-integration-core/src/main/java/org/springframework/integration/config/MessagingGatewayRegistrar.java

@@ -99,7 +99,6 @@ 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);
@@ -125,7 +124,7 @@ public class MessagingGatewayRegistrar implements ImportBeanDefinitionRegistrar,
 				String headerExpression = (String) header.get("expression");
 				boolean hasValue = StringUtils.hasText(headerValue);
 
-				if (!(hasValue ^ StringUtils.hasText(headerExpression))) {
+				if (hasValue == StringUtils.hasText(headerExpression)) {
 					throw new BeanDefinitionStoreException("exactly one of 'value' or 'expression' " +
 							"is required on a gateway's header.");
 				}
@@ -157,9 +156,6 @@ public class MessagingGatewayRegistrar implements ImportBeanDefinitionRegistrar,
 		else if (StringUtils.hasText(asyncExecutor)) {
 			gatewayProxyBuilder.addPropertyReference("asyncExecutor", asyncExecutor);
 		}
-		if (StringUtils.hasText(reactorEnvironment)) {
-			gatewayProxyBuilder.addPropertyReference("reactorEnvironment", reactorEnvironment);
-		}
 		if (StringUtils.hasText(mapper)) {
 			gatewayProxyBuilder.addPropertyReference("mapper", mapper);
 		}

spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java

@@ -73,7 +73,6 @@ public class GatewayParser implements BeanDefinitionParser {
 		}
 
 		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",

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

@@ -118,8 +118,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 
 	private volatile Class<?> asyncSubmitListenableType;
 
-	private volatile Object reactorEnvironment;
-
 	private volatile boolean initialized;
 
 	private final Object initializationMonitor = new Object();
@@ -256,12 +254,11 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	 * service interface).
 	 * @param reactorEnvironment the Reactor Environment.
 	 * @since 4.1
+	 * @deprecated with no-op in favor of global JVM-wide Reactor configuration.
 	 */
+	@Deprecated
 	public void setReactorEnvironment(Object reactorEnvironment) {
-		if (!Environment.class.getName().equals(reactorEnvironment.getClass().getName())) {
-			throw new IllegalArgumentException("The 'reactorEnvironment' must be instance of 'reactor.Environment'");
-		}
-		this.reactorEnvironment = reactorEnvironment;
+
 	}
 
 	@Override
@@ -364,10 +361,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 			}
 		}
 		if (reactorPresent && Promise.class.isAssignableFrom(returnType)) {
-			if (this.reactorEnvironment == null) {
-				throw new IllegalStateException("'reactorEnvironment' is required in case of 'Promise' return type.");
-			}
-			return Promises.<Object>task((Environment) this.reactorEnvironment,
+			return Promises.<Object>task(Environment.initializeIfEmpty(),
 					reactor.fn.Functions.supplier(new AsyncInvocationTask(invocation)));
 		}
 		return this.doInvoke(invocation, true);

spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd

@@ -789,14 +789,13 @@
 			<xsd:attribute name="reactor-environment" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation>
-						<![CDATA[
+					DEPRECATED with no-op in favor of global JVM-wide Reactor configuration.
 					Provide a reference to 'reactor.core.Environment'
 					to use for any of the interface methods that have a 'reactor.core.composable.Promise' return type.
 					The Reactor's Environment will only be used for those async methods; the sync methods
 					will be invoked in the caller's thread.
 					This attribute is required if any 'service-interface' methods
 					have a 'reactor.core.composable.Promise' return type.
-							]]>
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>

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

@@ -49,8 +49,7 @@
 	<gateway id="promise"
 			 service-interface="org.springframework.integration.gateway.TestService"
 			 default-request-channel="requestChannel"
-			 default-reply-channel="replyChannel"
-			 reactor-environment="reactorEnvironment"/>
+			 default-reply-channel="replyChannel"/>
 
 	<gateway id="asyncCompletable"
 			 service-interface="org.springframework.integration.gateway.TestService"
@@ -83,7 +82,7 @@
 			 async-executor="testExecutor">
 		<default-header name="baz" value="qux"/>
 		<method name="oneWay" request-channel="otherRequestChannel"
-							  request-timeout="456" 
+							  request-timeout="456"
 							  reply-timeout="123"
 							  payload-expression="'fiz'"
 							  reply-channel="foo">
@@ -96,6 +95,4 @@
 
 	<beans:bean id="testExecutor" class="org.springframework.integration.config.xml.GatewayParserTests$TestExecutor"/>
 
-	<beans:bean id="reactorEnvironment" class="reactor.Environment" destroy-method="shutdown"/>
-
 </beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java

@@ -128,10 +128,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.test.context.support.AnnotationConfigContextLoader;
 import org.springframework.util.MultiValueMap;
 
-import reactor.Environment;
 import reactor.rx.Promise;
 import reactor.rx.Streams;
-import reactor.spring.context.config.EnableReactor;
 
 /**
  * @author Artem Bilan
@@ -139,7 +137,8 @@ import reactor.spring.context.config.EnableReactor;
  * @since 4.0
  */
 @ContextConfiguration(loader = AnnotationConfigContextLoader.class,
-		classes = {EnableIntegrationTests.ContextConfiguration.class, EnableIntegrationTests.ContextConfiguration2.class})
+		classes = {EnableIntegrationTests.ContextConfiguration.class,
+				EnableIntegrationTests.ContextConfiguration2.class})
 @RunWith(SpringJUnit4ClassRunner.class)
 @DirtiesContext
 public class EnableIntegrationTests {
@@ -606,9 +605,6 @@ public class EnableIntegrationTests {
 		assertNull(replyChannel.receive(10));
 	}
 
-	@Autowired
-	private Environment environment;
-
 	@Test
 	public void testPromiseGateway() throws Exception {
 
@@ -616,7 +612,6 @@ public class EnableIntegrationTests {
 		final CountDownLatch consumeLatch = new CountDownLatch(1);
 
 		Streams.just("1", "2", "3", "4", "5")
-				.dispatchOn(this.environment)
 				.map(Integer::parseInt)
 				.flatMap(this.testGateway::multiply)
 				.toList()
@@ -911,7 +906,6 @@ public class EnableIntegrationTests {
 	@EnableMessageHistory("${message.history.tracked.components}")
 	@EnablePublisher("publishedChannel")
 	@EnableAsync
-	@EnableReactor
 	public static class ContextConfiguration2 {
 
 		/*
@@ -1332,7 +1326,7 @@ public class EnableIntegrationTests {
 
 	@Target({ElementType.TYPE, ElementType.ANNOTATION_TYPE})
 	@Retention(RetentionPolicy.RUNTIME)
-	@MessagingGateway(defaultRequestChannel = "gatewayChannel", reactorEnvironment = "reactorEnv",
+	@MessagingGateway(defaultRequestChannel = "gatewayChannel",
 			defaultRequestTimeout = "${default.request.timeout:12300}", defaultReplyTimeout = "#{13400}",
 			defaultHeaders = @GatewayHeader(name = "foo", value = "FOO"))
 	public @interface TestMessagingGateway {

spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -19,7 +19,6 @@ package org.springframework.integration.gateway;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.mock;
@@ -31,11 +30,7 @@ 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 reactor.Environment;
-import reactor.fn.Consumer;
-import reactor.rx.Promise;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.annotation.Gateway;
@@ -50,6 +45,9 @@ import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.ListenableFutureCallback;
 
+import reactor.fn.Consumer;
+import reactor.rx.Promise;
+
 /**
  * @author Mark Fisher
  * @author Oleg Zhurakousky
@@ -59,15 +57,9 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
  */
 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();
@@ -255,7 +247,6 @@ public class AsyncGatewayTests {
 		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");
@@ -271,7 +262,6 @@ public class AsyncGatewayTests {
 		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();
@@ -289,7 +279,6 @@ public class AsyncGatewayTests {
 		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");
@@ -307,7 +296,6 @@ public class AsyncGatewayTests {
 		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");
@@ -327,24 +315,6 @@ public class AsyncGatewayTests {
 		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() {
 

spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests.java

@@ -61,7 +61,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.test.context.support.GenericXmlContextLoader;
 
 import reactor.rx.Streams;
-import reactor.spring.context.config.EnableReactor;
 
 /**
  * @author Artem Bilan
@@ -174,7 +173,6 @@ public class RoutingSlipTests {
 	}
 
 	@Configuration
-	@EnableReactor
 	@EnableIntegration
 	public static class RoutingSlipConfiguration {
 

spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java

@@ -66,9 +66,11 @@ public abstract class TestUtils {
 			value = accessor.getPropertyValue(tokens[i]);
 			if (value != null) {
 				accessor = new DirectFieldAccessor(value);
-			} else if (i == tokens.length - 1) {
+			}
+			else if (i == tokens.length - 1) {
 				return null;
-			} else {
+			}
+			else {
 				throw new IllegalArgumentException(
 						"intermediate property '" + tokens[i] + "' is null");
 			}
@@ -196,14 +198,15 @@ public abstract class TestUtils {
 	 * @param startingIndex the index to start scanning
 	 * @return the properties provided by the named component or null if none available
 	 */
-	public static Properties locateComponentInHistory(List<Properties> history, String componentName, int startingIndex) {
+	public static Properties locateComponentInHistory(List<Properties> history, String componentName,
+			int startingIndex) {
 		Assert.notNull(history, "'history' must not be null");
 		Assert.isTrue(StringUtils.hasText(componentName), "'componentName' must be provided");
 		Assert.isTrue(startingIndex < history.size(), "'startingIndex' can not be greater then size of history");
 		Properties component = null;
 		for (int i = startingIndex; i < history.size(); i++) {
 			Properties properties = history.get(i);
-			if (componentName.equals(properties.get("name"))){
+			if (componentName.equals(properties.get("name"))) {
 				component = properties;
 				break;
 			}
@@ -262,17 +265,22 @@ public abstract class TestUtils {
 		}
 
 		private MessageChannel resolveErrorChannel(Throwable t) {
-			Message<?> failedMessage = (t instanceof MessagingException) ?
-					((MessagingException) t).getFailedMessage() : null;
-			Object errorChannelHeader = failedMessage.getHeaders().getErrorChannel();
-			if (errorChannelHeader instanceof MessageChannel) {
-				return (MessageChannel) errorChannelHeader;
+			if (t instanceof MessagingException) {
+				Message<?> failedMessage = ((MessagingException) t).getFailedMessage();
+				Object errorChannelHeader = failedMessage.getHeaders().getErrorChannel();
+				if (errorChannelHeader instanceof MessageChannel) {
+					return (MessageChannel) errorChannelHeader;
+				}
+				Assert.isInstanceOf(String.class, errorChannelHeader, "Unsupported error channel header type. " +
+						"Expected MessageChannel or String, but actual type is [" +
+						errorChannelHeader.getClass() + "]");
+				return this.context.getBean((String) errorChannelHeader, MessageChannel.class);
+			}
+			else {
+				return null;
 			}
-			Assert.isInstanceOf(String.class, errorChannelHeader,
-					"Unsupported error channel header type. Expected MessageChannel or String, but actual type is [" +
-					errorChannelHeader.getClass() + "]");
-			return this.context.getBean((String) errorChannelHeader, MessageChannel.class);
 		}
 
 	}
+
 }

src/reference/asciidoc/gateway.adoc

@@ -584,9 +584,8 @@ String out = result.get(10, TimeUnit.SECONDS);
 ===== Reactor Promise
 
 Starting with _version 4.1_, the `GatewayProxyFactoryBean` allows the use of a `Reactor` with gateway interface methods, utilizing a https://github.com/reactor/reactor/wiki/Promises[`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).
+The internal `AsyncInvocationTask` is wrapped in a `reactor.function.Supplier`, using a default `RingBufferDispatcher` for the `Promise` consumption.
+Only methods with the `Promise<?>` return type are 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.
 
@@ -599,7 +598,7 @@ For example:
 
 [source,java]
 ----
-@MessagingGateway(reactorEnvironment = "reactorEnv")
+@MessagingGateway
 public static interface TestGateway {
 
 	@Gateway(requestChannel = "promiseChannel")
@@ -617,7 +616,6 @@ public static interface TestGateway {
 		...
 
 	Streams.defer(Arrays.asList("1", "2", "3", "4", "5"))
-			.env(this.environment)
 			.get()
 			.map(Integer::parseInt)
 			.mapMany(integer -> testGateway.multiply(integer))