INT-3937: Deprecate Reactor Environment usage

artembilan · artembilan · commit b5745af148dd · 2016-03-28T11:38:06.000-04:00
JIRA: https://jira.spring.io/browse/INT-3937
diff --git a/build.gradle b/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")
 	}
 }
 
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
@@ -135,7 +135,9 @@
 	 * <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 "";
 
 }
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
@@ -99,7 +99,6 @@ public BeanDefinitionHolder parse(Map<String, Object> gatewayAttributes) {
 		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 BeanDefinitionHolder parse(Map<String, Object> gatewayAttributes) {
 				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 BeanDefinitionHolder parse(Map<String, Object> gatewayAttributes) {
 		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);
 		}
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
@@ -73,7 +73,6 @@ public BeanDefinition parse(final Element element, ParserContext parserContext)
 		}
 
 		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",
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
@@ -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 void setGlobalMethodMetadata(GatewayMethodMetadata globalMethodMetadata)
 	 * 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 @@ else if (Future.class.isAssignableFrom(returnType)) {
 			}
 		}
 		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);
diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd b/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>
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
@@ -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>
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
@@ -128,18 +128,17 @@
 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
  * @author Gary Russell
  * @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,17 +605,13 @@ public void testBridgeAnnotations() {
 		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.just("1", "2", "3", "4", "5")
-				.dispatchOn(this.environment)
 				.map(Integer::parseInt)
 				.flatMap(this.testGateway::multiply)
 				.toList()
@@ -911,7 +906,6 @@ public Integer getInvoked() {
 	@EnableMessageHistory("${message.history.tracked.components}")
 	@EnablePublisher("publishedChannel")
 	@EnableAsync
-	@EnableReactor
 	public static class ContextConfiguration2 {
 
 		/*
@@ -1332,7 +1326,7 @@ public interface TestGateway2 {
 
 	@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 {
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
@@ -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 @@
 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.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.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 @@
  */
 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 void promiseWithMessageReturned() throws Exception {
 		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 void promiseWithPayloadReturned() throws Exception {
 		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 void promiseWithWildcardReturned() throws Exception {
 		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 void promiseWithConsumer() throws Exception {
 		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 void accept(String s) {
 		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() {
 
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests.java b/spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests.java
@@ -61,7 +61,6 @@
 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 Object getNextPath(Message<?> requestMessage, Object reply) {
 	}
 
 	@Configuration
-	@EnableReactor
 	@EnableIntegration
 	public static class RoutingSlipConfiguration {
 
diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java b/spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java
@@ -66,9 +66,11 @@ public static Object getPropertyValue(Object root, String propertyPath) {
 			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 void registerBean(String beanName, Object bean) {
 	 * @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 void handleError(Throwable t) {
 		}
 
 		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);
 		}
 
 	}
+
 }
diff --git a/src/reference/asciidoc/gateway.adoc b/src/reference/asciidoc/gateway.adoc

Original file line number	Diff line number	Diff line change
`@@ -317,7 +317,6 @@ project('spring-integration-core') {`
`317`	`317`
`318`	`318`	`testCompile ("org.aspectj:aspectjweaver:$aspectjVersion")`
`319`	`319`	`testCompile ("net.openhft:chronicle:$chronicleVersion")`
`320`		`- testCompile ("io.projectreactor.spring:reactor-spring-context:$reactorVersion")`
`321`	`320`	`}`
`322`	`321`	`}`
`323`	`322`
Original file line number	Diff line number	Diff line change
`@@ -135,7 +135,9 @@`
`135`	`135`	`* <p> This attribute is required in case of {@link reactor.rx.Promise} usage.`
`136`	`136`	`* @return the suggested reactor Environment bean name.`
`137`	`137`	`* @since 4.1`
	`138`	`+ * @deprecated with no-op in favor of global JVM-wide Reactor configuration.`
`138`	`139`	`*/`
	`140`	`+ @Deprecated`
`139`	`141`	`String reactorEnvironment() default "";`
`140`	`142`
`141`	`143`	`}`
Original file line number	Diff line number	Diff line change
`@@ -73,7 +73,6 @@ public BeanDefinition parse(final Element element, ParserContext parserContext)`
`73`	`73`	`}`
`74`	`74`
`75`	`75`	`gatewayAttributes.put("mapper", element.getAttribute("mapper"));`
`76`		`- gatewayAttributes.put("reactorEnvironment", element.getAttribute("reactor-environment"));`
`77`	`76`	`gatewayAttributes.put("defaultReplyTimeout",`
`78`	`77`	`element.getAttribute(isNested ? "reply-timeout" : "default-reply-timeout"));`
`79`	`78`	`gatewayAttributes.put("defaultRequestTimeout",`