Add Reactive Stream chapter into docs

* Improve `channel.adoc`: polishing, new channels, java config
* Make `LoadBalancingStrategy` in the `DirectChannel` ctor as
`@Nullable` to reflect the real logic behind

* More Dos and polishing according PR comments
* Mention WebFlux and RSocket endpoints in the `endpoint-summary.adoc`
table

* More reactive streams docs
* `SourcePollingChannelAdapter` polishing
* Wrap multi-value publisher into `Mono.just()` in the `AbstractMessageProducingHandler`
instead of emitting just only a first item for the standard reply

Doc polishing

Author: artembilan

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

@@ -20,6 +20,7 @@
 import org.springframework.integration.dispatcher.LoadBalancingStrategy;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
 import org.springframework.integration.dispatcher.UnicastingDispatcher;
+import org.springframework.lang.Nullable;
 
 /**
  * A channel that invokes a single subscriber for each sent Message.
@@ -30,6 +31,7 @@
  * @author Iwein Fuld
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class DirectChannel extends AbstractSubscribableChannel {
 
@@ -45,20 +47,20 @@ public DirectChannel() {
 	}
 
 	/**
-	 * Create a DirectChannel with a {@link LoadBalancingStrategy}. The
-	 * strategy <em>must not</em> be null.
-	 *
+	 * Create a DirectChannel with a {@link LoadBalancingStrategy}.
+	 * Can be {@code null} meaning that no balancing is applied;
+	 * every message is always going to be handled by the first subscriber.
 	 * @param loadBalancingStrategy The load balancing strategy implementation.
+	 * @see #setFailover(boolean)
 	 */
-	public DirectChannel(LoadBalancingStrategy loadBalancingStrategy) {
+	public DirectChannel(@Nullable LoadBalancingStrategy loadBalancingStrategy) {
 		this.dispatcher.setLoadBalancingStrategy(loadBalancingStrategy);
 	}
 
 
 	/**
 	 * Specify whether the channel's dispatcher should have failover enabled.
 	 * By default, it will. Set this value to 'false' to disable it.
-	 *
 	 * @param failover The failover boolean.
 	 */
 	public void setFailover(boolean failover) {
@@ -68,7 +70,6 @@ public void setFailover(boolean failover) {
 	/**
 	 * Specify the maximum number of subscribers supported by the
 	 * channel's dispatcher.
-	 *
 	 * @param maxSubscribers The maximum number of subscribers allowed.
 	 */
 	public void setMaxSubscribers(int maxSubscribers) {
@@ -85,9 +86,8 @@ protected UnicastingDispatcher getDispatcher() {
 	protected void onInit() {
 		super.onInit();
 		if (this.maxSubscribers == null) {
-			Integer max = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS,
-					Integer.class);
-			this.setMaxSubscribers(max);
+			Integer max = getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class);
+			setMaxSubscribers(max);
 		}
 	}
 

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

@@ -63,7 +63,7 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat
 	public PollingConsumer(PollableChannel inputChannel, MessageHandler handler) {
 		Assert.notNull(inputChannel, "inputChannel must not be null");
 		Assert.notNull(handler, "handler must not be null");
-		if (inputChannel instanceof NullChannel && logger.isWarnEnabled()) {
+		if (inputChannel instanceof NullChannel) {
 			logger.warn("The polling from the NullChannel does not have any effects: " +
 					"it doesn't forward messages sent to it. A NullChannel is the end of the flow.");
 		}
@@ -134,7 +134,7 @@ protected void handleMessage(Message<?> message) {
 		try {
 			if (this.channelInterceptors != null
 					&& ((ExecutorChannelInterceptorAware) this.inputChannel).hasExecutorInterceptors()) {
-				interceptorStack = new ArrayDeque<ExecutorChannelInterceptor>();
+				interceptorStack = new ArrayDeque<>();
 				theMessage = applyBeforeHandle(theMessage, interceptorStack);
 				if (theMessage == null) {
 					return;

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

@@ -147,11 +147,11 @@ protected void applyReceiveOnlyAdviceChain(Collection<Advice> chain) {
 			if (AopUtils.isAopProxy(this.source)) {
 				Advised advised = (Advised) this.source;
 				this.appliedAdvices.forEach(advised::removeAdvice);
-				chain.stream().forEach(advice -> advised.addAdvisor(adviceToReceiveAdvisor(advice)));
+				chain.forEach(advice -> advised.addAdvisor(adviceToReceiveAdvisor(advice)));
 			}
 			else {
 				ProxyFactory proxyFactory = new ProxyFactory(this.source);
-				chain.stream().forEach(advice -> proxyFactory.addAdvisor(adviceToReceiveAdvisor(advice)));
+				chain.forEach(advice -> proxyFactory.addAdvisor(adviceToReceiveAdvisor(advice)));
 				this.source = (MessageSource<?>) proxyFactory.getProxy(getBeanClassLoader());
 			}
 			this.appliedAdvices.clear();

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

@@ -27,6 +27,8 @@
 
 import org.reactivestreams.Publisher;
 
+import org.springframework.core.ReactiveAdapter;
+import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
@@ -346,8 +348,15 @@ private void asyncNonReactiveReply(Message<?> requestMessage, Object reply, Obje
 		}
 		else {
 			SettableListenableFuture<Object> settableListenableFuture = new SettableListenableFuture<>();
-			Mono.from((Publisher<?>) reply)
-					.subscribe(settableListenableFuture::set, settableListenableFuture::setException);
+			Mono<?> reactiveReply;
+			ReactiveAdapter adapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(null, reply);
+			if (adapter != null && adapter.isMultiValue()) {
+				reactiveReply = Mono.just(reply);
+			}
+			else {
+				reactiveReply = Mono.from((Publisher<?>) reply);
+			}
+			reactiveReply.subscribe(settableListenableFuture::set, settableListenableFuture::setException);
 			future = settableListenableFuture;
 		}
 		future.addCallback(new ReplyFutureCallback(requestMessage, replyChannel));