SI-5.0: ReactiveMessageChannel & ReactiveEndpoint

* Add `ReactiveMessageChannel` implementation based on the provided `Processor<?>` for internal dispatching logic.
Implement the `Publisher<?>` contract to allow downstream `Subscriber<?>`
* Add `ReactiveEndpoint` implementation to deal with the `ReactiveMessageChannel` through the Reactive Streams contract.
* Add appropriate `ReactiveEndpoint` building logic to the `ConsumerEndpointFactoryBean` and `AbstractMethodAnnotationPostProcessor`
* Rework Gateway to use `Publisher<?>` contract instead of `Promise` implementation.

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

gradle.properties

@@ -1,2 +1,2 @@
-version=4.3.2.BUILD-SNAPSHOT
+version=5.0.0.BUILD-SNAPSHOT
 org.gradle.daemon=true

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.
-	 * <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 "";
-
 }

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<Message<?>> {
+
+	private final Processor<Message<?>, Message<?>> processor;
+
+	private final ReactiveSession<Message<?>> reactiveSession;
+
+	public ReactiveMessageChannel() {
+		this(Broadcaster.passthrough());
+	}
+
+	public ReactiveMessageChannel(Processor<Message<?>, 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<? super Message<?>> subscriber) {
+		this.processor.subscribe(subscriber);
+	}
+
+}

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<Message<?>> publisher = (Publisher<Message<?>>) channel;
+				Subscriber<Message<?>> subscriber;
+				if (this.handler instanceof Subscriber) {
+					subscriber = (Subscriber<Message<?>>) 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() + "]");
 			}

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<T extends Annotation
 			Poller[] pollers = MessagingAnnotationUtils.resolveAttribute(annotations, "poller", Poller[].class);
 			Assert.state(ObjectUtils.isEmpty(pollers), "A '@Poller' should not be specified for Annotation-based " +
 					"endpoint, since '" + inputChannel + "' is a SubscribableChannel (not pollable).");
-			endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);
+			if (inputChannel instanceof Publisher) {
+				Publisher<Message<?>> publisher = (Publisher<Message<?>>) inputChannel;
+				Subscriber<Message<?>> subscriber;
+				if (handler instanceof Subscriber) {
+					subscriber = (Subscriber<Message<?>>) handler;
+				}
+				else {
+					//TODO errorConsumer, completeConsumer
+					subscriber = SubscriberFactory.consumer(handler::handleMessage);
+				}
+				endpoint = new ReactiveEndpoint(publisher, subscriber);
+			}
+			else {
+				endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);
+			}
 		}
 		return endpoint;
 	}

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<Message<?>> inputChannel;
+
+	private final Subscriber<Message<?>> subscriber;
+
+	private ReactiveSession<Message<?>> reactiveSession;
+
+	public ReactiveEndpoint(Publisher<Message<?>> inputChannel, Subscriber<Message<?>> 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();
+	}
+
+}

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.<Object>task(Environment.initializeIfEmpty(),
-					reactor.fn.Functions.supplier(new AsyncInvocationTask(invocation)));
+		if (reactorPresent && Publisher.class.isAssignableFrom(returnType)) {
+			return Promises.<Object>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();

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<String>("foo"));
 		assertTrue(latch4.await(1000, TimeUnit.MILLISECONDS));
 	}
-
+*/
 }

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;
+		}
+
+	}
+
+}

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

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<Integer> multiply(Integer value);
+		org.reactivestreams.Publisher<Integer> 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 {
 

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<Message<?>> promise = service.returnMessagePromise("foo");
-		Object result = promise.await(10, TimeUnit.SECONDS);
+		Publisher<Message<?>> 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<String> promise = service.returnStringPromise("foo");
-		Object result = promise.await(10, TimeUnit.SECONDS);
+		Publisher<String> 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<String> promise = service.returnStringPromise("foo");
+		Publisher<String> promise = service.returnStringPromise("foo");
 
 		final AtomicReference<String> result = new AtomicReference<String>();
 		final CountDownLatch latch = new CountDownLatch(1);
 
-		promise.onSuccess(new Consumer<String>() {
+		Promises.from(promise).onSuccess(new Consumer<String>() {
 			@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<String> returnStringPromise(String s);
+		Publisher<String> returnStringPromise(String s);
 
-		Promise<Message<?>> returnMessagePromise(String s);
+		Publisher<Message<?>> returnMessagePromise(String s);
 
-		Promise<?> returnSomethingPromise(String s);
+		Publisher<?> 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.rx.Promise;
-
 /**
  * @author Mark Fisher
  * @author Oleg Zhurakousky
@@ -52,7 +52,7 @@ public interface TestService {
 
 	Future<Message<?>> async(String s);
 
-	Promise<Message<?>> promise(String s);
+	Publisher<Message<?>> promise(String s);
 
 	CompletableFuture<String> completable(String s);