Introduce ReactiveSubscribableChannel and Others

* Move `SubscribableChannelPublisherAdapter` and `PollableChannelPublisherAdapter` logic to the `MessageChannelReactiveUtils` public API
* Rework `ReactiveConsumer` and `IntegrationFlowDefinition.toReactivePublisher()` to use `MessageChannelReactiveUtils`
* Add `ReactiveSubscribableChannel` interface to represent abstraction with capability to subscribe to `Publisher`
* Implement `ReactiveSubscribableChannel` in the `ReactiveChannel`
* Add `IntegrationFlows.from(Publisher)` factory and use newly introduced `ReactiveSubscribableChannel.subscribeTo()`
* Add `MessageChannels.reactive()` factory methods for the `ReactiveChannelSpec`
* Make `ReactiveChannel` as an `AbstractMessageChannel` to give an interception opportunity and gather metrics for `send()`
* A `ReactiveStreamsTests.testFromPublisher()` demonstrate how to subscribe to the `Flux` from an `IntegrationFlow` and at the same time get a gain of the runtime flow registration
* The `ReactiveChannelTests.testMessageChannelReactiveAdaptation()` demonstrates how to use `MessageChannelReactiveUtils.toPublisher()` (name can be changes though...)

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

@@ -0,0 +1,136 @@
+/*
+ * Copyright 2017 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 java.util.Iterator;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+
+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 reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
+import reactor.core.publisher.Mono;
+
+/**
+ * Utilities for adaptation {@link MessageChannel}s to the {@link Publisher}s.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public final class MessageChannelReactiveUtils {
+
+	private MessageChannelReactiveUtils() {
+		super();
+	}
+
+	@SuppressWarnings("unchecked")
+	public static <T> Publisher<Message<T>> toPublisher(MessageChannel messageChannel) {
+		if (messageChannel instanceof Publisher) {
+			return (Publisher<Message<T>>) messageChannel;
+		}
+		else if (messageChannel instanceof SubscribableChannel) {
+			return adaptSubscribableChannelToPublisher((SubscribableChannel) messageChannel);
+		}
+		else if (messageChannel instanceof PollableChannel) {
+			return adaptPollableChannelToPublisher((PollableChannel) messageChannel);
+		}
+		else {
+			throw new IllegalArgumentException("The 'messageChannel' must be an instance of Publisher, " +
+					"SubscribableChannel or PollableChannel, not: " + messageChannel);
+		}
+	}
+
+	private static <T> Publisher<Message<T>> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) {
+		return new SubscribableChannelPublisherAdapter<>(inputChannel);
+	}
+
+	private static <T> Publisher<Message<T>> adaptPollableChannelToPublisher(PollableChannel inputChannel) {
+		return new PollableChannelPublisherAdapter<>(inputChannel);
+	}
+
+
+	private final static class SubscribableChannelPublisherAdapter<T> implements Publisher<Message<T>> {
+
+		private final SubscribableChannel channel;
+
+		SubscribableChannelPublisherAdapter(SubscribableChannel channel) {
+			this.channel = channel;
+		}
+
+		@Override
+		@SuppressWarnings("unchecked")
+		public void subscribe(Subscriber<? super Message<T>> subscriber) {
+			Flux.
+					<Message<?>>create(emitter -> {
+								MessageHandler messageHandler = emitter::next;
+								this.channel.subscribe(messageHandler);
+								emitter.setCancellation(() -> this.channel.unsubscribe(messageHandler));
+							},
+							FluxSink.OverflowStrategy.IGNORE)
+					.subscribe((Subscriber<? super Message<?>>) subscriber);
+		}
+
+	}
+
+	private final static class PollableChannelPublisherAdapter<T> implements Publisher<Message<T>> {
+
+		private final PollableChannel channel;
+
+		PollableChannelPublisherAdapter(final PollableChannel channel) {
+			this.channel = channel;
+		}
+
+		@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<?>>delayMillis(100)
+					.repeat()
+					.concatMap(value -> Flux.fromIterable(() -> messageIterator))
+					.subscribe((Subscriber<? super Message<?>>) subscriber);
+		}
+
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -16,27 +16,40 @@
 
 package org.springframework.integration.channel;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
 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.BlockingSink;
 import reactor.core.publisher.DirectProcessor;
+import reactor.core.publisher.Operators;
 
 /**
  * @author Artem Bilan
+ *
  * @since 5.0
  */
