From 1ae05c325fc637a0dd1a2fa05be79c3904640ba4 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 28 Apr 2016 19:16:51 -0400 Subject: [PATCH] Upgrade to the latest Reactor --- build.gradle | 2 - .../integration/channel/ReactiveChannel.java | 72 ++++++++++++++++--- .../endpoint/ReactiveEndpoint.java | 10 +-- .../gateway/GatewayProxyFactoryBean.java | 2 +- .../config/xml/GatewayParserTests.java | 8 +-- .../gateway/AsyncGatewayTests.java | 34 ++++----- .../integration/gateway/TestService.java | 6 +- 7 files changed, 91 insertions(+), 43 deletions(-) diff --git a/build.gradle b/build.gradle index f1c2f5e353..40088343c9 100644 --- a/build.gradle +++ b/build.gradle @@ -305,8 +305,6 @@ project('spring-integration-core') { testCompile ("org.aspectj:aspectjweaver:$aspectjVersion") // testCompile ("net.openhft:chronicle:$chronicleVersion") // testCompile ("io.projectreactor:reactor-chronicle:$reactorVersion") - - testCompile "io.projectreactor:reactor-stream:$reactorVersion" } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java index 7b44f5c9db..4fe09f41ad 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * 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. @@ -19,12 +19,17 @@ package org.springframework.integration.channel; import org.reactivestreams.Processor; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; -import reactor.core.publisher.ProcessorGroup; -import reactor.core.subscriber.ReactiveSession; +import reactor.core.flow.Cancellation; +import reactor.core.publisher.EmitterProcessor; +import reactor.core.scheduler.Scheduler; +import reactor.core.subscriber.BaseSubscriber; +import reactor.core.subscriber.SignalEmitter; /** * @author Artem Bilan @@ -34,15 +39,33 @@ public class ReactiveChannel implements MessageChannel, Publisher> { private final Processor, Message> processor; - private final ReactiveSession> reactiveSession; + private final SignalEmitter> emitter; public ReactiveChannel() { - this(ProcessorGroup.>sync().get()); + this(EmitterProcessor.async(SyncScheduler.INSTANCE)); } public ReactiveChannel(Processor, Message> processor) { + Assert.notNull(processor, "'processor' must not be null"); this.processor = processor; - this.reactiveSession = ReactiveSession.create(processor); + this.emitter = SignalEmitter.create(processor); + } + + Subscriber> asSubscriber() { + return new BaseSubscriber>() { + + @Override + public void onSubscribe(Subscription subscription) { + Assert.notNull(subscription, "'subscription' must not be null"); + subscription.request(Long.MAX_VALUE); + } + + @Override + public void onNext(Message message) { + send(message); + } + + }; } @Override @@ -51,10 +74,8 @@ public class ReactiveChannel implements MessageChannel, Publisher> { } @Override - @SuppressWarnings("unchecked") public boolean send(Message message, long timeout) { - this.reactiveSession.submit(message, timeout); - return true; + return this.emitter.submit(message, timeout) > -1; } @Override @@ -62,4 +83,37 @@ public class ReactiveChannel implements MessageChannel, Publisher> { this.processor.subscribe(subscriber); } + + private static final class SyncScheduler implements Scheduler { + + private final static Scheduler INSTANCE = new SyncScheduler(); + + private final Worker worker = new Worker() { + + @Override + public Cancellation schedule(Runnable task) { + task.run(); + return () -> { + }; + } + + @Override + public void shutdown() { + + } + + }; + + @Override + public Cancellation schedule(Runnable task) { + return this.worker.schedule(task); + } + + @Override + public Worker createWorker() { + return this.worker; + } + + } + } 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 index b6e5bd994e..ad2e863263 100644 --- 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 @@ -23,7 +23,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; -import reactor.core.subscriber.ReactiveSession; +import reactor.core.subscriber.SignalEmitter; /** @@ -36,7 +36,7 @@ public class ReactiveEndpoint extends AbstractEndpoint { private final Subscriber> subscriber; - private ReactiveSession> reactiveSession; + private SignalEmitter> emitter; @SuppressWarnings("unchecked") public ReactiveEndpoint(MessageChannel inputChannel, Subscriber> subscriber) { @@ -54,13 +54,13 @@ public class ReactiveEndpoint extends AbstractEndpoint { @Override protected void doStart() { - this.reactiveSession = ReactiveSession.create(this.subscriber); - this.inputChannel.subscribe(this.reactiveSession); + this.emitter = SignalEmitter.create(this.subscriber); + this.inputChannel.subscribe(this.emitter); } @Override protected void doStop() { - this.reactiveSession.finish(); + this.emitter.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 b14913a17a..606b33940a 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 @@ -400,7 +400,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } } } - if (reactorPresent && Publisher.class.isAssignableFrom(returnType)) { + if (reactorPresent && Mono.class.isAssignableFrom(returnType)) { return Mono.fromCallable(new AsyncInvocationTask(invocation)); } return this.doInvoke(invocation, true); 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 78bcfa612b..b84799eba9 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 @@ -25,6 +25,7 @@ import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.time.Duration; import java.util.Map; import java.util.concurrent.Callable; import java.util.concurrent.CompletableFuture; @@ -36,7 +37,6 @@ 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; @@ -65,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.core.publisher.Mono; /** * @author Mark Fisher @@ -174,8 +174,8 @@ public class GatewayParserTests { MessageChannel replyChannel = context.getBean("replyChannel", MessageChannel.class); this.startResponder(requestChannel, replyChannel); TestService service = context.getBean("promise", TestService.class); - Publisher> result = service.promise("foo"); - Message reply = Promise.from(result).await(1, TimeUnit.SECONDS); + Mono> result = service.promise("foo"); + Message reply = result.get(Duration.ofSeconds(1)); assertEquals("foo", reply.getPayload()); assertNotNull(TestUtils.getPropertyValue(context.getBean("&promise"), "asyncExecutor")); } 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 8dfae0d6ef..cc25bc58a1 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 @@ -23,6 +23,7 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; +import java.time.Duration; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @@ -31,7 +32,6 @@ 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,8 +46,7 @@ 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; +import reactor.core.publisher.Mono; /** * @author Mark Fisher @@ -251,8 +250,8 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Publisher> promise = service.returnMessagePromise("foo"); - Object result = Promise.from(promise).await(10, TimeUnit.SECONDS); + Mono> promise = service.returnMessagePromise("foo"); + Object result = promise.get(Duration.ofSeconds(10)); assertEquals("foobar", ((Message) result).getPayload()); } @@ -267,8 +266,8 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Publisher promise = service.returnStringPromise("foo"); - Object result = Promise.from(promise).await(10, TimeUnit.SECONDS); + Mono promise = service.returnStringPromise("foo"); + Object result = promise.get(Duration.ofSeconds(10)); assertEquals("foobar", result); } @@ -283,8 +282,8 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Publisher promise = service.returnSomethingPromise("foo"); - Object result = Promise.from(promise).await(10, TimeUnit.SECONDS); + Mono promise = service.returnSomethingPromise("foo"); + Object result = promise.get(Duration.ofSeconds(10)); assertNotNull(result); assertEquals("foobar", result); } @@ -300,17 +299,14 @@ public class AsyncGatewayTests { proxyFactory.setBeanName("testGateway"); proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); - Publisher promise = service.returnStringPromise("foo"); + Mono promise = service.returnStringPromise("foo"); final AtomicReference result = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); - Promise.from(promise).doOnSuccess(new Consumer() { - @Override - public void accept(String s) { - result.set(s); - latch.countDown(); - } + promise.subscribe(s -> { + result.set(s); + latch.countDown(); }); latch.await(10, TimeUnit.SECONDS); @@ -364,11 +360,11 @@ public class AsyncGatewayTests { @Gateway(headers = @GatewayHeader(name = "method", expression = "#gatewayMethod.name")) Future returnCustomFutureWithTypeFuture(String s); - Publisher returnStringPromise(String s); + Mono returnStringPromise(String s); - Publisher> returnMessagePromise(String s); + Mono> returnMessagePromise(String s); - Publisher returnSomethingPromise(String s); + Mono 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 0b0e64838d..55569ffa19 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.core.publisher.Mono; + /** * @author Mark Fisher * @author Oleg Zhurakousky @@ -52,7 +52,7 @@ public interface TestService { Future> async(String s); - Publisher> promise(String s); + Mono> promise(String s); CompletableFuture completable(String s);