Fix FluxMessageChannel for Multi Sources

* Rename `ReactiveConsumer` to `ReactiveStreamsConsumer`
* Rename `FluxSubscribableChannel` to `ReactiveStreamsSubscribableChannel`
* Remove the `processor` functionality from the `FluxMessageChannel`
in favor of internal `FluxSink` as it is recommended by the Project Reactor:
> Most of the time, you should try to avoid using a Processor.
They are harder to use correctly and prone to some corner cases.

* Make connectable, upstream publishers for the `FluxMessageChannel` as
bridges to the internal `sink` via `this::send`.
This way we are able to receive data from multi sources.
When the source is completed (e.g. `Mono` in case of WebFlux response),
the downstream flow isn't completed.
* Rework `MessageChannelReactiveUtils#PollableChannelPublisherAdapter`
to be based on the `Flux.create()` and `onRequest()` to poll channel for messages
* Add one more request to the `ReactiveHttpRequestExecutingMessageHandler`
to be sure that we consume different `Mono`s by the `FluxSubscribableChannel`
properly without completion
* Upgrade to the `spring-io-plugin:0.0.7.RELEASE`

Add `IntegrationConsumer` implementation to the `ReactiveStreamsConsumer`

build.gradle

@@ -4,7 +4,7 @@ buildscript {
 	}
 	dependencies {
 		classpath 'io.spring.gradle:dependency-management-plugin:1.0.2.RELEASE'
-		classpath 'io.spring.gradle:spring-io-plugin:0.0.6.RELEASE'
+		classpath 'io.spring.gradle:spring-io-plugin:0.0.7.RELEASE'
 		classpath 'io.spring.gradle:docbook-reference-plugin:0.3.1'
 		classpath 'org.asciidoctor:asciidoctor-gradle-plugin:1.5.0'
 	}

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

@@ -18,22 +18,21 @@ package org.springframework.integration.channel;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 
 import org.springframework.messaging.Message;
-import org.springframework.util.Assert;
 
-import reactor.core.publisher.DirectProcessor;
+import reactor.core.publisher.ConnectableFlux;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.FluxProcessor;
 import reactor.core.publisher.FluxSink;
 
 /**
  * The {@link AbstractMessageChannel} implementation for the
- * Reactive Streams {@link Publisher} based on the Project Reactor {@link FluxProcessor}.
+ * Reactive Streams {@link Publisher} based on the Project Reactor {@link Flux}.
  *
  * @author Artem Bilan
  * @author Gary Russell
@@ -41,26 +40,21 @@ import reactor.core.publisher.FluxSink;
  * @since 5.0
  */
 public class FluxMessageChannel extends AbstractMessageChannel