-public class ReactiveChannel implements MessageChannel, Publisher<Message<?>> {
+public class ReactiveChannel extends AbstractMessageChannel
+		implements Publisher<Message<?>>, ReactiveSubscribableChannel {
+
+	private final List<Subscriber<? super Message<?>>> subscribers = new ArrayList<>();
+
+	private final List<Publisher<Message<?>>> publishers = new CopyOnWriteArrayList<>();
 
 	private final Processor<Message<?>, Message<?>> processor;
 
 	private final BlockingSink<Message<?>> sink;
 
+	private volatile boolean upstreamSubscribed;
+
 	public ReactiveChannel() {
 		this(DirectProcessor.create());
 	}
@@ -48,18 +61,55 @@ public class ReactiveChannel implements MessageChannel, Publisher<Message<?>> {
 	}
 
 	@Override
-	public boolean send(Message<?> message) {
-		return send(message, -1);
-	}
-
-	@Override
-	public boolean send(Message<?> message, long timeout) {
+	protected boolean doSend(Message<?> message, long timeout) {
 		return this.sink.submit(message, timeout) > -1;
 	}
 
 	@Override
 	public void subscribe(Subscriber<? super Message<?>> subscriber) {
-		this.processor.subscribe(subscriber);
+		this.subscribers.add(subscriber);
+		this.processor.subscribe(new Operators.SubscriberAdapter<Message<?>, Message<?>>(subscriber) {
+
+			@Override
+			protected void doCancel() {
+				super.doCancel();
+				ReactiveChannel.this.subscribers.remove(subscriber);
+			}
+
+		});
+
+		if (!this.upstreamSubscribed) {
+			this.publishers.forEach(this::doSubscribeTo);
+		}
+	}
+
+	@Override
+	public void subscribeTo(Publisher<Message<?>> publisher) {
+		this.publishers.add(publisher);
+		if (!this.subscribers.isEmpty()) {
+			doSubscribeTo(publisher);
+		}
+	}
+
+	private void doSubscribeTo(Publisher<Message<?>> publisher) {
+		publisher.subscribe(new Operators.SubscriberAdapter<Message<?>, Message<?>>(this.processor) {
+
+			@Override
+			protected void doOnSubscribe(Subscription subscription) {
+				super.doOnSubscribe(subscription);
+				ReactiveChannel.this.upstreamSubscribed = true;
+			}
+
+			@Override
+			protected void doComplete() {
+				super.doComplete();
+				ReactiveChannel.this.publishers.remove(publisher);
+				if (ReactiveChannel.this.publishers.isEmpty()) {
+					ReactiveChannel.this.upstreamSubscribed = false;
+				}
+			}
+
+		});
 	}
 
 }

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2017 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.Publisher;
+
+import org.springframework.messaging.Message;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public interface ReactiveSubscribableChannel {
+
+	void subscribeTo(Publisher<Message<?>> publisher);
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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,15 @@ package org.springframework.integration.dsl;
 import java.util.Queue;
 import java.util.concurrent.Executor;
 
+import org.reactivestreams.Processor;
+
 import org.springframework.integration.dsl.channel.DirectChannelSpec;
 import org.springframework.integration.dsl.channel.ExecutorChannelSpec;
 import org.springframework.integration.dsl.channel.MessageChannels;
 import org.springframework.integration.dsl.channel.PriorityChannelSpec;
 import org.springframework.integration.dsl.channel.PublishSubscribeChannelSpec;
 import org.springframework.integration.dsl.channel.QueueChannelSpec;
+import org.springframework.integration.dsl.channel.ReactiveChannelSpec;
 import org.springframework.integration.dsl.channel.RendezvousChannelSpec;
 import org.springframework.integration.store.ChannelMessageStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
@@ -128,6 +131,23 @@ public class Channels {
 		return MessageChannels.executor(id, executor);
 	}
 
+
+	public ReactiveChannelSpec reactive() {
+		return MessageChannels.reactive();
+	}
+
+	public ReactiveChannelSpec reactive(String id) {
+		return MessageChannels.reactive(id);
+	}
+
+	public ReactiveChannelSpec reactive(Processor<Message<?>, Message<?>> processor) {
+		return MessageChannels.reactive(processor);
+	}
+
+	public ReactiveChannelSpec reactive(String id, Processor<Message<?>, Message<?>> processor) {
+		return MessageChannels.reactive(id, processor);
+	}
+
 	Channels() {
 		super();
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -38,6 +38,7 @@ import org.springframework.integration.aggregator.ResequencingMessageHandler;
 import org.springframework.integration.channel.ChannelInterceptorAware;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FixedSubscriberChannel;
+import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.ReactiveChannel;
 import org.springframework.integration.channel.interceptor.WireTap;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
@@ -48,7 +49,6 @@ import org.springframework.integration.dsl.channel.MessageChannelSpec;
 import org.springframework.integration.dsl.channel.WireTapSpec;
 import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
 import org.springframework.integration.dsl.support.MessageChannelReference;
-import org.springframework.integration.endpoint.ReactiveConsumer;
 import org.springframework.integration.expression.ControlBusMethodFilter;
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.filter.ExpressionEvaluatingSelector;
@@ -2726,8 +2726,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		}
 		else {
 			if (channelForPublisher != null) {
-				Publisher<?> messagePublisher = ReactiveConsumer.adaptToPublisher(channelForPublisher);
-				publisher = (Publisher<Message<T>>) messagePublisher;
+				publisher = MessageChannelReactiveUtils.toPublisher(channelForPublisher);
 			}
 			else {
 				MessageChannel reactiveChannel = new ReactiveChannel();

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

@@ -17,10 +17,12 @@
 package org.springframework.integration.dsl;
 
 import java.util.function.Consumer;
-import java.util.function.Function;
+
+import org.reactivestreams.Publisher;
 
 import org.springframework.integration.annotation.MessagingGateway;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.ReactiveChannel;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.channel.MessageChannelSpec;
 import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
@@ -30,6 +32,7 @@ import org.springframework.integration.endpoint.MethodInvokingMessageSource;
 import org.springframework.integration.gateway.AnnotationGatewayProxyFactoryBean;
 import org.springframework.integration.gateway.GatewayProxyFactoryBean;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
@@ -78,20 +81,6 @@ public final class IntegrationFlows {
 				: from(messageChannelName);
 	}
 
-	/**
-	 * Populate the {@link MessageChannel} object to the
-	 * {@link IntegrationFlowBuilder} chain using the fluent API from {@link Channels} factory.
-	 * The {@link org.springframework.integration.dsl.IntegrationFlow} {@code inputChannel}.
-	 * @param channels the {@link Function} to use method chain to configure.
-	 * {@link MessageChannel} via {@link Channels} factory.
-	 * @return new {@link IntegrationFlowBuilder}.
-	 * @see Channels
-	 */
-	public static IntegrationFlowBuilder from(Function<Channels, MessageChannelSpec<?, ?>> channels) {
-		Assert.notNull(channels);
-		return from(channels.apply(new Channels()));
-	}
-
 	/**
 	 * Populate the {@link MessageChannel} object to the
 	 * {@link IntegrationFlowBuilder} chain using the fluent API from {@link MessageChannelSpec}.
@@ -309,6 +298,18 @@ public final class IntegrationFlows {
 				.addComponent(gatewayProxyFactoryBean);
 	}
 
+	/**
+	 * Populate a {@link ReactiveChannel} to the {@link IntegrationFlowBuilder} chain
+	 * and subscribe it to the provided {@link Publisher}.
+	 * @param publisher the {@link Publisher} to subscribe to.
+	 * @return new {@link IntegrationFlowBuilder}.
+	 */
+	public static IntegrationFlowBuilder from(Publisher<Message<?>> publisher) {
+		ReactiveChannel reactiveChannel = new ReactiveChannel();
+		reactiveChannel.subscribeTo(publisher);
+		return from((MessageChannel) reactiveChannel);
+	}
+
 	private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway,
 			IntegrationFlowBuilder integrationFlowBuilder) {
 		MessageChannel outputChannel = inboundGateway.getRequestChannel();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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,6 +19,8 @@ package org.springframework.integration.dsl.channel;
 import java.util.Queue;
 import java.util.concurrent.Executor;
 
+import org.reactivestreams.Processor;
+
 import org.springframework.integration.store.ChannelMessageStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.messaging.Message;
@@ -123,6 +125,22 @@ public final class MessageChannels {
 		return MessageChannels.<S>publishSubscribe(executor).id(id);
 	}
 
+	public static ReactiveChannelSpec reactive() {
+		return new ReactiveChannelSpec();
+	}
+
+	public static ReactiveChannelSpec reactive(String id) {
+		return reactive().id(id);
+	}
+
+	public static ReactiveChannelSpec reactive(Processor<Message<?>, Message<?>> processor) {
+		return new ReactiveChannelSpec(processor);
+	}
+
+	public static ReactiveChannelSpec reactive(String id, Processor<Message<?>, Message<?>> processor) {
+		return reactive(processor).id(id);
+	}
+
 	private MessageChannels() {
 		super();
 	}

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

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2017 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.dsl.channel;
+
+import org.reactivestreams.Processor;
+
+import org.springframework.integration.channel.ReactiveChannel;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public class ReactiveChannelSpec extends MessageChannelSpec<ReactiveChannelSpec, ReactiveChannel> {
+
+	ReactiveChannelSpec() {
+		this.channel = new ReactiveChannel();
+	}
+
+	ReactiveChannelSpec(Processor<Message<?>, Message<?>> processor) {
+		this.channel = new ReactiveChannel(processor);
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -16,20 +16,17 @@
 
 package org.springframework.integration.endpoint;
 
-import java.util.Iterator;
 import java.util.function.Consumer;
 
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
+import org.springframework.integration.channel.MessageChannelReactiveUtils;
 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;
 
@@ -37,9 +34,6 @@ import reactor.core.Disposable;
 import reactor.core.Exceptions;
 import reactor.core.Receiver;
 import reactor.core.Trackable;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.FluxSink;
-import reactor.core.publisher.Mono;
 import reactor.core.publisher.Operators;
 
 
@@ -65,12 +59,8 @@ public class ReactiveConsumer extends AbstractEndpoint {
 		Assert.notNull(inputChannel);
 		Assert.notNull(subscriber);
 
-		if (inputChannel instanceof Publisher) {
-			this.publisher = (Publisher<Message<?>>) inputChannel;
-		}
-		else {
-			this.publisher = adaptToPublisher(inputChannel);
-		}
+		Publisher<?> messagePublisher = MessageChannelReactiveUtils.toPublisher(inputChannel);
+		this.publisher = (Publisher<Message<?>>) messagePublisher;
 
 		this.subscriber = new Operators.SubscriberAdapter<Message<?>, Message<?>>(subscriber) {
 
@@ -111,89 +101,6 @@ public class ReactiveConsumer extends AbstractEndpoint {
 		this.subscriber.cancel();
 	}
 
-	public static 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 static Publisher<Message<?>> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) {
-		return new SubscribableChannelPublisherAdapter(inputChannel);
-	}
-
-	private static Publisher<Message<?>> adaptPollableChannelToPublisher(PollableChannel inputChannel) {
-		return new PollableChannelPublisherAdapter(inputChannel);
-	}
-
-
-	private final static class SubscribableChannelPublisherAdapter implements Publisher<Message<?>> {
-
-		private final SubscribableChannel channel;
-
-		SubscribableChannelPublisherAdapter(SubscribableChannel channel) {
-			this.channel = channel;
-		}
-
-		@Override
-		public void subscribe(Subscriber<? super Message<?>> subscriber) {
-			Flux.
-					<Message<?>>create(emitter -> {
-								MessageHandler messageHandler = emitter::next;
-								this.channel.subscribe(messageHandler);
-								emitter.setCancellation(() -> this.channel.unsubscribe(messageHandler));
-							},
-							FluxSink.OverflowStrategy.IGNORE)
-					.subscribe(subscriber);
-		}
-
-	}
-
-	private final static class PollableChannelPublisherAdapter implements Publisher<Message<?>> {
-
-		private final PollableChannel channel;
-
-
-		PollableChannelPublisherAdapter(final PollableChannel channel) {
-			this.channel = channel;
-		}
-
-		@Override
-		public void subscribe(Subscriber<? super Message<?>> 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<?>>delayMillis(100)
-					.repeat()
-					.concatMap(value -> Flux.fromIterable(() -> messageIterator))
-					.subscribe(subscriber);
-		}
-
-	}
 
 	private static final class ConsumerSubscriber implements Subscriber<Message<?>>, Receiver, Disposable, Trackable {
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -16,11 +16,18 @@
 
 package org.springframework.integration.channel.reactive;
 
+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.isOneOf;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -29,16 +36,21 @@ 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.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.channel.ReactiveChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
+import reactor.core.publisher.Flux;
+
 /**
  * @author Artem Bilan
  * @since 5.0
@@ -50,6 +62,9 @@ public class ReactiveChannelTests {
 	@Autowired
 	private MessageChannel reactiveChannel;
 
+	@Autowired
+	private MessageChannel queueChannel;
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testReactiveMessageChannel() throws InterruptedException {
@@ -60,8 +75,9 @@ public class ReactiveChannelTests {
 				this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build());
 			}
 			catch (Exception e) {
-				assertThat(e.getCause(), instanceOf(MessageHandlingException.class));
-				assertThat(e.getCause().getCause(), instanceOf(IllegalStateException.class));
+				assertThat(e, instanceOf(MessageDeliveryException.class));
+				assertThat(e.getCause().getCause(), instanceOf(MessageHandlingException.class));
+				assertThat(e.getCause().getCause().getCause(), instanceOf(IllegalStateException.class));
 				assertThat(e.getMessage(), containsString("intentional"));
 			}
 		}
@@ -73,6 +89,24 @@ public class ReactiveChannelTests {
 		}
 	}
 
+	@Test
+	public void testMessageChannelReactiveAdaptation() throws InterruptedException {
+		CountDownLatch done = new CountDownLatch(2);
+		List<String> results = new ArrayList<>();
+
+		Flux.from(MessageChannelReactiveUtils.<String>toPublisher(this.queueChannel))
+				.map(Message::getPayload)
+				.map(String::toUpperCase)
+				.doOnNext(results::add)
+				.subscribe(v -> done.countDown());
+
+		this.queueChannel.send(new GenericMessage<>("foo"));
+		this.queueChannel.send(new GenericMessage<>("bar"));
+
+		assertTrue(done.await(10, TimeUnit.SECONDS));
+		assertThat(results, contains("FOO", "BAR"));
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class TestConfiguration {
@@ -90,6 +124,11 @@ public class ReactiveChannelTests {
 			return "" + payload;
 		}
 
+		@Bean
+		public MessageChannel queueChannel() {
+			return new QueueChannel();
+		}
+
 	}
 
 }

spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -242,7 +242,7 @@ public class CorrelationHandlerTests {
 		@Bean
 		@DependsOn("barrierFlow")
 		public IntegrationFlow releaseBarrierFlow(MessageTriggerAction barrierTriggerAction) {
-			return IntegrationFlows.from(c -> c.queue("releaseChannel"))
+			return IntegrationFlows.from(MessageChannels.queue("releaseChannel"))
 					.trigger(barrierTriggerAction,
 							e -> e.poller(p -> p.fixedDelay(100)))
 					.get();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -33,7 +33,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.log4j.Level;
-import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.reactivestreams.Publisher;
@@ -44,9 +43,12 @@ import org.springframework.context.Lifecycle;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
+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;
@@ -67,7 +69,7 @@ import reactor.core.publisher.Flux;
 @DirtiesContext
 public class ReactiveStreamsTests {
 
-	@Rule
+//	@Rule
 	public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.DEBUG, "org.springframework.integration");
 
 	@Autowired
@@ -86,6 +88,9 @@ public class ReactiveStreamsTests {
 	@Qualifier("inputChannel")
 	private MessageChannel inputChannel;
 
+	@Autowired
+	private IntegrationFlowContext integrationFlowContext;
+
 	@Test
 	public void testReactiveFlow() throws Exception {
 		List<String> results = new ArrayList<>();
@@ -99,7 +104,7 @@ public class ReactiveStreamsTests {
 		this.messageSource.start();
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		String[] strings = results.toArray(new String[results.size()]);
-		assertArrayEquals(new String[] { "A", "B", "C", "D", "E", "F" }, strings);
+		assertArrayEquals(new String[] {"A", "B", "C", "D", "E", "F"}, strings);
 		this.messageSource.stop();
 	}
 
@@ -119,17 +124,17 @@ public class ReactiveStreamsTests {
 
 		Future<List<Integer>> future =
 				Executors.newSingleThreadExecutor().submit(() ->
-					Flux.just("11,12,13")
-							.map(v -> v.split(","))
-							.flatMapIterable(Arrays::asList)
-							.map(Integer::parseInt)
-							.<Message<Integer>>map(GenericMessage<Integer>::new)
-							.concatWith(this.pollablePublisher)
-							.take(7)
-							.map(Message::getPayload)
-							.log("org.springframework.integration.flux")
-							.collectList()
-							.block(Duration.ofSeconds(10))
+						Flux.just("11,12,13")
+								.map(v -> v.split(","))
+								.flatMapIterable(Arrays::asList)
+								.map(Integer::parseInt)
+								.<Message<Integer>>map(GenericMessage<Integer>::new)
+								.concatWith(this.pollablePublisher)
+								.take(7)
+								.map(Message::getPayload)
+								.log("org.springframework.integration.flux")
+								.collectList()
+								.block(Duration.ofSeconds(10))
 				);
 
 		this.inputChannel.send(new GenericMessage<>("6,7,8,9,10"));
@@ -141,6 +146,33 @@ public class ReactiveStreamsTests {
 		assertEquals(7, integers.size());
 	}
 
+	@Test
+	public void testFromPublisher() {
+		Flux<Message<?>> messageFlux = Flux.just("1,2,3,4")
+				.map(v -> v.split(","))
+				.flatMapIterable(Arrays::asList)
+				.map(Integer::parseInt)
+				.log("org.springframework.integration.flux")
+				.map(GenericMessage<Integer>::new);
+
+		QueueChannel resultChannel = new QueueChannel();
+
+		IntegrationFlow integrationFlow =
+				IntegrationFlows.from(messageFlux)
+						.log("org.springframework.integration.flux2")
+						.<Integer, Integer>transform(p -> p * 2)
+						.channel(resultChannel)
+						.get();
+
+		this.integrationFlowContext.registration(integrationFlow)
+				.register();
+
+		for (int i = 0; i < 4; i++) {
+			Message<?> receive = resultChannel.receive(10000);
+			assertNotNull(receive);
+			assertEquals((i + 1) * 2, receive.getPayload());
+		}
+	}
 
 	@Configuration
 	@EnableIntegration