diff --git a/build.gradle b/build.gradle index 57c1e4a784..66a1b4e493 100644 --- 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 index 207b1f88ce..489c66cfb8 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 @@ -135,7 +135,9 @@ public @interface MessagingGateway { *

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 index 9d3fffdaae..dd39596c25 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 @@ -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); } 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 1b4337c457..bbe754681f 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 @@ -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", 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 7e59b39aef..c5641c2eb4 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 @@ -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.task((Environment) this.reactorEnvironment, + return Promises.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 index 3ac59ea820..4ea91ea4c9 100644 --- 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 @@ - diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml index 616d655612..0189e7a60f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests-context.xml @@ -49,8 +49,7 @@ + default-reply-channel="replyChannel"/> @@ -96,6 +95,4 @@ - - 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 5d84e774e8..a633ad59ca 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 @@ -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 { 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 98ea8054ea..bcdfba280f 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-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> 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 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() { 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 index 0fcb325823..718e8d5d5e 100644 --- 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.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 { 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 index 611854efe5..dce72e89b0 100644 --- 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 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 history, String componentName, int startingIndex) { + public static Properties locateComponentInHistory(List 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); } } + } diff --git a/src/reference/asciidoc/gateway.adoc b/src/reference/asciidoc/gateway.adoc index 0b40c3a104..b93ba2e7e8 100644 --- a/src/reference/asciidoc/gateway.adoc +++ b/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))