GH-2765: Add discardChannel for splitter (#2883)

* GH-2765: Add discardChannel for splitter

Fixes https://github.com/spring-projects/spring-integration/issues/2765

When encountering empty collections, splitter should be able to send
the result to a discard channel.
Currently, when encountering an empty collection,
the splitter ends the flow.
Some use-cases may rely on a custom split function which may returns
empty collections.
These use-cases should be able to define a discard channel
so they can proceed with a possible compensation flow.

* Add `discardChannel` option to the `AbstractMessageSplitter`
* Delegate `discardChannel` population from everywhere it is possible:
DSL, XML, `AbstractMessageSplitter` extension like `FileSplitter` etc.
* Fix `FileSplitterTests` for broken charset
* Document new feature; fix some typos and out-dated code sample

* * Fix `SplitterFactoryBean` for NPE on the `discardChannelName`
propagation

* * Check `this.discardChannel` first
* `Assert.state()` in `doInit()` for mutual exclusiveness

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

@@ -23,6 +23,7 @@ import org.springframework.integration.splitter.AbstractMessageSplitter;
 import org.springframework.integration.splitter.DefaultMessageSplitter;
 import org.springframework.integration.splitter.ExpressionEvaluatingSplitter;
 import org.springframework.integration.splitter.MethodInvokingSplitter;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -42,6 +43,10 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe
 
 	private String delimiters;
 
+	private MessageChannel discardChannel;
+
+	private String discardChannelName;
+
 	public void setApplySequence(boolean applySequence) {
 		this.applySequence = applySequence;
 	}
@@ -50,6 +55,14 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe
 		this.delimiters = delimiters;
 	}
 
