Upgrade to the latest Reactor

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"
 	}
 }
 

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<Message<?>> {
 
 	private final Processor<Message<?>, Message<?>> processor;
 
-	private final ReactiveSession<Message<?>> reactiveSession;
+	private final SignalEmitter<Message<?>> emitter;
 
 	public ReactiveChannel() {
-		this(ProcessorGroup.<Message<?>>sync().get());
+		this(EmitterProcessor.async(SyncScheduler.INSTANCE));
 	}
 
 	public ReactiveChannel(Processor<Message<?>, Message<?>> processor) {
+		Assert.notNull(processor, "'processor' must not be null");
 		this.processor = processor;
-		this.reactiveSession = ReactiveSession.create(processor);
+		this.emitter = SignalEmitter.create(processor);
+	}
+
+	Subscriber<Message<?>> asSubscriber() {
+		return new BaseSubscriber<Message<?>>() {
+
+			@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<Message<?>> {
 	}
 
 	@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<Message<?>> {
 		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;
+		}
+
+	}
+
 }

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<Message<?>> subscriber;
 
-	private ReactiveSession<Message<?>> reactiveSession;
+	private SignalEmitter<Message<?>> emitter;
 
 	@SuppressWarnings("unchecked")
 	public ReactiveEndpoint(MessageChannel inputChannel, Subscriber<Message<?>> 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();
 	}
 
 }

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);

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<Message<?>> result = service.promise("foo");
-		Message<?> reply = Promise.from(result).await(1, TimeUnit.SECONDS);
+		Mono<Message<?>> result = service.promise("foo");
+		Message<?> reply = result.get(Duration.ofSeconds(1));
 		assertEquals("foo", reply.getPayload());
 		assertNotNull(TestUtils.getPropertyValue(context.getBean("&promise"), "asyncExecutor"));
 	}

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<Message<?>> promise = service.returnMessagePromise("foo");
-		Object result = Promise.from(promise).await(10, TimeUnit.SECONDS);
+		Mono<Message<?>> 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<String> promise = service.returnStringPromise("foo");
-		Object result = Promise.from(promise).await(10, TimeUnit.SECONDS);
+		Mono<String> 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<String> promise = service.returnStringPromise("foo");
+		Mono<String> promise = service.returnStringPromise("foo");
 
 		final AtomicReference<String> result = new AtomicReference<String>();
 		final CountDownLatch latch = new CountDownLatch(1);
 
-		Promise.from(promise).doOnSuccess(new Consumer<String>() {
-			@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<String> returnStringPromise(String s);
+		Mono<String> returnStringPromise(String s);
 
-		Publisher<Message<?>> returnMessagePromise(String s);
+		Mono<Message<?>> returnMessagePromise(String s);
 
-		Publisher<?> returnSomethingPromise(String s);
+		Mono<?> returnSomethingPromise(String s);
 
 	}
 

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<Message<?>> async(String s);
 
-	Publisher<Message<?>> promise(String s);
+	Mono<Message<?>> promise(String s);
 
 	CompletableFuture<String> completable(String s);