-		implements Publisher<Message<?>>, FluxSubscribableChannel {
+		implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel {
 
 	private final List<Subscriber<? super Message<?>>> subscribers = new ArrayList<>();
 
-	private final List<Publisher<Message<?>>> publishers = new CopyOnWriteArrayList<>();
+	private final Map<Publisher<Message<?>>, ConnectableFlux<Message<?>>> publishers = new ConcurrentHashMap<>();
 
-	private final FluxProcessor<Message<?>, Message<?>> processor;
+	private final Flux<Message<?>> flux;
 
-	private final FluxSink<Message<?>> sink;
-
-	private volatile boolean upstreamSubscribed;
+	private FluxSink<Message<?>> sink;
 
 	public FluxMessageChannel() {
-		this(DirectProcessor.create());
-	}
-
-	public FluxMessageChannel(FluxProcessor<Message<?>, Message<?>> processor) {
-		Assert.notNull(processor, "'processor' must not be null");
-		this.processor = processor;
-		this.sink = processor.sink();
+		this.flux =
+				Flux.<Message<?>>create(emitter -> this.sink = emitter, FluxSink.OverflowStrategy.IGNORE)
+						.publish()
+						.autoConnect();
 	}
 
 	@Override
@@ -73,32 +67,26 @@ public class FluxMessageChannel extends AbstractMessageChannel
 	public void subscribe(Subscriber<? super Message<?>> subscriber) {
 		this.subscribers.add(subscriber);
 
-		this.processor.doOnCancel(() -> FluxMessageChannel.this.subscribers.remove(subscriber))
+		this.flux.doOnCancel(() -> this.subscribers.remove(subscriber))
+				.retry()
 				.subscribe(subscriber);
 
-		if (!this.upstreamSubscribed) {
-			this.publishers.forEach(this::doSubscribeTo);
-		}
+		this.publishers.values().forEach(ConnectableFlux::connect);
 	}
 
 	@Override
-	public void subscribeTo(Flux<Message<?>> publisher) {
-		this.publishers.add(publisher);
+	public void subscribeTo(Publisher<Message<?>> publisher) {
+		ConnectableFlux<Message<?>> connectableFlux =
+				Flux.from(publisher)
+						.doOnComplete(() -> this.publishers.remove(publisher))
+						.doOnNext(this::send)
+						.publish();
+
+		this.publishers.put(publisher, connectableFlux);
+
 		if (!this.subscribers.isEmpty()) {
-			doSubscribeTo(publisher);
+			connectableFlux.connect();
 		}
 	}
 
-	private void doSubscribeTo(Publisher<Message<?>> publisher) {
-		Flux.from(publisher)
-				.doOnSubscribe(s -> FluxMessageChannel.this.upstreamSubscribed = true)
-				.doOnComplete(() -> {
-					FluxMessageChannel.this.publishers.remove(publisher);
-					if (FluxMessageChannel.this.publishers.isEmpty()) {
-						FluxMessageChannel.this.upstreamSubscribed = false;
-					}
-				})
-				.subscribe(this.processor);
-	}
-
 }

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

@@ -16,9 +16,6 @@
 
 package org.springframework.integration.channel;
 
-import java.time.Duration;
-import java.util.Iterator;
-
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 
@@ -30,7 +27,7 @@ import org.springframework.messaging.SubscribableChannel;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
-import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 /**
  * Utilities for adaptation {@link MessageChannel}s to the {@link Publisher}s.
@@ -105,31 +102,17 @@ public final class MessageChannelReactiveUtils {
 		@Override
 		@SuppressWarnings("unchecked")
 		public void subscribe(Subscriber<? super Message<T>> subscriber) {
-			Iterator<Message<?>> messageIterator = new Iterator<Message<?>>() {
-
-				private Message<?> next = null;
-
-				@Override
-				public Message<?> next() {
-					Message<?> message = this.next;
-					this.next = null;
-					return message;
-				}
-
-				@Override
-				public boolean hasNext() {
-					if (this.next == null) {
-						this.next = PollableChannelPublisherAdapter.this.channel.receive(0);
-					}
-					return this.next != null;
-				}
-
-			};
-
-			Mono.<Message<?>>delay(Duration.ofMillis(100))
-					.repeat()
-					.concatMap(value -> Flux.fromIterable(() -> messageIterator))
-					.subscribe((Subscriber<? super Message<?>>) subscriber);
+			Flux
+					.<Message<T>>create(sink ->
+									sink.onRequest(n -> {
+										Message<?> m;
+										while (n-- > 0 && (m = this.channel.receive()) != null) {
+											sink.next((Message<T>) m);
+										}
+									}),
+							FluxSink.OverflowStrategy.IGNORE)
+					.subscribeOn(Schedulers.elastic())
+					.subscribe(subscriber);
 		}
 
 	}

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

@@ -16,9 +16,9 @@
 
 package org.springframework.integration.channel;
 
-import org.springframework.messaging.Message;
+import org.reactivestreams.Publisher;
 
-import reactor.core.publisher.Flux;
+import org.springframework.messaging.Message;
 
 /**
  * @author Artem Bilan
@@ -26,8 +26,8 @@ import reactor.core.publisher.Flux;
  *
  * @since 5.0
  */
-public interface FluxSubscribableChannel {
+public interface ReactiveStreamsSubscribableChannel {
 
-	void subscribeTo(Flux<Message<?>> publisher);
+	void subscribeTo(Publisher<Message<?>> publisher);
 
 }

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

@@ -39,7 +39,7 @@ 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.ReactiveConsumer;
+import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.advice.HandleMessageAdvice;
 import org.springframework.integration.scheduling.PollerMetadata;
@@ -286,7 +286,7 @@ public class ConsumerEndpointFactoryBean
 				this.endpoint = pollingConsumer;
 			}
 			else {
-				this.endpoint = new ReactiveConsumer(channel, this.handler);
+				this.endpoint = new ReactiveStreamsConsumer(channel, this.handler);
 			}
 			this.endpoint.setBeanName(this.beanName);
 			this.endpoint.setBeanFactory(this.beanFactory);

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

@@ -55,7 +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.ReactiveConsumer;
+import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -321,7 +321,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 			Assert.state(ObjectUtils.isEmpty(pollers), "A '@Poller' should not be specified for Annotation-based " +
 					"endpoint, since '" + inputChannel + "' is a SubscribableChannel (not pollable).");
 			if (inputChannel instanceof Publisher) {
-				endpoint = new ReactiveConsumer(inputChannel, handler);
+				endpoint = new ReactiveStreamsConsumer(inputChannel, handler);
 			}
 			else {
 				endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);

spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java

@@ -31,8 +31,6 @@ import org.springframework.integration.store.ChannelMessageStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.messaging.Message;
 
-import reactor.core.publisher.FluxProcessor;
-
 /**
  * @author Artem Bilan
  * @author Gary Russell
@@ -140,14 +138,6 @@ public class Channels {
 		return MessageChannels.flux(id);
 	}
 
-	public FluxMessageChannelSpec flux(FluxProcessor<Message<?>, Message<?>> processor) {
-		return MessageChannels.flux(processor);
-	}
-
-	public FluxMessageChannelSpec flux(String id, FluxProcessor<Message<?>, Message<?>> processor) {
-		return MessageChannels.flux(id, processor);
-	}
-
 	Channels() {
 		super();
 	}

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java

@@ -35,8 +35,6 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
-import reactor.core.publisher.Flux;
-
 /**
  * The central factory for fluent {@link IntegrationFlowBuilder} API.
  *
@@ -307,10 +305,10 @@ public final class IntegrationFlows {
 	 * @param publisher the {@link Publisher} to subscribe to.
 	 * @return new {@link IntegrationFlowBuilder}.
 	 */
-	public static IntegrationFlowBuilder from(Flux<Message<?>> publisher) {
+	public static IntegrationFlowBuilder from(Publisher<Message<?>> publisher) {
 		FluxMessageChannel reactiveChannel = new FluxMessageChannel();
 		reactiveChannel.subscribeTo(publisher);
-		return from(reactiveChannel);
+		return from((MessageChannel) reactiveChannel);
 	}
 
 	private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway,

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

@@ -17,9 +17,6 @@
 package org.springframework.integration.dsl.channel;
 
 import org.springframework.integration.channel.FluxMessageChannel;
-import org.springframework.messaging.Message;
-
-import reactor.core.publisher.FluxProcessor;
 
 /**
  * @author Artem Bilan
@@ -33,8 +30,4 @@ public class FluxMessageChannelSpec extends MessageChannelSpec<FluxMessageChanne
 		this.channel = new FluxMessageChannel();
 	}
 
-	FluxMessageChannelSpec(FluxProcessor<Message<?>, Message<?>> processor) {
-		this.channel = new FluxMessageChannel(processor);
-	}
-
 }

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

@@ -23,8 +23,6 @@ import org.springframework.integration.store.ChannelMessageStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.messaging.Message;
 
-import reactor.core.publisher.FluxProcessor;
-
 /**
  * @author Artem Bilan
  * @author Gary Russell
@@ -135,15 +133,6 @@ public final class MessageChannels {
 				.id(id);
 	}
 
-	public static FluxMessageChannelSpec flux(String id, FluxProcessor<Message<?>, Message<?>> processor) {
-		return flux(processor)
-				.id(id);
-	}
-
-	public static FluxMessageChannelSpec flux(FluxProcessor<Message<?>, Message<?>> processor) {
-		return new FluxMessageChannelSpec(processor);
-	}
-
 	private MessageChannels() {
 		super();
 	}

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

@@ -25,6 +25,8 @@ import org.reactivestreams.Subscription;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.router.MessageRouter;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -33,16 +35,18 @@ import org.springframework.util.Assert;
 import org.springframework.util.ErrorHandler;
 
 import reactor.core.Disposable;
-import reactor.core.Exceptions;
 import reactor.core.publisher.BaseSubscriber;
-import reactor.core.publisher.Operators;
 
 
 /**
  * @author Artem Bilan
  * @since 5.0
  */
-public class ReactiveConsumer extends AbstractEndpoint {
+public class ReactiveStreamsConsumer extends AbstractEndpoint implements IntegrationConsumer {
+
+	private final MessageChannel inputChannel;
+
+	private final MessageHandler messageHandler;
 
 	private final Publisher<Message<Object>> publisher;
 
@@ -55,28 +59,56 @@ public class ReactiveConsumer extends AbstractEndpoint {
 	private volatile Subscription subscription;
 
 	@SuppressWarnings("unchecked")
-	public ReactiveConsumer(MessageChannel inputChannel, MessageHandler messageHandler) {
+	public ReactiveStreamsConsumer(MessageChannel inputChannel, MessageHandler messageHandler) {
 		this(inputChannel,
 				messageHandler instanceof Subscriber
 						? (Subscriber<Message<?>>) messageHandler
 						: new MessageHandlerSubscriber(messageHandler));
 	}
 
-	public ReactiveConsumer(MessageChannel inputChannel, final Subscriber<Message<?>> subscriber) {
+	public ReactiveStreamsConsumer(MessageChannel inputChannel, final Subscriber<Message<?>> subscriber) {
+		this.inputChannel = inputChannel;
 		Assert.notNull(inputChannel, "'inputChannel' must not be null");
 		Assert.notNull(subscriber, "'subscriber' must not be null");
 
 		this.publisher = MessageChannelReactiveUtils.toPublisher(inputChannel);
-
 		this.subscriber = subscriber;
-
 		this.lifecycleDelegate = subscriber instanceof Lifecycle ? (Lifecycle) subscriber : null;
+		if (subscriber instanceof MessageHandlerSubscriber) {
+			this.messageHandler = ((MessageHandlerSubscriber) subscriber).messageHandler;
+		}
+		else {
+			this.messageHandler = this.subscriber::onNext;
+		}
 	}
 
 	public void setErrorHandler(ErrorHandler errorHandler) {
 		this.errorHandler = errorHandler;
 	}
 
+	@Override
+	public MessageChannel getInputChannel() {
+		return this.inputChannel;
+	}
+
+	@Override
+	public MessageChannel getOutputChannel() {
+		if (this.messageHandler instanceof MessageProducer) {
+			return ((MessageProducer) this.messageHandler).getOutputChannel();
+		}
+		else if (this.messageHandler instanceof MessageRouter) {
+			return ((MessageRouter) this.messageHandler).getDefaultOutputChannel();
+		}
+		else {
+			return null;
+		}
+	}
+
+	@Override
+	public MessageHandler getHandler() {
+		return this.messageHandler;
+	}
+
 	@Override
 	protected void onInit() throws Exception {
 		super.onInit();
@@ -93,11 +125,11 @@ public class ReactiveConsumer extends AbstractEndpoint {
 		}
 		this.publisher.subscribe(new BaseSubscriber<Message<?>>() {
 
-			private final Subscriber<Message<?>> delegate = ReactiveConsumer.this.subscriber;
+			private final Subscriber<Message<?>> delegate = ReactiveStreamsConsumer.this.subscriber;
 
 			public void hookOnSubscribe(Subscription s) {
 				this.delegate.onSubscribe(s);
-				ReactiveConsumer.this.subscription = s;
+				ReactiveStreamsConsumer.this.subscription = s;
 			}
 
 			public void hookOnNext(Message<?> message) {
@@ -105,7 +137,7 @@ public class ReactiveConsumer extends AbstractEndpoint {
 					this.delegate.onNext(message);
 				}
 				catch (Exception e) {
-					ReactiveConsumer.this.errorHandler.handleError(e);
+					ReactiveStreamsConsumer.this.errorHandler.handleError(e);
 					hookOnError(e);
 				}
 			}
@@ -160,18 +192,11 @@ public class ReactiveConsumer extends AbstractEndpoint {
 
 		@Override
 		public void onError(Throwable t) {
-			if (t == null) {
-				throw Exceptions.argumentIsNullException();
-			}
-			onComplete();
-			Operators.onErrorDropped(t);
 		}
 
 		@Override
 		public void onComplete() {
-			if (this.subscription != null) {
-				this.subscription = null;
-			}
+			dispose();
 		}
 
 		@Override
@@ -185,7 +210,7 @@ public class ReactiveConsumer extends AbstractEndpoint {
 
 		@Override
 		public boolean isDisposed() {
-			return false;
+			return this.subscription == null;
 		}
 
 

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

@@ -20,7 +20,6 @@ import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
 import org.springframework.core.Ordered;
-import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.context.Orderable;
 import org.springframework.integration.history.MessageHistory;
@@ -37,9 +36,6 @@ import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
-import org.springframework.util.ErrorHandler;
-
-import reactor.core.publisher.Operators;
 
 /**
  * Base class for MessageHandler implementations that provides basic validation
@@ -72,8 +68,6 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 
 	private volatile boolean loggingEnabled = true;
 
-	private ErrorHandler reactiveErrorHandler;
-
 	@Override
 	public boolean isLoggingEnabled() {
 		return this.loggingEnabled;
@@ -104,16 +98,6 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		this.shouldTrack = shouldTrack;
 	}
 
-	/**
-	 * Set the error handler to use when an exception occurs when this handler
-	 * is invoked as a reactive {@link Subscriber}.
-	 * @param reactiveErrorHandler the error handler.
-	 * @since 5.0
-	 */
-	public void setReactiveErrorHandler(ErrorHandler reactiveErrorHandler) {
-		this.reactiveErrorHandler = reactiveErrorHandler;
-	}
-
 	@Override
 	public void configureMetrics(AbstractMessageHandlerMetrics metrics) {
 		Assert.notNull(metrics, "'metrics' must not be null");
@@ -125,14 +109,6 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		if (this.statsEnabled) {
 			this.handlerMetrics.setFullStatsEnabled(true);
 		}
-		if (this.reactiveErrorHandler == null) {
-			if (getBeanFactory() != null) {
-				this.reactiveErrorHandler = new MessagePublishingErrorHandler(getChannelResolver());
-			}
-			else {
-				this.reactiveErrorHandler = new MessagePublishingErrorHandler();
-			}
-		}
 	}
 
 	@Override
@@ -176,17 +152,12 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 
 	@Override
 	public void onNext(Message<?> message) {
-		try {
-			handleMessage(message);
-		}
-		catch (MessagingException e) {
-			this.reactiveErrorHandler.handleError(e);
-		}
+		handleMessage(message);
 	}
 
 	@Override
 	public void onError(Throwable throwable) {
-		Operators.onErrorDropped(throwable);
+
 	}
 
 	@Override

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

@@ -24,7 +24,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import org.reactivestreams.Publisher;
 
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
-import org.springframework.integration.channel.FluxSubscribableChannel;
+import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.routingslip.RoutingSlipRouteStrategy;
@@ -147,6 +147,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 		return false;
 	}
 
+	@SuppressWarnings("unchecked")
 	protected void produceOutput(Object reply, final Message<?> requestMessage) {
 		final MessageHeaders requestHeaders = requestMessage.getHeaders();
 
@@ -190,7 +191,8 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 		}
 
 		if (this.async && (reply instanceof ListenableFuture<?> || reply instanceof Publisher<?>)) {
-			if (reply instanceof ListenableFuture<?> || !(getOutputChannel() instanceof FluxSubscribableChannel)) {
+			if (reply instanceof ListenableFuture<?> ||
+					!(getOutputChannel() instanceof ReactiveStreamsSubscribableChannel)) {
 				ListenableFuture<?> future;
 				if (reply instanceof ListenableFuture<?>) {
 					future = (ListenableFuture<?>) reply;
@@ -235,9 +237,10 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 				});
 			}
 			else {
-				((FluxSubscribableChannel) getOutputChannel())
-						.subscribeTo(Flux.from((Publisher<?>) reply)
-								.map(result -> createOutputMessage(result, requestHeaders)));
+				((ReactiveStreamsSubscribableChannel) getOutputChannel())
+						.subscribeTo(
+								Flux.from((Publisher<?>) reply)
+										.map(result -> createOutputMessage(result, requestHeaders)));
 			}
 		}
 		else {

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

@@ -22,8 +22,8 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
-import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -46,28 +46,26 @@ import org.reactivestreams.Subscription;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
-import org.springframework.integration.endpoint.ReactiveConsumer;
+import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
 import org.springframework.integration.handler.MethodInvokingMessageHandler;
 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;
-
 /**
  * @author Artem Bilan
  *
  * @since 5.0
  */
-public class ReactiveConsumerTests {
+public class ReactiveStreamsConsumerTests {
 
 	@Test
-	public void testReactiveConsumerReactiveChannel() throws InterruptedException {
-		FluxMessageChannel testChannel = new FluxMessageChannel(EmitterProcessor.create(false));
+	public void testReactiveStreamsConsumerFluxMessageChannel() throws InterruptedException {
+		FluxMessageChannel testChannel = new FluxMessageChannel();
 
 		List<Message<?>> result = new LinkedList<>();
 		CountDownLatch stopLatch = new CountDownLatch(2);
@@ -79,7 +77,7 @@ public class ReactiveConsumerTests {
 
 		MessageHandler testSubscriber = new MethodInvokingMessageHandler(messageHandler, (String) null);
 
-		ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber);
+		ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber);
 		reactiveConsumer.setBeanFactory(mock(BeanFactory.class));
 		reactiveConsumer.afterPropertiesSet();
 		reactiveConsumer.start();
@@ -103,7 +101,7 @@ public class ReactiveConsumerTests {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void testReactiveConsumerDirectChannel() throws InterruptedException {
+	public void testReactiveStreamsConsumerDirectChannel() throws InterruptedException {
 		DirectChannel testChannel = new DirectChannel();
 
 		Subscriber<Message<?>> testSubscriber = (Subscriber<Message<?>>) Mockito.mock(Subscriber.class);
@@ -117,7 +115,7 @@ public class ReactiveConsumerTests {
 				.given(testSubscriber)
 				.onNext(any(Message.class));
 
-		ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber);
+		ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber);
 		reactiveConsumer.setBeanFactory(mock(BeanFactory.class));
 		reactiveConsumer.afterPropertiesSet();
 		reactiveConsumer.start();
@@ -163,7 +161,7 @@ public class ReactiveConsumerTests {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void testReactiveConsumerPollableChannel() throws InterruptedException {
+	public void testReactiveStreamsConsumerPollableChannel() throws InterruptedException {
 		QueueChannel testChannel = new QueueChannel();
 
 		Subscriber<Message<?>> testSubscriber = (Subscriber<Message<?>>) Mockito.mock(Subscriber.class);
@@ -177,7 +175,7 @@ public class ReactiveConsumerTests {
 				.given(testSubscriber)
 				.onNext(any(Message.class));
 
-		ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber);
+		ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber);
 		reactiveConsumer.setBeanFactory(mock(BeanFactory.class));
 		reactiveConsumer.afterPropertiesSet();
 		reactiveConsumer.start();
@@ -223,7 +221,7 @@ public class ReactiveConsumerTests {
 	}
 
 	@Test
-	public void testReactiveConsumerViaConsumerEndpointFactoryBean() throws Exception {
+	public void testReactiveStreamsConsumerViaConsumerEndpointFactoryBean() throws Exception {
 		FluxMessageChannel testChannel = new FluxMessageChannel();
 
 		List<Message<?>> result = new LinkedList<>();

spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java

@@ -32,7 +32,6 @@ import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.apache.log4j.Level;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.reactivestreams.Publisher;
@@ -49,7 +48,6 @@ import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.channel.MessageChannels;
 import org.springframework.integration.dsl.context.IntegrationFlowContext;
-import org.springframework.integration.test.rule.Log4jLevelAdjuster;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
@@ -69,9 +67,6 @@ import reactor.core.publisher.Flux;
 @DirtiesContext
 public class ReactiveStreamsTests {
 
-//	@Rule
-	public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.DEBUG, "org.springframework.integration");
-
 	@Autowired
 	@Qualifier("reactiveFlow")
 	private Publisher<Message<String>> publisher;

spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java

@@ -18,26 +18,29 @@ package org.springframework.integration.http.outbound;
 
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
-import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
+import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeader;
 
-import java.time.Duration;
+import java.util.List;
 
 import org.junit.Test;
+import org.reactivestreams.Subscriber;
 
 import org.springframework.http.HttpStatus;
 import org.springframework.http.client.reactive.ClientHttpConnector;
-import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.http.HttpHeaders;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.test.web.reactive.server.HttpHandlerConnector;
 import org.springframework.web.reactive.function.client.WebClient;
 
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 /**
  * @author Shiliang Li
@@ -66,12 +69,15 @@ public class ReactiveHttpRequestExecutingMessageHandlerTests {
 		FluxMessageChannel ackChannel = new FluxMessageChannel();
 		reactiveHandler.setOutputChannel(ackChannel);
 		reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build());
+		reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build());
 
-		Message<?> ack = Mono.from(ackChannel).block(Duration.ofSeconds(10));
-
-		assertNotNull(ack);
-		assertNotNull(ack.getHeaders());
-		assertEquals(ack.getHeaders().get(HttpHeaders.STATUS_CODE), HttpStatus.OK);
+		StepVerifier.create(ackChannel, 2)
+				.assertNext(m -> assertThat(m, hasHeader(HttpHeaders.STATUS_CODE, HttpStatus.OK)))
+				.assertNext(m -> assertThat(m, hasHeader(HttpHeaders.STATUS_CODE, HttpStatus.OK)))
+				.then(() ->
+						((Subscriber<?>) TestUtils.getPropertyValue(ackChannel, "subscribers", List.class).get(0))
+								.onComplete())
+				.verifyComplete();
 	}
 
 	@Test