Make AbstractMessageHandler as Subscriber

* Upgrade to the latest Reactor-2.5.0
* Fix all `Promise` and `Streams` mentioning to the `Mono` and `Flux`
* Add `ReactiveEndpoint.SubscribableChannelPublisherAdapter` to adapt `SubscribableChannel` into a `Publisher` for Reactive downstream

build.gradle

@@ -34,6 +34,7 @@ allprojects {
 			maven { url 'https://repo.spring.io/libs-snapshot' }
 		}
 		maven { url 'https://repo.spring.io/libs-milestone' }
+		maven { url 'https://repo.spring.io/libs-staging-local' }
 		mavenCentral()
 	}
 
@@ -114,7 +115,7 @@ subprojects { subproject ->
 		mockitoVersion = '1.10.19'
 		mysqlVersion = '5.1.34'
 		nettyVersion = '4.0.27.Final'
-		openJpaVersion = '2.4.0'
+		openJpaVersion = '2.4.1'
 		pahoMqttClientVersion = '1.0.2'
 		postgresVersion = '9.1-901-1.jdbc4'
 		reactorVersion = '2.5.0.BUILD-SNAPSHOT'

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml

@@ -3,18 +3,14 @@
 	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xmlns:si-amqp="http://www.springframework.org/schema/integration/amqp"
 	xmlns:int="http://www.springframework.org/schema/integration"
-	xmlns:rabbit="http://www.springframework.org/schema/rabbit"
 	xsi:schemaLocation="http://www.springframework.org/schema/integration/amqp http://www.springframework.org/schema/integration/amqp/spring-integration-amqp.xsd
 		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
-		http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
 		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
 
 	<int:channel id="requests"/>
 
 	<si-amqp:inbound-gateway id="gateway" request-channel="requests" queue-names="test" reply-timeout="1234"
-			connection-factory="rabbitConnectionFactory" message-converter="testConverter">
-		<reactive/>
-	</si-amqp:inbound-gateway>
+			connection-factory="rabbitConnectionFactory" message-converter="testConverter"/>
 
 	<bean id="rabbitConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
 		<constructor-arg value="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>

spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java

@@ -1,11 +1,11 @@
 /*
- * Copyright 2016 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.
  * You may obtain a copy of the License at
  *
- *       http://www.apache.org/licenses/LICENSE-2.0
+ *      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,
@@ -25,10 +25,9 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
-import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.DirectProcessor;
 import reactor.core.subscriber.BaseSubscriber;
-import reactor.core.subscriber.SignalEmitter;
-import reactor.core.util.PlatformDependent;
+import reactor.core.subscriber.SubmissionEmitter;
 
 /**
  * @author Artem Bilan
@@ -38,19 +37,19 @@ public class ReactiveChannel implements MessageChannel, Publisher<Message<?>> {
 
 	private final Processor<Message<?>, Message<?>> processor;
 
-	private final SignalEmitter<Message<?>> emitter;
+	private final SubmissionEmitter<Message<?>> emitter;
 
 	public ReactiveChannel() {
-		this(EmitterProcessor.create(PlatformDependent.SMALL_BUFFER_SIZE, Integer.MAX_VALUE, false));
+		this(DirectProcessor.create());
 	}
 
 	public ReactiveChannel(Processor<Message<?>, Message<?>> processor) {
 		Assert.notNull(processor, "'processor' must not be null");
 		this.processor = processor;
-		this.emitter = SignalEmitter.create(processor);
+		this.emitter = SubmissionEmitter.create(processor);
 	}
 
-	Subscriber<Message<?>> asSubscriber() {
+	public Subscriber<Message<?>> asSubscriber() {
 		return new BaseSubscriber<Message<?>>() {
 
 			@Override

spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java

@@ -295,7 +295,6 @@ public class ConsumerEndpointFactoryBean
 					subscriber = (Subscriber<Message<?>>) this.handler;
 				}
 				else {
-					//TODO errorConsumer, completeConsumer
 					subscriber = Subscribers.consumer(this.handler::handleMessage);
 				}
 				this.endpoint = new ReactiveEndpoint(channel, subscriber);

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java

@@ -307,7 +307,8 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 	}
 
 	@SuppressWarnings("unchecked")
-	protected AbstractEndpoint doCreateEndpoint(MessageHandler handler, MessageChannel inputChannel,List<Annotation> annotations) {
+	protected AbstractEndpoint doCreateEndpoint(MessageHandler handler, MessageChannel inputChannel,
+			List<Annotation> annotations) {
 		AbstractEndpoint endpoint;
 		if (inputChannel instanceof PollableChannel) {
 			PollingConsumer pollingConsumer = new PollingConsumer((PollableChannel) inputChannel, handler);
@@ -422,7 +423,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 		return name + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX;
 	}
 
-	protected void setOutputChannelIfPresent(List<Annotation> annotations, AbstractReplyProducingMessageHandler	handler) {
+	protected void setOutputChannelIfPresent(List<Annotation> annotations, AbstractReplyProducingMessageHandler handler) {
 		String outputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, "outputChannel", String.class);
 		if (StringUtils.hasText(outputChannelName)) {
 			handler.setOutputChannelName(outputChannelName);

spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java

@@ -18,12 +18,21 @@ package org.springframework.integration.endpoint;
 
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 
+import org.springframework.integration.channel.MessagePublishingErrorHandler;
+import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.SubscribableChannel;
 import org.springframework.util.Assert;
+import org.springframework.util.ErrorHandler;
 
-import reactor.core.subscriber.SignalEmitter;
+import reactor.core.publisher.DirectProcessor;
+import reactor.core.subscriber.SubscriberBarrier;
+import reactor.core.util.Exceptions;
 
 
 /**
@@ -32,35 +41,142 @@ import reactor.core.subscriber.SignalEmitter;
  */
 public class ReactiveEndpoint extends AbstractEndpoint {
 
-	private final Publisher<Message<?>> inputChannel;
+	private final Publisher<Message<?>> publisher;
 
-	private final Subscriber<Message<?>> subscriber;
+	private final SubscriberBarrier<Message<?>, Message<?>> subscriber;
+
+	private ErrorHandler errorHandler;
 
-	private SignalEmitter<Message<?>> emitter;
 
 	@SuppressWarnings("unchecked")
 	public ReactiveEndpoint(MessageChannel inputChannel, Subscriber<Message<?>> subscriber) {
 		Assert.notNull(inputChannel);
 		Assert.notNull(subscriber);
 		if (inputChannel instanceof Publisher) {
-			this.inputChannel = (Publisher<Message<?>>) inputChannel;
+			this.publisher = (Publisher<Message<?>>) inputChannel;
 		}
 		else {
-			//TODO: Wrap all other channels to the Publisher<?>
-			this.inputChannel = null;
+			this.publisher = adaptToPublisher(inputChannel);
+		}
+		this.subscriber = new SubscriberBarrier<Message<?>, Message<?>>(subscriber) {
+
+			@Override
+			protected void doNext(Message<?> message) {
+				try {
+					super.doNext(message);
+				}
+				catch (Exception e) {
+					Exceptions.throwIfFatal(e);
+					ReactiveEndpoint.this.errorHandler.handleError(e);
+				}
+			}
+
+		};
+	}
+
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
+	@Override
+	protected void onInit() throws Exception {
+		super.onInit();
+		if (this.errorHandler == null) {
+			Assert.notNull(getBeanFactory(), "BeanFactory is required");
+			this.errorHandler = new MessagePublishingErrorHandler(
+					new BeanFactoryChannelResolver(getBeanFactory()));
 		}
-		this.subscriber = subscriber;
 	}
 
 	@Override
 	protected void doStart() {
-		this.emitter = SignalEmitter.create(this.subscriber);
-		this.inputChannel.subscribe(this.emitter);
+		this.publisher.subscribe(this.subscriber);
 	}
 
 	@Override
 	protected void doStop() {
-		this.emitter.finish();
+		this.subscriber.cancel();
+	}
+
+	private Publisher<Message<?>> adaptToPublisher(MessageChannel inputChannel) {
+		if (inputChannel instanceof SubscribableChannel) {
+			return adaptSubscribableChannelToPublisher((SubscribableChannel) inputChannel);
+		}
+		else if (inputChannel instanceof PollableChannel) {
+			return adaptPollableChannelToPublisher((PollableChannel) inputChannel);
+		}
+		else {
+			throw new IllegalArgumentException("The 'inputChannel' must be an instance of SubscribableChannel or " +
+					"PollableChannel, not: " + inputChannel);
+		}
+	}
+
+	private Publisher<Message<?>> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) {
+		return new SubscribableChannelPublisherAdapter(inputChannel);
+	}
+
+	private Publisher<Message<?>> adaptPollableChannelToPublisher(PollableChannel inputChannel) {
+		return null;
+	}
+
+
+	private static class SubscribableChannelPublisherAdapter
+			implements Publisher<Message<?>>, Subscriber<Message<?>>, Subscription {
+
+		private final DirectProcessor<Message<?>> delegate = DirectProcessor.create();
+
+		private final MessageHandler subscriberAdapter = this.delegate.connectEmitter()::accept;
+
+		private final SubscribableChannel channel;
+
+		private Subscriber<? super Message<?>> actualSubscriber;
+
+		private Subscription actualSubscription;
+
+
+		private SubscribableChannelPublisherAdapter(SubscribableChannel channel) {
+			this.channel = channel;
+		}
+
+		@Override
+		public void subscribe(Subscriber<? super Message<?>> subscriber) {
+			this.actualSubscriber = subscriber;
+			this.delegate.subscribe(this);
+			this.channel.subscribe(this.subscriberAdapter);
+		}
+
+		@Override
+		public void onSubscribe(Subscription subscription) {
+			this.actualSubscription = subscription;
+			this.actualSubscriber.onSubscribe(this);
+		}
+
+		@Override
+		public void onNext(Message<?> message) {
+			this.actualSubscriber.onNext(message);
+		}
+
+		@Override
+		public void onError(Throwable t) {
+			this.actualSubscriber.onError(t);
+		}
+
+		@Override
+		public void onComplete() {
+			this.actualSubscriber.onComplete();
+		}
+
+		@Override
+		public void request(long n) {
+			this.actualSubscription.request(n);
+		}
+
+		@Override
+		public void cancel() {
+			this.channel.unsubscribe(this.subscriberAdapter);
+			this.actualSubscription.cancel();
+		}
+
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java

@@ -29,7 +29,6 @@ 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;
@@ -46,7 +45,6 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.support.TaskExecutorAdapter;
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.annotation.Gateway;
 import org.springframework.integration.annotation.GatewayHeader;
 import org.springframework.integration.context.IntegrationProperties;
@@ -85,11 +83,6 @@ import reactor.core.publisher.Mono;
 public class GatewayProxyFactoryBean extends AbstractEndpoint
 		implements TrackableComponent, FactoryBean<Object>, MethodInterceptor, BeanClassLoaderAware {
 
-	private static final SpelExpressionParser PARSER = new SpelExpressionParser();
-
-	private static final boolean reactorPresent = ClassUtils.isPresent("reactor.rx.Promise",
-			GatewayProxyFactoryBean.class.getClassLoader());
-
 	private volatile Class<?> serviceInterface;
 
 	private volatile MessageChannel defaultRequestChannel;
@@ -400,7 +393,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 				}
 			}
 		}
