Add IntegrationFlowDefinition.nullChannel() (#2555)

* Add `IntegrationFlowDefinition.nullChannel()`

When a `NullChannel` is used in the middle of the flow, it may be not so
obvious why our flow is stopped after accidentally added the next
endpoint

* For convenient add a terminal `nullChannel()` operator into the
`IntegrationFlowDefinition`

* * Add WARN about `NullChannel` subscription from the endpoints

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

@@ -2961,6 +2961,17 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		return new PublisherIntegrationFlow<>(this.integrationComponents, publisher);
 	}
 
+	/**
+	 * Add a {@value IntegrationContextUtils#NULL_CHANNEL_BEAN_NAME} bean into this flow
+	 * definition as a terminal operator.
+	 * @return The {@link IntegrationFlow} instance based on this definition.
+	 * @since 5.1
+	 */
+	public IntegrationFlow nullChannel() {
+		return channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME)
+				.get();
+	}
+
 	@SuppressWarnings("unchecked")
 	private <S extends ConsumerEndpointSpec<S, ? extends MessageHandler>> B register(S endpointSpec,
 			Consumer<S> endpointConfigurer) {

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

@@ -25,6 +25,7 @@ import org.reactivestreams.Subscriber;
 
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.ExecutorChannelInterceptorAware;
+import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.router.MessageRouter;
@@ -62,6 +63,10 @@ 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()) {
+			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.");
+		}
 		this.inputChannel = inputChannel;
 		this.handler = handler;
 		if (this.inputChannel instanceof ExecutorChannelInterceptorAware) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -25,6 +25,7 @@ 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.channel.NullChannel;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.router.MessageRouter;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
@@ -41,6 +42,7 @@ import reactor.core.publisher.BaseSubscriber;
 
 /**
  * @author Artem Bilan
+ *
  * @since 5.0
  */
 public class ReactiveStreamsConsumer extends AbstractEndpoint implements IntegrationConsumer {
@@ -72,6 +74,11 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
 		Assert.notNull(inputChannel, "'inputChannel' must not be null");
 		Assert.notNull(subscriber, "'subscriber' must not be null");
 
+		if (inputChannel instanceof NullChannel && logger.isWarnEnabled()) {
+			logger.warn("The consuming from the NullChannel does not have any effects: " +
+					"it doesn't forward messages sent to it. A NullChannel is the end of the flow.");
+		}
+
 		this.publisher = MessageChannelReactiveUtils.toPublisher(inputChannel);
 		this.subscriber = subscriber;
 		this.lifecycleDelegate = subscriber instanceof Lifecycle ? (Lifecycle) subscriber : null;

spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java

@@ -56,6 +56,7 @@ import org.springframework.integration.annotation.MessagingGateway;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FixedSubscriberChannel;
+import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.context.IntegrationContextUtils;
@@ -481,6 +482,23 @@ public class IntegrationFlowTests {
 		assertEquals("dedicatedTaskScheduler-1", threadNameReference.get());
 	}
 
+	@Autowired
+	private MessageChannel flowWithNullChannelInput;
+
+	@Autowired
+	private NullChannel nullChannel;
+
+	@Test
+	public void testNullChannelInTheEndOfFlow() {
+		this.nullChannel.setCountsEnabled(true);
+
+		this.flowWithNullChannelInput.send(new GenericMessage<>("foo"));
+
+		assertEquals(1, this.nullChannel.getSendCount());
+
+		this.nullChannel.setCountsEnabled(false);
+	}
+
 	@MessagingGateway
 	public interface ControlBusGateway {
 
@@ -495,7 +513,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow supplierFlow() {
 			return IntegrationFlows.from(() -> "foo")
-					.<String, String>transform(p -> p.toUpperCase())
+					.<String, String>transform(String::toUpperCase)
 					.channel("suppliedChannel")
 					.get();
 		}
@@ -594,7 +612,7 @@ public class IntegrationFlowTests {
 					.channel("foo")
 					.fixedSubscriberChannel()
 					.<String, Integer>transform(Integer::parseInt)
-					.<Integer, Foo>transform(i -> new Foo(i))
+					.transform(Foo::new)
 					.transform(new PayloadSerializingTransformer(),
 							c -> c.autoStartup(false).id("payloadSerializingTransformer"))
 					.channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue"))
@@ -841,6 +859,12 @@ public class IntegrationFlowTests {
 			return new ThreadPoolTaskScheduler();
 		}
 
+		@Bean
+		public IntegrationFlow flowWithNullChannel() {
+			return IntegrationFlows.from("flowWithNullChannelInput")
+					.nullChannel();
+		}
+
 	}
 
 	@Service