diff --git a/build.gradle b/build.gradle index 0ddeca18b6..805079343d 100644 --- a/build.gradle +++ b/build.gradle @@ -73,16 +73,10 @@ subprojects { subproject -> } compileJava { - sourceCompatibility = 1.6 - targetCompatibility = 1.6 + sourceCompatibility = 1.8 + targetCompatibility = 1.8 } - compileTestJava { - sourceCompatibility = 1.6 - targetCompatibility = 1.6 - } - - ext { activeMqVersion = '5.13.2' aspectjVersion = '1.8.9' @@ -123,7 +117,7 @@ subprojects { subproject -> openJpaVersion = '2.4.0' pahoMqttClientVersion = '1.0.2' postgresVersion = '9.1-901-1.jdbc4' - reactorVersion = '2.0.8.RELEASE' + reactorVersion = '2.5.0.BUILD-SNAPSHOT' romeToolsVersion = '1.6.0' servletApiVersion = '3.1.0' slf4jVersion = "1.7.21" @@ -289,11 +283,6 @@ project('spring-integration-amqp') { project('spring-integration-core') { description = 'Spring Integration Core' - compileTestJava { - sourceCompatibility = 1.8 - targetCompatibility = 1.8 - } - dependencies { compile "org.springframework:spring-core:$springVersion" compile "org.springframework:spring-aop:$springVersion" @@ -317,7 +306,11 @@ project('spring-integration-core') { compile("com.esotericsoftware:kryo-shaded:$kryoShadedVersion", optional) testCompile ("org.aspectj:aspectjweaver:$aspectjVersion") - testCompile ("net.openhft:chronicle:$chronicleVersion") +// testCompile ("net.openhft:chronicle:$chronicleVersion") +// testCompile ("io.projectreactor:reactor-chronicle:$reactorVersion") + + testCompile "org.testng:testng:6.8.21" + testCompile "org.reactivestreams:reactive-streams-tck:1.0.0" } } diff --git a/gradle.properties b/gradle.properties index 3f49498608..9136e989dd 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1,2 +1,2 @@ -version=4.3.2.BUILD-SNAPSHOT +version=5.0.0.BUILD-SNAPSHOT org.gradle.daemon=true 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 489c66cfb8..6f9280b611 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 @@ -127,17 +127,4 @@ public @interface MessagingGateway { */ String mapper() default ""; - /** - * Provide a reference to an {@link reactor.Environment} - * to use for any of the interface methods that have a {@link reactor.rx.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.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/channel/ReactiveMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveMessageChannel.java new file mode 100644 index 0000000000..ef521cd897 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveMessageChannel.java @@ -0,0 +1,65 @@ +/* + * Copyright 2015 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.channel; + +import org.reactivestreams.Processor; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; + +import reactor.core.subscription.ReactiveSession; +import reactor.rx.broadcast.Broadcaster; + +/** + * @author Artem Bilan + * @since 5.0 + */ +public class ReactiveMessageChannel implements MessageChannel, Publisher> { + + private final Processor, Message> processor; + + private final ReactiveSession> reactiveSession; + + public ReactiveMessageChannel() { + this(Broadcaster.passthrough()); + } + + public ReactiveMessageChannel(Processor, Message> processor) { + this.processor = processor; + this.reactiveSession = ReactiveSession.create(processor); + } + + @Override + public boolean send(Message message) { + return send(message, -1); + } + + @Override + @SuppressWarnings("unchecked") + public boolean send(Message message, long timeout) { + this.reactiveSession.submit(message, timeout); + return true; + } + + @Override + public void subscribe(Subscriber> subscriber) { + this.processor.subscribe(subscriber); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 96bd24ca1a..fa919b454e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -21,6 +21,8 @@ import java.util.List; import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; import org.springframework.aop.framework.Advised; import org.springframework.aop.framework.ProxyFactory; @@ -39,9 +41,11 @@ import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.endpoint.ReactiveEndpoint; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.PollableChannel; @@ -52,6 +56,8 @@ import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; +import reactor.core.subscriber.SubscriberFactory; + /** * @author Mark Fisher * @author Oleg Zhurakousky @@ -240,6 +246,7 @@ public class ConsumerEndpointFactoryBean return this.endpoint.getClass(); } + @SuppressWarnings("unchecked") private void initializeEndpoint() throws Exception { synchronized (this.initializationMonitor) { if (this.initialized) { @@ -282,6 +289,18 @@ public class ConsumerEndpointFactoryBean pollingConsumer.setBeanFactory(this.beanFactory); this.endpoint = pollingConsumer; } + else if (channel instanceof Publisher) { + Publisher> publisher = (Publisher>) channel; + Subscriber> subscriber; + if (this.handler instanceof Subscriber) { + subscriber = (Subscriber>) this.handler; + } + else { + //TODO errorConsumer, completeConsumer + subscriber = SubscriberFactory.consumer(this.handler::handleMessage); + } + this.endpoint = new ReactiveEndpoint(publisher, subscriber); + } else { throw new IllegalArgumentException("unsupported channel type: [" + channel.getClass() + "]"); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index dfc7496cae..32edf75988 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -26,6 +26,8 @@ import java.util.List; import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; import org.springframework.aop.TargetSource; import org.springframework.aop.framework.Advised; @@ -53,6 +55,7 @@ import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.AbstractPollingEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.endpoint.ReactiveEndpoint; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; @@ -61,6 +64,7 @@ import org.springframework.integration.router.AbstractMessageRouter; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.integration.util.MessagingAnnotationUtils; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.PollableChannel; @@ -76,6 +80,8 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; +import reactor.core.subscriber.SubscriberFactory; + /** * Base class for Method-level annotation post-processors. * @@ -311,7 +317,21 @@ public abstract class AbstractMethodAnnotationPostProcessor> publisher = (Publisher>) inputChannel; + Subscriber> subscriber; + if (handler instanceof Subscriber) { + subscriber = (Subscriber>) handler; + } + else { + //TODO errorConsumer, completeConsumer + subscriber = SubscriberFactory.consumer(handler::handleMessage); + } + endpoint = new ReactiveEndpoint(publisher, subscriber); + } + else { + endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler); + } } return endpoint; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java new file mode 100644 index 0000000000..e681c17919 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java @@ -0,0 +1,58 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.endpoint; + +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; + +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +import reactor.core.subscription.ReactiveSession; + +/** + * @author Artem Bilan + * @since 5.0 + */ +public class ReactiveEndpoint extends AbstractEndpoint { + + private final Publisher> inputChannel; + + private final Subscriber> subscriber; + + private ReactiveSession> reactiveSession; + + public ReactiveEndpoint(Publisher> inputChannel, Subscriber> subscriber) { + Assert.isInstanceOf(Publisher.class, inputChannel, + "The 'inputChannel', must implement org.reactivestreams.Publisher."); + Assert.notNull(subscriber); + this.inputChannel = inputChannel; + this.subscriber = subscriber; + } + + @Override + protected void doStart() { + this.reactiveSession = ReactiveSession.create(this.subscriber); + this.inputChannel.subscribe(this.reactiveSession); + } + + @Override + protected void doStop() { + this.reactiveSession.finish(); + } + +} 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 b7e63f47a5..d0d33ce2d4 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 @@ -29,6 +29,7 @@ import java.util.concurrent.Future; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; +import org.reactivestreams.Publisher; import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.AopUtils; @@ -65,8 +66,6 @@ import org.springframework.util.ObjectUtils; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; -import reactor.Environment; -import reactor.rx.Promise; import reactor.rx.Promises; /** @@ -288,19 +287,6 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint 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 - * @deprecated with no-op in favor of global JVM-wide Reactor configuration. - */ - @Deprecated - public void setReactorEnvironment(Object reactorEnvironment) { - - } - @Override public void setBeanClassLoader(ClassLoader beanClassLoader) { this.beanClassLoader = beanClassLoader; @@ -413,9 +399,8 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } } } - if (reactorPresent && Promise.class.isAssignableFrom(returnType)) { - return Promises.task(Environment.initializeIfEmpty(), - reactor.fn.Functions.supplier(new AsyncInvocationTask(invocation))); + if (reactorPresent && Publisher.class.isAssignableFrom(returnType)) { + return Promises.task(() -> new AsyncInvocationTask(invocation)); } return this.doInvoke(invocation, true); } @@ -653,7 +638,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint if (Future.class.isAssignableFrom(expectedReturnType)) { return (T) source; } - if (reactorPresent && Promise.class.isAssignableFrom(expectedReturnType)) { + if (reactorPresent && Publisher.class.isAssignableFrom(expectedReturnType)) { return (T) source; } if (this.getConversionService() != null) { @@ -667,7 +652,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint private static boolean hasReturnParameterizedWithMessage(Method method, boolean runningOnCallerThread) { if (!runningOnCallerThread && (Future.class.isAssignableFrom(method.getReturnType()) - || (reactorPresent && Promise.class.isAssignableFrom(method.getReturnType())))) { + || (reactorPresent && Publisher.class.isAssignableFrom(method.getReturnType())))) { Type returnType = method.getGenericReturnType(); if (returnType instanceof ParameterizedType) { Type[] typeArgs = ((ParameterizedType) returnType).getActualTypeArguments(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java index 44f63fcca5..ec3ba54ae5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java @@ -21,7 +21,6 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; -import java.io.IOException; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; @@ -38,10 +37,6 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; -import reactor.io.codec.JavaSerializationCodec; -import reactor.io.queue.PersistentQueue; -import reactor.io.queue.spec.PersistentQueueSpec; - /** * @author Mark Fisher * @author Artem Bilan @@ -257,6 +252,7 @@ public class QueueChannelTests { @Rule public final TemporaryFolder tempFolder = new TemporaryFolder(); + /*TODO: No Reactor Chronicle artifact @Test public void testReactorPersistentQueue() throws InterruptedException, IOException { final AtomicBoolean messageReceived = new AtomicBoolean(false); @@ -368,5 +364,5 @@ public class QueueChannelTests { queue.add(new GenericMessage("foo")); assertTrue(latch4.await(1000, TimeUnit.MILLISECONDS)); } - +*/ } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveMessageChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveMessageChannelTests.java new file mode 100644 index 0000000000..34cc9cd3a8 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveMessageChannelTests.java @@ -0,0 +1,85 @@ +/* + * Copyright 2015 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.channel.reactive; + +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.channel.ReactiveMessageChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import reactor.Processors; + +/** + * @author Artem Bilan + * @since 5.0 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class ReactiveMessageChannelTests { + + @Autowired + private MessageChannel reactiveChannel; + + @Test + @SuppressWarnings("unchecked") + public void testReactiveMessageChannel() throws InterruptedException { + QueueChannel replyChannel = new QueueChannel(); + + for (int i = 0; i < 10; i++) { + this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build()); + } + + for (int i = 0; i < 10; i++) { + Message receive = replyChannel.receive(10000); + assertNotNull(receive); + System.out.println("Receive: " + receive.getPayload()); + } + } + + @Configuration + @EnableIntegration + public static class TestConfiguration { + + @Bean + public MessageChannel reactiveChannel() { + return new ReactiveMessageChannel(Processors.queue()); + } + + @ServiceActivator(inputChannel = "reactiveChannel") + public String handle(int payload) { + System.out.println("CurrentThread: " + Thread.currentThread() + " for payload: " + payload); + return "" + payload; + } + + } + +} 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 2d2f507510..5ac4f2d57a 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 @@ -36,6 +36,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; import org.junit.Test; import org.junit.runner.RunWith; +import org.reactivestreams.Publisher; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanNameAware; @@ -64,7 +65,7 @@ import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import reactor.rx.Promise; +import reactor.rx.Promises; /** * @author Mark Fisher @@ -173,8 +174,8 @@ public class GatewayParserTests { 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(10, TimeUnit.SECONDS); + Publisher> result = service.promise("foo"); + Message reply = Promises.from(result).await(1, TimeUnit.SECONDS); assertEquals("foo", reply.getPayload()); assertNotNull(TestUtils.getPropertyValue(context.getBean("&promise"), "asyncExecutor")); } 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 220fa06812..6ff8522d96 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 @@ -127,7 +127,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.support.AnnotationConfigContextLoader; import org.springframework.util.MultiValueMap; -import reactor.rx.Promise; import reactor.rx.Streams; /** @@ -1335,7 +1334,7 @@ public class EnableIntegrationTests { void sendAsync(String payload); @Gateway(requestChannel = "promiseChannel") - Promise multiply(Integer value); + org.reactivestreams.Publisher multiply(Integer value); } @@ -1350,7 +1349,7 @@ public class EnableIntegrationTests { @Target({ ElementType.TYPE, ElementType.ANNOTATION_TYPE }) @Retention(RetentionPolicy.RUNTIME) @MessagingGateway(defaultRequestChannel = "gatewayChannel", - defaultRequestTimeout = "${default.request.timeout:12300}", defaultReplyTimeout = "#{13400}", + 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 bcdfba280f..810a5a8111 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 @@ -31,6 +31,7 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; +import org.reactivestreams.Publisher; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.annotation.Gateway; @@ -46,7 +47,7 @@ import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; import reactor.fn.Consumer; -import reactor.rx.Promise; +import reactor.rx.Promises; /** * @author Mark Fisher @@ -57,6 +58,7 @@ import reactor.rx.Promise; */ public class AsyncGatewayTests { + // TODO: changed from 0 because of recurrent failure: is this right? private final long safety = 100; @@ -249,8 +251,8 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Promise> promise = service.returnMessagePromise("foo"); - Object result = promise.await(10, TimeUnit.SECONDS); + Publisher> promise = service.returnMessagePromise("foo"); + Object result = Promises.from(promise).await(10, TimeUnit.SECONDS); assertEquals("foobar", ((Message) result).getPayload()); } @@ -265,8 +267,8 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Promise promise = service.returnStringPromise("foo"); - Object result = promise.await(10, TimeUnit.SECONDS); + Publisher promise = service.returnStringPromise("foo"); + Object result = Promises.from(promise).await(10, TimeUnit.SECONDS); assertEquals("foobar", result); } @@ -281,8 +283,8 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Promise promise = service.returnSomethingPromise("foo"); - Object result = promise.await(10, TimeUnit.SECONDS); + Publisher promise = service.returnSomethingPromise("foo"); + Object result = Promises.from(promise).await(10, TimeUnit.SECONDS); assertNotNull(result); assertEquals("foobar", result); } @@ -298,12 +300,12 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Promise promise = service.returnStringPromise("foo"); + Publisher promise = service.returnStringPromise("foo"); final AtomicReference result = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); - promise.onSuccess(new Consumer() { + Promises.from(promise).onSuccess(new Consumer() { @Override public void accept(String s) { result.set(s); @@ -362,11 +364,11 @@ public class AsyncGatewayTests { @Gateway(headers = @GatewayHeader(name = "method", expression = "#gatewayMethod.name")) Future returnCustomFutureWithTypeFuture(String s); - Promise returnStringPromise(String s); + Publisher returnStringPromise(String s); - Promise> returnMessagePromise(String s); + Publisher> returnMessagePromise(String s); - Promise returnSomethingPromise(String s); + Publisher 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 2147aa6e63..0b0e64838d 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 @@ -19,11 +19,11 @@ package org.springframework.integration.gateway; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Future; +import org.reactivestreams.Publisher; + import org.springframework.messaging.Message; import org.springframework.messaging.handler.annotation.Payload; -import reactor.rx.Promise; - /** * @author Mark Fisher * @author Oleg Zhurakousky @@ -52,7 +52,7 @@ public interface TestService { Future> async(String s); - Promise> promise(String s); + Publisher> promise(String s); CompletableFuture completable(String s);