+	public void setDiscardChannel(MessageChannel discardChannel) {
+		this.discardChannel = discardChannel;
+	}
+
+	public void setDiscardChannelName(String discardChannelName) {
+		this.discardChannelName = discardChannelName;
+	}
+
 	@Override
 	protected MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) {
 		Assert.notNull(targetObject, "targetObject must not be null");
@@ -90,6 +103,12 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe
 
 	protected AbstractMessageSplitter configureSplitter(AbstractMessageSplitter splitter) {
 		postProcessReplyProducer(splitter);
+		if (this.discardChannel != null) {
+			splitter.setDiscardChannel(this.discardChannel);
+		}
+		else if (StringUtils.hasText(this.discardChannelName)) {
+			splitter.setDiscardChannelName(this.discardChannelName);
+		}
 		return splitter;
 	}
 

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

@@ -28,6 +28,7 @@ import org.springframework.integration.config.SplitterFactoryBean;
  * @author Mark Fisher
  * @author Iwein Fuld
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class SplitterParser extends AbstractDelegatingConsumerEndpointParser {
 
@@ -45,6 +46,7 @@ public class SplitterParser extends AbstractDelegatingConsumerEndpointParser {
 	void postProcess(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delimiters");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "discard-channel", "discardChannelName");
 	}
 
 }

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

@@ -18,6 +18,7 @@ package org.springframework.integration.dsl;
 
 import org.springframework.integration.splitter.AbstractMessageSplitter;
 import org.springframework.integration.splitter.DefaultMessageSplitter;
+import org.springframework.messaging.MessageChannel;
 
 /**
  * A {@link ConsumerEndpointSpec} for a {@link AbstractMessageSplitter} implementations.
@@ -43,7 +44,7 @@ public final class SplitterEndpointSpec<S extends AbstractMessageSplitter>
 	 */
 	public SplitterEndpointSpec<S> applySequence(boolean applySequence) {
 		this.handler.setApplySequence(applySequence);
-		return _this();
+		return this;
 	}
 
 	/**
@@ -65,4 +66,45 @@ public final class SplitterEndpointSpec<S extends AbstractMessageSplitter>
 		return this;
 	}
 
+	/**
+	 * Specify a channel where rejected Messages should be sent. If the discard
+	 * channel is null (the default), rejected Messages will be dropped.
+	 * A "Rejected Message" means that split function has returned an empty result (but not null):
+	 * no items to iterate for sending.
+	 * @param discardChannel The discard channel.
+	 * @return the endpoint spec.
+	 * @since 5.2
+	 * @see DefaultMessageSplitter#setDelimiters(String)
+	 */
+	public SplitterEndpointSpec<S> discardChannel(MessageChannel discardChannel) {
+		this.handler.setDiscardChannel(discardChannel);
+		return this;
+	}
+
+	/**
+	 * Specify a channel bean name where rejected Messages should be sent. If the discard
+	 * channel is null (the default), rejected Messages will be dropped.
+	 * A "Rejected Message" means that split function has returned an empty result (but not null):
+	 * no items to iterate for sending.
+	 * @param discardChannelName The discard channel bean name.
+	 * @return the endpoint spec.
+	 * @since 5.2
+	 * @see DefaultMessageSplitter#setDelimiters(String)
+	 */
+	public SplitterEndpointSpec<S> discardChannel(String discardChannelName) {
+		this.handler.setDiscardChannelName(discardChannelName);
+		return this;
+	}
+
+	/**
+	 * Configure a subflow to run for discarded messages instead of a
+	 * {@link #discardChannel(MessageChannel)}.
+	 * @param discardFlow the discard flow.
+	 * @return the endpoint spec.
+	 * @since 5.2
+	 */
+	public SplitterEndpointSpec<S> discardFlow(IntegrationFlow discardFlow) {
+		return discardChannel(obtainInputChannelFromFlow(discardFlow));
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java

@@ -30,13 +30,17 @@ import org.reactivestreams.Publisher;
 
 import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.DiscardingMessageHandler;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.json.JacksonPresent;
 import org.springframework.integration.util.FunctionIterator;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
 
 import com.fasterxml.jackson.core.TreeNode;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 /**
  * Base class for Message-splitting handlers.
@@ -47,10 +51,15 @@ import reactor.core.publisher.Flux;
  * @author Ruslan Stelmachenko
  * @author Gary Russell
  */
-public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
+public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler
+		implements DiscardingMessageHandler {
 
 	private boolean applySequence = true;
 
+	private MessageChannel discardChannel;
+
+	private String discardChannelName;
+
 	/**
 	 * Set the applySequence flag to the specified value. Defaults to true.
 	 * @param applySequence true to apply sequence information.
@@ -59,10 +68,54 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 		this.applySequence = applySequence;
 	}
 
+	/**
+	 * Specify a channel where rejected Messages should be sent. If the discard
+	 * channel is null (the default), rejected Messages will be dropped.
+	 * A "Rejected Message" means that split function has returned an empty result (but not null):
+	 * no items to iterate for sending.
+	 * @param discardChannel The discard channel.
+	 * @since 5.2
+	 */
+	public void setDiscardChannel(MessageChannel discardChannel) {
+		this.discardChannel = discardChannel;
+	}
+
+	/**
+	 * Specify a channel bean name (resolved to {@link MessageChannel} lazily)
+	 * where rejected Messages should be sent. If the discard
+	 * channel is null (the default), rejected Messages will be dropped.
+	 * A "Rejected Message" means that split function has returned an empty result (but not null):
+	 * no items to iterate for sending.
+	 * @param discardChannelName The discard channel bean name.
+	 * @since 5.2
+	 */
+	public void setDiscardChannelName(String discardChannelName) {
+		Assert.hasText(discardChannelName, "'discardChannelName' must not be empty");
+		this.discardChannelName = discardChannelName;
+	}
+
+	@Override
+	public MessageChannel getDiscardChannel() {
+		if (this.discardChannel == null) {
+			String channelName = this.discardChannelName;
+			if (channelName != null) {
+				this.discardChannel = getChannelResolver().resolveDestination(channelName);
+				this.discardChannelName = null;
+			}
+		}
+		return this.discardChannel;
+	}
+
+	@Override
+	protected void doInit() {
+		Assert.state(!(this.discardChannelName != null && this.discardChannel != null),
+				"'discardChannelName' and 'discardChannel' are mutually exclusive.");
+	}
+
 	@Override
 	@SuppressWarnings("unchecked")
 	protected final Object handleRequestMessage(Message<?> message) {
-		Object result = this.splitMessage(message);
+		Object result = splitMessage(message);
 		// return null if 'null'
 		if (result == null) {
 			return null;
@@ -137,6 +190,10 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 		}
 
 		if (iterator != null && !iterator.hasNext()) {
+			MessageChannel discardingChannel = getDiscardChannel();
+			if (discardingChannel != null) {
+				this.messagingTemplate.send(discardingChannel, message);
+			}
 			return null;
 		}
 
@@ -154,7 +211,16 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 				object -> createBuilder(object, headers, correlationId, sequenceNumber.getAndIncrement(), sequenceSize);
 
 		if (reactive) {
-			return flux.map(messageBuilderFunction);
+			return flux
+					.map(messageBuilderFunction)
+					.switchIfEmpty(
+							Mono.defer(() -> {
+								MessageChannel discardingChannel = getDiscardChannel();
+								if (discardingChannel != null) {
+									this.messagingTemplate.send(discardingChannel, message);
+								}
+								return Mono.empty();
+							}));
 		}
 		else {
 			return new FunctionIterator<>(result instanceof AutoCloseable && !result.equals(iterator)
@@ -195,6 +261,7 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 
 	private AbstractIntegrationMessageBuilder<?> createBuilder(Object item, Map<String, Object> headers,
 			Object correlationId, int sequenceNumber, int sequenceSize) {
+
 		AbstractIntegrationMessageBuilder<?> builder;
 		if (item instanceof Message) {
 			builder = getMessageBuilderFactory().fromMessage((Message<?>) item);

spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.2.xsd

@@ -3669,6 +3669,19 @@
 						</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
+				<xsd:attribute name="discard-channel" type="xsd:string">
+					<xsd:annotation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type type="org.springframework.messaging.MessageChannel" />
+							</tool:annotation>
+						</xsd:appinfo>
+						<xsd:documentation>
+							The channel where the splitter will send the messages that return an empty container from
+							split function.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>

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

@@ -21,7 +21,6 @@ import static org.assertj.core.api.Assertions.assertThat;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-import java.util.concurrent.Executor;
 import java.util.stream.Collectors;
 
 import org.junit.Test;
@@ -31,6 +30,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
 import org.springframework.integration.channel.QueueChannel;
@@ -47,10 +47,10 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
+import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.TextNode;
 
 /**
@@ -90,6 +90,9 @@ public class CorrelationHandlerTests {
 	@Autowired
 	private PollableChannel releaseChannel;
 
+	@Autowired
+	private PollableChannel discardChannel;
+
 	@Test
 	public void testSplitterResequencer() {
 		QueueChannel replyChannel = new QueueChannel();
@@ -150,17 +153,24 @@ public class CorrelationHandlerTests {
 		assertThat(out.getPayload()).isEqualTo("bar");
 	}
 
+	@Test
+	public void testSplitterDiscard() {
+		this.splitAggregateInput.send(new GenericMessage<>(new ArrayList<>()));
+		Message<?> receive = this.discardChannel.receive(10_000);
+		assertThat(receive)
+				.isNotNull()
+				.extracting(Message::getPayload)
+				.isInstanceOf(ArrayNode.class)
+				.extracting("_children")
+				.element(0)
+				.asList()
+				.hasSize(0);
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {
 
-		@Bean
-		public Executor taskExecutor() {
-			ThreadPoolTaskExecutor tpte = new ThreadPoolTaskExecutor();
-			tpte.setCorePoolSize(50);
-			return tpte;
-		}
-
 		@Bean
 		public TestSplitterPojo testSplitterData() {
 			List<String> first = new ArrayList<>();
@@ -175,22 +185,22 @@ public class CorrelationHandlerTests {
 		}
 
 		@Bean
-		public MessageChannelSpec<?, ?> executorChannel() {
-			return MessageChannels.executor(taskExecutor());
+		public MessageChannelSpec<?, ?> executorChannel(TaskExecutor taskExecutor) {
+			return MessageChannels.executor(taskExecutor);
 		}
 
 		@Bean
 		@SuppressWarnings("rawtypes")
-		public IntegrationFlow splitResequenceFlow(MessageChannel executorChannel) {
+		public IntegrationFlow splitResequenceFlow(MessageChannel executorChannel, TaskExecutor taskExecutor) {
 			return f -> f.enrichHeaders(s -> s.header("FOO", "BAR"))
 					.split("testSplitterData", "buildList", c -> c.applySequence(false))
 					.channel(executorChannel)
 					.split(Message.class, Message::getPayload, c -> c.applySequence(false))
-					.channel(MessageChannels.executor(taskExecutor()))
+					.channel(MessageChannels.executor(taskExecutor))
 					.split(s -> s
 							.applySequence(false)
 							.delimiters(","))
-					.channel(MessageChannels.executor(taskExecutor()))
+					.channel(MessageChannels.executor(taskExecutor))
 					.<String, Integer>transform(Integer::parseInt)
 					.enrichHeaders(h ->
 							h.headerFunction(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, Message::getPayload))
@@ -203,8 +213,9 @@ public class CorrelationHandlerTests {
 		public IntegrationFlow splitAggregateFlow() {
 			return IntegrationFlows.from("splitAggregateInput", true)
 					.transform(Transformers.toJson(ObjectToJsonTransformer.ResultType.NODE))
-					.split()
-					.channel(MessageChannels.executor(taskExecutor()))
+					.split((splitter) -> splitter
+							.discardFlow((subFlow) -> subFlow.channel((c) -> c.queue("discardChannel"))))
+					.channel(MessageChannels.flux())
 					.resequence()
 					.aggregate()
 					.get();

spring-integration-file/src/main/java/org/springframework/integration/file/config/FileSplitterParser.java

@@ -45,6 +45,7 @@ public class FileSplitterParser extends AbstractConsumerEndpointParser {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "first-line-as-header");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "discard-channel", "discardChannelName");
 		return builder;
 	}
 

spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.2.xsd

@@ -806,6 +806,19 @@ Only files matching this regular expression will be picked up by this adapter.
 					<xsd:union memberTypes="xsd:boolean xsd:string"/>
 				</xsd:simpleType>
 			</xsd:attribute>
+			<xsd:attribute name="discard-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.messaging.MessageChannel" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						The channel where the splitter will send the messages that return an empty container from
+						split function.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 		</xsd:complexType>
 	</xsd:element>
 

spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests-context.xml

@@ -22,6 +22,7 @@
 					   send-timeout="5"
 					   auto-startup="false"
 					   order="2"
-					   phase="1"/>
+					   phase="1"
+					   discard-channel="nullChannel"/>
 
 </beans>

spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests.java

@@ -66,6 +66,7 @@ public class FileSplitterParserTests {
 		assertThat(TestUtils.getPropertyValue(this.splitter, "charset")).isEqualTo(Charset.forName("UTF-8"));
 		assertThat(TestUtils.getPropertyValue(this.splitter, "messagingTemplate.sendTimeout")).isEqualTo(5L);
 		assertThat(TestUtils.getPropertyValue(this.splitter, "firstLineHeaderName")).isEqualTo("foo");
+		assertThat(TestUtils.getPropertyValue(this.splitter, "discardChannelName")).isEqualTo("nullChannel");
 		assertThat(this.splitter.getOutputChannel()).isSameAs(this.out);
 		assertThat(this.splitter.getOrder()).isEqualTo(2);
 		assertThat(this.fullBoat.getInputChannel()).isSameAs(this.in);

spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java

@@ -28,6 +28,7 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.io.Reader;
 import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.Date;
 import java.util.List;
 
@@ -80,7 +81,7 @@ public class FileSplitterTests {
 
 	private static File file;
 
-	private static final String SAMPLE_CONTENT = "HelloWorld\n????";
+	private static final String SAMPLE_CONTENT = "HelloWorld\näöüß";
 
 	@Autowired
 	private MessageChannel input1;
@@ -97,7 +98,7 @@ public class FileSplitterTests {
 	@BeforeClass
 	public static void setup() throws IOException {
 		file = File.createTempFile("foo", ".txt");
-		FileCopyUtils.copy(SAMPLE_CONTENT.getBytes("UTF-8"),
+		FileCopyUtils.copy(SAMPLE_CONTENT.getBytes(StandardCharsets.UTF_8),
 				new FileOutputStream(file, false));
 	}
 
@@ -108,24 +109,24 @@ public class FileSplitterTests {
 
 	@Test
 	public void testFileSplitter() throws Exception {
-		this.input1.send(new GenericMessage<File>(file));
+		this.input1.send(new GenericMessage<>(file));
 		Message<?> receive = this.output.receive(10000);
 		assertThat(receive).isNotNull(); //HelloWorld
 		assertThat(receive.getPayload()).isEqualTo("HelloWorld");
 		assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(2);
 		receive = this.output.receive(10000);
-		assertThat(receive).isNotNull(); //????
-		assertThat(receive.getPayload()).isEqualTo("????");
+		assertThat(receive).isNotNull();  //äöüß
+		assertThat(receive.getPayload()).isEqualTo("äöüß");
 		assertThat(receive.getHeaders().get(FileHeaders.ORIGINAL_FILE)).isEqualTo(file);
 		assertThat(receive.getHeaders().get(FileHeaders.FILENAME)).isEqualTo(file.getName());
 		assertThat(this.output.receive(1)).isNull();
 
-		this.input1.send(new GenericMessage<String>(file.getAbsolutePath()));
+		this.input1.send(new GenericMessage<>(file.getAbsolutePath()));
 		receive = this.output.receive(10000);
 		assertThat(receive).isNotNull(); //HelloWorld
 		assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(2);
 		receive = this.output.receive(10000);
-		assertThat(receive).isNotNull(); //????
+		assertThat(receive).isNotNull(); //äöüß
 		assertThat(receive.getHeaders().get(FileHeaders.ORIGINAL_FILE)).isEqualTo(file);
 		assertThat(receive.getHeaders().get(FileHeaders.FILENAME)).isEqualTo(file.getName());
 		assertThat(this.output.receive(1)).isNull();
@@ -135,54 +136,56 @@ public class FileSplitterTests {
 		assertThat(receive).isNotNull(); //HelloWorld
 		assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(2);
 		receive = this.output.receive(10000);
-		assertThat(receive).isNotNull(); //????
+		assertThat(receive).isNotNull(); //äöüß
 		assertThat(this.output.receive(1)).isNull();
 
-		this.input2.send(new GenericMessage<File>(file));
+		this.input2.send(new GenericMessage<>(file));
 		receive = this.output.receive(10000);
 		assertThat(receive).isNotNull(); //HelloWorld
 		assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(0);
 		receive = this.output.receive(10000);
-		assertThat(receive).isNotNull(); //????
+		assertThat(receive).isNotNull(); //äöüß
 		assertThat(this.output.receive(1)).isNull();
 
-		this.input2.send(new GenericMessage<InputStream>(new ByteArrayInputStream(SAMPLE_CONTENT.getBytes("UTF-8"))));
+		this.input2.send(new GenericMessage<InputStream>(
+				new ByteArrayInputStream(SAMPLE_CONTENT.getBytes(StandardCharsets.UTF_8))));
 		receive = this.output.receive(10000);
 		assertThat(receive).isNotNull(); //HelloWorld
 		assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(0);
 		receive = this.output.receive(10000);
-		assertThat(receive).isNotNull(); //????
+		assertThat(receive).isNotNull(); //äöüß
 		assertThat(this.output.receive(1)).isNull();
 
 		try {
-			this.input2.send(new GenericMessage<String>("bar"));
+			this.input2.send(new GenericMessage<>("bar"));
 			fail("FileNotFoundException expected");
 		}
 		catch (Exception e) {
 			assertThat(e.getCause()).isInstanceOf(FileNotFoundException.class);
 			assertThat(e.getMessage()).contains("failed to read file [bar]");
 		}
-		this.input2.send(new GenericMessage<Date>(new Date()));
+		this.input2.send(new GenericMessage<>(new Date()));
 		receive = this.output.receive(10000);
 		assertThat(receive).isNotNull();
 		assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(1);
 		assertThat(receive.getPayload()).isInstanceOf(Date.class);
 		assertThat(this.output.receive(1)).isNull();
 
-		this.input3.send(new GenericMessage<File>(file));
+		this.input3.send(new GenericMessage<>(file));
 		receive = this.output.receive(10000);
 		assertThat(receive).isNotNull(); //HelloWorld
 		assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(0);
 		receive = this.output.receive(10000);
-		assertThat(receive).isNotNull(); //????
+		assertThat(receive).isNotNull(); //äöüß
 		assertThat(this.output.receive(1)).isNull();
 
-		this.input3.send(new GenericMessage<InputStream>(new ByteArrayInputStream(SAMPLE_CONTENT.getBytes("UTF-8"))));
+		this.input3.send(new GenericMessage<>(
+				new ByteArrayInputStream(SAMPLE_CONTENT.getBytes(StandardCharsets.UTF_8))));
 		receive = this.output.receive(10000);
 		assertThat(receive).isNotNull(); //HelloWorld
 		assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(0);
 		receive = this.output.receive(10000);
-		assertThat(receive).isNotNull(); //????
+		assertThat(receive).isNotNull(); //äöüß
 		assertThat(this.output.receive(1)).isNull();
 	}
 
@@ -191,7 +194,7 @@ public class FileSplitterTests {
 		QueueChannel outputChannel = new QueueChannel();
 		FileSplitter splitter = new FileSplitter(true, true);
 		splitter.setOutputChannel(outputChannel);
-		splitter.handleMessage(new GenericMessage<File>(file));
+		splitter.handleMessage(new GenericMessage<>(file));
 		Message<?> received = outputChannel.receive(0);
 		assertThat(received).isNotNull();
 		assertThat(received.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isNull();
@@ -218,7 +221,7 @@ public class FileSplitterTests {
 		FileSplitter splitter = new FileSplitter(true, true);
 		splitter.setOutputChannel(outputChannel);
 		File file = File.createTempFile("empty", ".txt");
-		splitter.handleMessage(new GenericMessage<File>(file));
+		splitter.handleMessage(new GenericMessage<>(file));
 		Message<?> received = outputChannel.receive(0);
 		assertThat(received).isNotNull();
 		assertThat(received.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isNull();
@@ -246,7 +249,7 @@ public class FileSplitterTests {
 		QueueChannel outputChannel = new QueueChannel();
 		FileSplitter splitter = new FileSplitter(true, true, true);
 		splitter.setOutputChannel(outputChannel);
-		splitter.handleMessage(new GenericMessage<File>(file));
+		splitter.handleMessage(new GenericMessage<>(file));
 		Message<?> received = outputChannel.receive(0);
 		assertThat(received).isNotNull();
 		assertThat(received.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isNull();
@@ -263,7 +266,7 @@ public class FileSplitterTests {
 		assertThat(received).isNotNull();
 		assertThat(received.getHeaders().get(FileHeaders.MARKER)).isEqualTo("END");
 		assertThat(received.getPayload()).isInstanceOf(String.class);
-		fileMarker = objectMapper.fromJson((String) received.getPayload(), FileSplitter.FileMarker.class);
+		fileMarker = objectMapper.fromJson(received.getPayload(), FileSplitter.FileMarker.class);
 		assertThat(fileMarker.getMark()).isEqualTo(FileSplitter.FileMarker.Mark.END);
 		assertThat(fileMarker.getFilePath()).isEqualTo(file.getAbsolutePath());
 		assertThat(fileMarker.getLineCount()).isEqualTo(2);

spring-integration-xml/src/main/java/org/springframework/integration/xml/config/XPathMessageSplitterParser.java

@@ -42,7 +42,8 @@ public class XPathMessageSplitterParser extends AbstractConsumerEndpointParser {
 		Assert.isTrue(xPathExpressionNodes.getLength() <= 1, "At most one xpath-expression child may be specified.");
 		boolean hasChild = xPathExpressionNodes.getLength() == 1;
 		boolean hasReference = StringUtils.hasText(xPathExpressionRef);
-		Assert.isTrue(hasChild ^ hasReference, "Exactly one of 'xpath-expression' or 'xpath-expression-ref' is required.");
+		Assert.isTrue(hasChild ^ hasReference,
+				"Exactly one of 'xpath-expression' or 'xpath-expression-ref' is required.");
 		if (hasChild) {
 			Element xpathExpressionElement = (Element) xPathExpressionNodes.item(0);
 			builder.addConstructorArgValue(xpathExpressionElement.getAttribute("expression"));
@@ -51,12 +52,14 @@ public class XPathMessageSplitterParser extends AbstractConsumerEndpointParser {
 		else {
 			builder.addConstructorArgReference(xPathExpressionRef);
 		}
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "doc-builder-factory", "documentBuilder");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "doc-builder-factory",
+				"documentBuilder");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "create-documents");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "iterator");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "output-properties");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "discard-channel", "discardChannelName");
 		return builder;
 	}
 

spring-integration-xml/src/main/resources/org/springframework/integration/xml/config/spring-integration-xml-5.2.xsd

@@ -820,6 +820,19 @@
 							<xsd:union memberTypes="xsd:boolean xsd:string" />
 						</xsd:simpleType>
 					</xsd:attribute>
+					<xsd:attribute name="discard-channel" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="org.springframework.messaging.MessageChannel" />
+								</tool:annotation>
+							</xsd:appinfo>
+							<xsd:documentation>
+								The channel where the splitter will send the messages that return an empty container from
+								split function.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

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

@@ -30,7 +30,8 @@
 					apply-sequence="false"
 					create-documents="true"
 					output-properties="outputProperties"
-					iterator="false">
+					iterator="false"
+					discard-channel="nullChannel">
 		<xpath-expression expression="/orders/order"/>
 	</xpath-splitter>
 

spring-integration-xml/src/test/java/org/springframework/integration/xml/config/XPathSplitterParserTests.java

@@ -45,13 +45,16 @@ import org.springframework.util.MultiValueMap;
 @DirtiesContext
 public class XPathSplitterParserTests {
 
-	@Autowired @Qualifier("xpathSplitter.handler")
+	@Autowired
+	@Qualifier("xpathSplitter.handler")
 	private MessageHandler xpathSplitter;
 
-	@Autowired @Qualifier("xpathSplitter")
+	@Autowired
+	@Qualifier("xpathSplitter")
 	private EventDrivenConsumer consumer;
 
-	@Autowired @Qualifier("outputProperties")
+	@Autowired
+	@Qualifier("outputProperties")
 	private Properties outputProperties;
 
 	@Autowired
@@ -64,10 +67,11 @@ public class XPathSplitterParserTests {
 		assertThat(TestUtils.getPropertyValue(this.xpathSplitter, "returnIterator", Boolean.class)).isFalse();
 		assertThat(TestUtils.getPropertyValue(this.xpathSplitter, "outputProperties")).isSameAs(this.outputProperties);
 		assertThat(TestUtils.getPropertyValue(this.xpathSplitter,
-				"xpathExpression.xpathExpression.xpath.m_patternString",
-				String.class)).isEqualTo("/orders/order");
+				"xpathExpression.xpathExpression.xpath.m_patternString", String.class))
+				.isEqualTo("/orders/order");
 		assertThat(TestUtils.getPropertyValue(xpathSplitter, "order")).isEqualTo(2);
 		assertThat(TestUtils.getPropertyValue(xpathSplitter, "messagingTemplate.sendTimeout")).isEqualTo(123L);
+		assertThat(TestUtils.getPropertyValue(this.xpathSplitter, "discardChannelName")).isEqualTo("nullChannel");
 		assertThat(TestUtils.getPropertyValue(consumer, "phase")).isEqualTo(-1);
 		assertThat(TestUtils.getPropertyValue(consumer, "autoStartup", Boolean.class)).isFalse();
 		@SuppressWarnings("unchecked")

src/reference/asciidoc/dsl.adoc

@@ -455,15 +455,13 @@ The following example shows how to use the `split()` method by providing a lambd
 @Bean
 public IntegrationFlow splitFlow() {
     return IntegrationFlows.from("splitInput")
-              .split(s ->
-                      s.applySequence(false).get().getT2().setDelimiters(","))
-              .channel(MessageChannels.executor(this.taskExecutor()))
+              .split(s -> s.applySequence(false).delimiters(","))
+              .channel(MessageChannels.executor(taskExecutor()))
               .get();
 }
 ----
 
 The preceding example creates a splitter that splits a message containing a comma-delimited `String`.
-Note: The `getT2()` method comes from a `Tuple` `Collection`, which is the result of `EndpointSpec.get()`, and represents a pair of `ConsumerEndpointFactoryBean` and `DefaultMessageSplitter` for the preceding example.
 
 Also see <<java-dsl-class-cast>>.
 
@@ -1157,7 +1155,7 @@ Otherwise, creating such a configuration by using `IntegrationFlow` does not mak
 By default a `GatewayProxyFactoryBean` gets a conventional bean name, such as `[FLOW_BEAN_NAME.gateway]`.
 You can change that ID by using the `@MessagingGateway.name()` attribute or the overloaded `from(Class<?> serviceInterface, String beanName)` factory method.
 
-With Java 8, you can even create an integration fateway with the `java.util.function` interfaces, as the following example shows:
+With Java 8, you can even create an integration gateway with the `java.util.function` interfaces, as the following example shows:
 
 ====
 [source,java]

src/reference/asciidoc/splitter.adoc

@@ -64,6 +64,10 @@ In this case, the target `Iterator` is built on their iteration functionality.
 
 In addition, if the splitter's output channel is an instance of a `ReactiveStreamsSubscribableChannel`, the `AbstractMessageSplitter` produces a `Flux` result instead of an `Iterator`, and the output channel is subscribed to this `Flux` for back-pressure-based splitting on downstream flow demand.
 
+Starting with version 5.2, the splitter supports a `discardChannel` option for sending those request messages for which a split function has returned an empty container (collection, array, stream, `Flux` etc.).
+In this case there is just no item to iterate for sending to the `outputChannel`.
+The `null` splitting result remains as an end of flow indicator.
+
 ==== Configuring a Splitter with XML
 
 A splitter can be configured through XML as follows:
@@ -73,11 +77,12 @@ A splitter can be configured through XML as follows:
 ----
 <int:channel id="inputChannel"/>
 
-<int:splitter id="splitter"         <1>
-  ref="splitterBean"                <2>
-  method="split"                    <3>
-  input-channel="inputChannel"      <4>
-  output-channel="outputChannel" /> <5>
+<int:splitter id="splitter"           <1>
+  ref="splitterBean"                  <2>
+  method="split"                      <3>
+  input-channel="inputChannel"        <4>
+  output-channel="outputChannel"      <5>
+  discard-channel="discardChannel" /> <6>
 
 <int:channel id="outputChannel"/>
 
@@ -94,6 +99,8 @@ Optional.
 Required.
 <5> The channel to which the splitter sends the results of splitting the incoming message.
 Optional (because incoming messages can specify a reply channel themselves).
+<6> The channel to which the request message is sent in case of empty splitting result.
+Optional (the will stop as in case of `null` result).
 ====
 
 We recommend using a `ref` attribute if the custom splitter implementation can be referenced in other `<splitter>` definitions.

src/reference/asciidoc/whats-new.adoc

@@ -19,6 +19,9 @@ See <<rate-limiter-advice>> for more information.
 The `JsonToObjectTransformer` now supports generics for the target object to deserialize into.
 See <<json-transformers>> for more information.
 
+RThe `splitter` now supports a `discardChannel` configuration option.
+See <<splitter>> for more information.
+
 [[x5.2-amqp]]
 ==== AMQP Changes