-		if (reactorPresent && Mono.class.isAssignableFrom(returnType)) {
+		if (Mono.class.isAssignableFrom(returnType)) {
 			return Mono.fromCallable(new AsyncInvocationTask(invocation));
 		}
 		return this.doInvoke(invocation, true);
@@ -530,7 +523,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 					}
 					headerExpressions.put(name, hasValue
 							? new LiteralExpression(value)
-							: PARSER.parseExpression(expression));
+							: EXPRESSION_PARSER.parseExpression(expression));
 				}
 			}
 
@@ -639,7 +632,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 		if (Future.class.isAssignableFrom(expectedReturnType)) {
 			return (T) source;
 		}
-		if (reactorPresent && Publisher.class.isAssignableFrom(expectedReturnType)) {
+		if (Mono.class.isAssignableFrom(expectedReturnType)) {
 			return (T) source;
 		}
 		if (this.getConversionService() != null) {
@@ -653,7 +646,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
 	private static boolean hasReturnParameterizedWithMessage(Method method, boolean runningOnCallerThread) {
 		if (!runningOnCallerThread &&
 				(Future.class.isAssignableFrom(method.getReturnType())
-				|| (reactorPresent && Publisher.class.isAssignableFrom(method.getReturnType())))) {
+				|| Mono.class.isAssignableFrom(method.getReturnType()))) {
 			Type returnType = method.getGenericReturnType();
 			if (returnType instanceof ParameterizedType) {
 				Type[] typeArgs = ((ParameterizedType) returnType).getActualTypeArguments();

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java

@@ -16,6 +16,9 @@
 
 package org.springframework.integration.handler;
 
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
 import org.springframework.core.Ordered;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.context.Orderable;
@@ -34,6 +37,8 @@ import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 
+import reactor.core.util.Exceptions;
+
 /**
  * Base class for MessageHandler implementations that provides basic validation
  * and error handling capabilities. Asserts that the incoming Message is not
@@ -46,7 +51,8 @@ import org.springframework.util.Assert;
  */
 @IntegrationManagedResource
 public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler,
-		MessageHandlerMetrics, ConfigurableMetricsAware<AbstractMessageHandlerMetrics>, TrackableComponent, Orderable {
+		MessageHandlerMetrics, ConfigurableMetricsAware<AbstractMessageHandlerMetrics>, TrackableComponent, Orderable,
+		Subscriber<Message<?>> {
 
 	private volatile boolean shouldTrack = false;
 
@@ -118,7 +124,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		boolean countsEnabled = this.countsEnabled;
 		AbstractMessageHandlerMetrics handlerMetrics = this.handlerMetrics;
 		try {
-			if (message != null && this.shouldTrack) {
+			if (this.shouldTrack) {
 				message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
 			}
 			if (countsEnabled) {
@@ -140,6 +146,31 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		}
 	}
 
+	@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) {
+		handleMessage(message);
+	}
+
+	@Override
+	public void onError(Throwable throwable) {
+		Exceptions.throwIfFatal(throwable);
+		if (throwable instanceof MessagingException) {
+			throw (MessagingException) throwable;
+		}
+		throw new MessagingException("Error occurred in message handler [" + this + "]", throwable);
+	}
+
+	@Override
+	public void onComplete() {
+
+	}
+
 	protected abstract void handleMessageInternal(Message<?> message) throws Exception;
 
 	@Override

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java

@@ -5,7 +5,7 @@
  * 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
+ *      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,
@@ -16,7 +16,9 @@
 
 package org.springframework.integration.channel.reactive;
 
+import static org.hamcrest.Matchers.isOneOf;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -56,10 +58,11 @@ public class ReactiveChannelTests {
 			this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build());
 		}
 
-		for (int i = 0; i < 10; i++) {
+		for (int i = 0; i < 9; i++) {
 			Message<?> receive = replyChannel.receive(10000);
 			assertNotNull(receive);
-			System.out.println("Receive: " + receive.getPayload());
+			assertThat(receive.getPayload(), isOneOf("0", "1", "2", "3", "4", "6", "7", "8", "9"));
+			System .out.println("Receive: " + receive.getPayload());
 		}
 	}
 
@@ -74,7 +77,10 @@ public class ReactiveChannelTests {
 
 		@ServiceActivator(inputChannel = "reactiveChannel")
 		public String handle(int payload) {
-			System.out.println("CurrentThread: " + Thread.currentThread() + " for payload: " + payload);
+			if (payload == 5) {
+				throw new IllegalStateException("intentional");
+			}
+			System .out.println("CurrentThread: " + Thread.currentThread() + " for payload: " + payload);
 			return "" + payload;
 		}
 

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveEndpointTests.java

@@ -0,0 +1,141 @@
+/*
+ * 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.channel.reactive;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.hamcrest.Matchers;
+import org.junit.Test;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.ReactiveChannel;
+import org.springframework.integration.endpoint.ReactiveEndpoint;
+import org.springframework.integration.handler.MethodInvokingMessageHandler;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageDeliveryException;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.support.GenericMessage;
+
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.test.TestSubscriber;
+
+/**
+ * @author Artem Bilan
+ * @since 5.0
+ */
+public class ReactiveEndpointTests {
+
+	@Test
+	public void testReactiveEndpointReactiveChannel() throws InterruptedException {
+		ReactiveChannel testChannel =
+				new ReactiveChannel(EmitterProcessor.create(false));
+
+		List<Message<?>> result = new LinkedList<>();
+		CountDownLatch stopLatch = new CountDownLatch(2);
+
+		MessageHandler messageHandler = m -> {
+			result.add(m);
+			stopLatch.countDown();
+		};
+
+		MethodInvokingMessageHandler testSubscriber = new MethodInvokingMessageHandler(messageHandler, (String) null);
+
+		ReactiveEndpoint reactiveEndpoint = new ReactiveEndpoint(testChannel, testSubscriber);
+		reactiveEndpoint.setBeanFactory(mock(BeanFactory.class));
+		reactiveEndpoint.afterPropertiesSet();
+		reactiveEndpoint.start();
+
+		Message<?> testMessage = new GenericMessage<>("test");
+		testChannel.send(testMessage);
+
+		reactiveEndpoint.stop();
+
+		testChannel.send(testMessage);
+
+		reactiveEndpoint.start();
+
+		Message<?> testMessage2 = new GenericMessage<>("test2");
+		testChannel.send(testMessage2);
+
+		assertTrue(stopLatch.await(10, TimeUnit.SECONDS));
+		assertThat(result, Matchers.<Message<?>>contains(testMessage, testMessage2));
+	}
+
+
+	@Test
+	public void testReactiveEndpointDirectChannel() {
+		DirectChannel testChannel = new DirectChannel();
+
+		TestSubscriber<Message<?>> testSubscriber = new TestSubscriber<>();
+
+		ReactiveEndpoint reactiveEndpoint = new ReactiveEndpoint(testChannel, testSubscriber);
+		reactiveEndpoint.setBeanFactory(mock(BeanFactory.class));
+		reactiveEndpoint.afterPropertiesSet();
+		reactiveEndpoint.start();
+
+		Message<?> testMessage = new GenericMessage<>("test");
+		testChannel.send(testMessage);
+
+		testSubscriber.assertSubscribed();
+		testSubscriber.assertNoError();
+		testSubscriber.assertNotComplete();
+
+		testSubscriber.assertValues(testMessage);
+
+		reactiveEndpoint.stop();
+
+		try {
+			testChannel.send(testMessage);
+			fail("MessageDeliveryException");
+		}
+		catch (Exception e) {
+			assertThat(e, instanceOf(MessageDeliveryException.class));
+		}
+
+		new DirectFieldAccessor(testSubscriber).setPropertyValue("s", null);
+		TestUtils.getPropertyValue(testSubscriber, "values", List.class).clear();
+
+		reactiveEndpoint.start();
+
+		testSubscriber.request(1);
+
+		testMessage = new GenericMessage<>("test2");
+
+		testChannel.send(testMessage);
+
+		testSubscriber.assertValues(testMessage);
+
+		testChannel.send(testMessage);
+
+		testSubscriber.assertError(IllegalStateException.class);
+		testSubscriber.assertErrorMessage("Can't deliver value due to lack of requests");
+
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java

@@ -57,10 +57,6 @@ import reactor.core.publisher.Mono;
  */
 public class AsyncGatewayTests {
 
-
-	// TODO: changed from 0 because of recurrent failure: is this right?
-	private final long safety = 100;
-
 	@Test
 	public void futureWithMessageReturned() throws Exception {
 		QueueChannel requestChannel = new QueueChannel();
@@ -73,10 +69,7 @@ public class AsyncGatewayTests {
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = (TestEchoService) proxyFactory.getObject();
 		Future<Message<?>> f = service.returnMessage("foo");
-		long start = System.currentTimeMillis();
 		Object result = f.get(10000, TimeUnit.MILLISECONDS);
-		long elapsed = System.currentTimeMillis() - start;
-		assertTrue(elapsed >= 200);
 		assertNotNull(result);
 		assertEquals("foobar", ((Message<?>) result).getPayload());
 	}
@@ -210,10 +203,7 @@ public class AsyncGatewayTests {
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = (TestEchoService) proxyFactory.getObject();
 		Future<String> f = service.returnString("foo");
-		long start = System.currentTimeMillis();
 		Object result = f.get(10000, TimeUnit.MILLISECONDS);
-		long elapsed = System.currentTimeMillis() - start;
-		assertTrue(elapsed >= 200 - safety);
 		assertNotNull(result);
 		assertEquals("foobar", result);
 	}
@@ -230,10 +220,7 @@ public class AsyncGatewayTests {
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = (TestEchoService) proxyFactory.getObject();
 		Future<?> f = service.returnSomething("foo");
-		long start = System.currentTimeMillis();
 		Object result = f.get(10000, TimeUnit.MILLISECONDS);
-		long elapsed = System.currentTimeMillis() - start;
-		assertTrue(elapsed >= 200 - safety);
 		assertTrue(result instanceof String);
 		assertEquals("foobar", result);
 	}
@@ -344,7 +331,7 @@ public class AsyncGatewayTests {
 	}
 
 
-	interface TestEchoService {
+	private interface TestEchoService {
 
 		Future<String> returnString(String s);