INT-4269: Add Stream & Flux Support for Splitter

JIRA: https://jira.spring.io/browse/INT-4269

* Check if `outputChannel` is `ReactiveStreamsSubscribableChannel`
to let back-pressure splitting
* Build `Flux` or `Iterator` depending in the `reactive` state
* Allow to get a size of the `iterator` if it is possible,
for example `XPathMessageSplitter`

Add tests

Fix raw type and unused import

* Add JavaDocs to the `AbstractMessageSplitter#obtainSizeIfPossible()`
* Add asserts for the `sequenceSize` populataiton in the `XPathMessageSplitter`
* Document `Stream` & `Flux` support in the splitter

Minor doc polishing

spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.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.
@@ -23,12 +23,19 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.stream.Stream;
 
+import org.reactivestreams.Publisher;
+
+import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.util.FunctionIterator;
 import org.springframework.messaging.Message;
 
+import reactor.core.publisher.Flux;
+
 /**
  * Base class for Message-splitting handlers.
  *
@@ -57,32 +64,75 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 			return null;
 		}
 
-		Iterator<Object> iterator;
+		boolean reactive = getOutputChannel() instanceof ReactiveStreamsSubscribableChannel;
+		setAsync(reactive);
+
+		Iterator<Object> iterator = null;
+		Flux<Object> flux = null;
+
 		final int sequenceSize;
-		if (result instanceof Collection) {
-			Collection<Object> items = (Collection<Object>) result;
-			sequenceSize = items.size();
-			iterator = items.iterator();
+
+		if (result instanceof Iterable<?>) {
+			Iterable<Object> iterable = (Iterable<Object>) result;
+			sequenceSize = obtainSizeIfPossible(iterable);
+			if (reactive) {
+				flux = Flux.fromIterable(iterable);
+			}
+			else {
+				iterator = iterable.iterator();
+			}
 		}
 		else if (result.getClass().isArray()) {
 			Object[] items = (Object[]) result;
 			sequenceSize = items.length;
-			iterator = Arrays.asList(items).iterator();
-		}
-		else if (result instanceof Iterable<?>) {
-			sequenceSize = 0;
-			iterator = ((Iterable<Object>) result).iterator();
+			if (reactive) {
+				flux = Flux.fromArray(items);
+			}
+			else {
+				iterator = Arrays.asList(items).iterator();
+			}
 		}
 		else if (result instanceof Iterator<?>) {
+			Iterator<Object> iter = (Iterator<Object>) result;
+			sequenceSize = obtainSizeIfPossible(iter);
+			if (reactive) {
+				flux = Flux.fromIterable(() -> iter);
+			}
+			else {
+				iterator = iter;
+			}
+		}
+		else if (result instanceof Stream<?>) {
+			Stream<Object> stream = ((Stream<Object>) result);
 			sequenceSize = 0;
-			iterator = (Iterator<Object>) result;
+			if (reactive) {
+				flux = Flux.fromStream(stream);
+			}
+			else {
+				iterator = stream.iterator();
+			}
+		}
+		else if (result instanceof Publisher<?>) {
+			Publisher<Object> publisher = (Publisher<Object>) result;
+			sequenceSize = 0;
+			if (reactive) {
+				flux = Flux.from(publisher);
+			}
+			else {
+				iterator = Flux.from((Publisher<Object>) result).toIterable().iterator();
+			}
 		}
 		else {
 			sequenceSize = 1;
-			iterator = Collections.singleton(result).iterator();
+			if (reactive) {
+				flux = Flux.just(result);
+			}
+			else {
+				iterator = Collections.singleton(result).iterator();
+			}
 		}
 
-		if (!iterator.hasNext()) {
+		if (iterator != null && !iterator.hasNext()) {
 			return null;
 		}
 
@@ -91,13 +141,42 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 			messageHeaders = new HashMap<>(messageHeaders);
 			addHeaders(message, messageHeaders);
 		}
+
 		final Map<String, Object> headers = messageHeaders;
 		final Object correlationId = message.getHeaders().getId();
 		final AtomicInteger sequenceNumber = new AtomicInteger(1);
 
-		return new FunctionIterator<Object, AbstractIntegrationMessageBuilder<?>>(iterator,
-				object ->
-						createBuilder(object, headers, correlationId, sequenceNumber.getAndIncrement(), sequenceSize));
+		Function<Object, AbstractIntegrationMessageBuilder<?>> messageBuilderFunction =
+				object -> createBuilder(object, headers, correlationId, sequenceNumber.getAndIncrement(), sequenceSize);
+
+		if (reactive) {
+			return flux.map(messageBuilderFunction);
+		}
+		else {
+			return new FunctionIterator<>(iterator, messageBuilderFunction);
+		}
+	}
+
+	/**
+	 * Obtain a size of the provided {@link Iterable}. Default implementation returns
+	 * {@link Collection#size()} if the iterable is a collection, or {@code 0} otherwise.
+	 * @param iterable the {@link Iterable} to obtain the size
+	 * @return the size of the {@link Iterable}
+	 * @since 5.0
+	 */
+	protected int obtainSizeIfPossible(Iterable<?> iterable) {
+		return iterable instanceof Collection ? ((Collection<?>) iterable).size() : 0;
+	}
+
+	/**
+	 * Obtain a size of the provided {@link Iterator}.
+	 * Default implementation returns {@code 0}.
+	 * @param iterator the {@link Iterator} to obtain the size
+	 * @return the size of the {@link Iterator}
+	 * @since 5.0
+	 */
+	protected int obtainSizeIfPossible(Iterator<?> iterator) {
+		return 0;
 	}
 
 	private AbstractIntegrationMessageBuilder<?> createBuilder(Object item, Map<String, Object> headers,
@@ -146,10 +225,15 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 
 	@Override
 	protected void produceOutput(Object result, Message<?> requestMessage) {
-		Iterator<?> iterator = (Iterator<?>) result;
-		while (iterator.hasNext()) {
-			super.produceOutput(iterator.next(), requestMessage);
+		if (result instanceof Iterator<?>) {
+			Iterator<?> iterator = (Iterator<?>) result;
+			while (iterator.hasNext()) {
+				super.produceOutput(iterator.next(), requestMessage);
 
+			}
+		}
+		else {
+			super.produceOutput(result, requestMessage);
 		}
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-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.
@@ -26,7 +26,7 @@ import java.util.function.Function;
  * @author Artem Bilan
  * @since 4.1
  */
-public final class FunctionIterator<T, V> implements Iterator<V> {
+public class FunctionIterator<T, V> implements Iterator<V> {
 
 	private final Iterator<T> iterator;
 

spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 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,29 +16,40 @@
 
 package org.springframework.integration.splitter;
 
-import static org.junit.Assert.assertEquals;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.stream.Stream;
 
 import org.junit.Test;
-import org.springframework.messaging.Message;
+import org.reactivestreams.Subscriber;
+
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
 
 /**
  * @author Mark Fisher
  * @author Iwein Fuld
  * @author Gunnar Hillert
+ * @author Artem Bilan
  */
 public class DefaultSplitterTests {
 
@@ -65,7 +76,7 @@ public class DefaultSplitterTests {
 
 	@Test
 	public void splitMessageWithCollectionPayload() throws Exception {
-		List<String> payload = Arrays.asList(new String[] { "x", "y", "z" });
+		List<String> payload = Arrays.asList("x", "y", "z");
 		Message<List<String>> message = MessageBuilder.withPayload(payload).build();
 		QueueChannel replyChannel = new QueueChannel();
 		DefaultMessageSplitter splitter = new DefaultMessageSplitter();
@@ -108,4 +119,112 @@ public class DefaultSplitterTests {
 		Message<?> output = replyChannel.receive(15);
 		assertThat(output, is(nullValue()));
 	}
+
+	@Test
+	public void splitStream() {
+		Message<?> message = new GenericMessage<>(
+				Stream.generate(Math::random)
+						.limit(10));
+		QueueChannel outputChannel = new QueueChannel();
+		DefaultMessageSplitter splitter = new DefaultMessageSplitter();
+		splitter.setOutputChannel(outputChannel);
+		splitter.handleMessage(message);
+		for (int i = 0; i < 10; i++) {
+			Message<?> reply = outputChannel.receive(0);
+			assertEquals(message.getHeaders().getId(), new IntegrationMessageHeaderAccessor(reply).getCorrelationId());
+		}
+		assertNull(outputChannel.receive(10));
+	}
+
+	@Test
+	public void splitFlux() {
+		Message<?> message = new GenericMessage<>(
+				Flux
+						.generate(() -> 0,
+								(state, sink) -> {
+									if (state == 10) {
+										sink.complete();
+									}
+									else {
+										sink.next(state);
+									}
+									return ++state;
+								}));
+		QueueChannel outputChannel = new QueueChannel();
+		DefaultMessageSplitter splitter = new DefaultMessageSplitter();
+		splitter.setOutputChannel(outputChannel);
+		splitter.handleMessage(message);
+		for (int i = 0; i < 10; i++) {
+			Message<?> reply = outputChannel.receive(0);
+			assertEquals(message.getHeaders().getId(), new IntegrationMessageHeaderAccessor(reply).getCorrelationId());
+		}
+		assertNull(outputChannel.receive(10));
+	}
+
+	@Test
+	public void splitArrayPayloadReactive() {
+		Message<?> message = new GenericMessage<>(new String[] { "x", "y", "z" });
+		FluxMessageChannel replyChannel = new FluxMessageChannel();
+		DefaultMessageSplitter splitter = new DefaultMessageSplitter();
+		splitter.setOutputChannel(replyChannel);
+
+		splitter.handleMessage(message);
+
+		Flux<String> testFlux =
+				Flux.from(replyChannel)
+						.map(Message::getPayload)
+						.cast(String.class);
+
+		StepVerifier.create(testFlux)
+				.expectNext("x", "y", "z")
+				.then(() ->
+						((Subscriber<?>) TestUtils.getPropertyValue(replyChannel, "subscribers", List.class).get(0))
+								.onComplete())
+				.verifyComplete();
+	}
+
+	@Test
+	public void splitStreamReactive() {
+		Message<?> message = new GenericMessage<>(Stream.of("x", "y", "z"));
+		FluxMessageChannel replyChannel = new FluxMessageChannel();
+		DefaultMessageSplitter splitter = new DefaultMessageSplitter();
+		splitter.setOutputChannel(replyChannel);
+
+		splitter.handleMessage(message);
+
+		Flux<String> testFlux =
+				Flux.from(replyChannel)
+						.map(Message::getPayload)
+						.cast(String.class);
+
+		StepVerifier.create(testFlux)
+				.expectNext("x", "y", "z")
+				.then(() ->
+						((Subscriber<?>) TestUtils.getPropertyValue(replyChannel, "subscribers", List.class).get(0))
+								.onComplete())
+				.verifyComplete();
+	}
+
+	@Test
+	public void splitFluxReactive() {
+		Message<?> message = new GenericMessage<>(Flux.just("x", "y", "z"));
+		FluxMessageChannel replyChannel = new FluxMessageChannel();
+		DefaultMessageSplitter splitter = new DefaultMessageSplitter();
+		splitter.setOutputChannel(replyChannel);
+
+		splitter.handleMessage(message);
+
+		Flux<String> testFlux =
+				Flux.from(replyChannel)
+						.map(Message::getPayload)
+						.cast(String.class);
+
+		StepVerifier.create(testFlux)
+				.expectNext("x", "y", "z")
+				.then(() ->
+						((Subscriber<?>) TestUtils.getPropertyValue(replyChannel, "subscribers", List.class).get(0))
+								.onComplete())
+				.verifyComplete();
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-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.
@@ -23,6 +23,9 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
+import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeader;
+import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeaderKey;
+import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload;
 
 import java.io.ByteArrayInputStream;
 import java.io.File;
@@ -34,11 +37,13 @@ import java.io.InputStream;
 import java.io.Reader;
 import java.nio.charset.Charset;
 import java.util.Date;
+import java.util.List;
 
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.reactivestreams.Subscriber;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -46,12 +51,14 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.ImportResource;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.annotation.Splitter;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.splitter.FileSplitter.FileMarker;
 import org.springframework.integration.support.json.JsonObjectMapper;
 import org.springframework.integration.support.json.JsonObjectMapperProvider;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -63,9 +70,12 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.test.context.support.AnnotationConfigContextLoader;
 import org.springframework.util.FileCopyUtils;
 
+import reactor.test.StepVerifier;
+
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 4.1.2
  */
 @ContextConfiguration(loader = AnnotationConfigContextLoader.class)
@@ -75,7 +85,7 @@ public class FileSplitterTests {
 
 	private static File file;
 
-	static final String SAMPLE_CONTENT = "HelloWorld\näöüß";
+	private static final String SAMPLE_CONTENT = "HelloWorld\näöüß";
 
 	@Autowired
 	private MessageChannel input1;
@@ -264,6 +274,39 @@ public class FileSplitterTests {
 		assertEquals(2, fileMarker.getLineCount());
 	}
 
+	@Test
+	public void testFileSplitterReactive() {
+		FluxMessageChannel outputChannel = new FluxMessageChannel();
+		FileSplitter splitter = new FileSplitter(true, true);
+		splitter.setApplySequence(true);
+		splitter.setOutputChannel(outputChannel);
+		splitter.handleMessage(new GenericMessage<>(file));
+
+
+		StepVerifier.create(outputChannel)
+				.assertNext(m -> {
+					assertThat(m, hasHeaderKey(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE));
+					assertThat(m, hasHeader(FileHeaders.MARKER, "START"));
+					assertThat(m, hasPayload(instanceOf(FileSplitter.FileMarker.class)));
+					FileMarker fileMarker = (FileSplitter.FileMarker) m.getPayload();
+					assertEquals(FileMarker.Mark.START, fileMarker.getMark());
+					assertEquals(file.getAbsolutePath(), fileMarker.getFilePath());
+				})
+				.expectNextCount(2)
+				.assertNext(m -> {
+					assertThat(m, hasHeader(FileHeaders.MARKER, "END"));
+					assertThat(m, hasPayload(instanceOf(FileSplitter.FileMarker.class)));
+					FileMarker fileMarker = (FileSplitter.FileMarker) m.getPayload();
+					assertEquals(FileMarker.Mark.END, fileMarker.getMark());
+					assertEquals(file.getAbsolutePath(), fileMarker.getFilePath());
+					assertEquals(2, fileMarker.getLineCount());
+				})
+				.then(() ->
+						((Subscriber<?>) TestUtils.getPropertyValue(outputChannel, "subscribers", List.class).get(0))
+								.onComplete())
+				.verifyComplete();
+	}
+
 	@Configuration
 	@EnableIntegration
 	@ImportResource("classpath:org/springframework/integration/file/splitter/FileSplitterTests-context.xml")

spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.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.
@@ -23,6 +23,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.function.Function;
 
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -187,6 +188,21 @@ public class XPathMessageSplitter extends AbstractMessageSplitter {
 		}
 	}
 
+	@Override
+	protected int obtainSizeIfPossible(Iterator<?> iterator) {
+		Iterator<?> theIterator = iterator;
+
+		if (iterator instanceof TransformFunctionIterator) {
+			theIterator = ((TransformFunctionIterator) iterator).delegate;
+		}
+
+		if (theIterator instanceof NodeListIterator) {
+			return ((NodeListIterator) theIterator).nodeList.getLength();
+		}
+
+		return 0;
+	}
+
 	@SuppressWarnings("unchecked")
 	private Object splitDocument(Document document) throws Exception {
 		Object nodes = splitNode(document);
@@ -209,16 +225,17 @@ public class XPathMessageSplitter extends AbstractMessageSplitter {
 			return splitStrings;
 		}
 		else {
-			return new FunctionIterator<>((Iterator<Node>) nodes, node -> {
-				StringResult result = new StringResult();
-				try {
-					transformer.transform(new DOMSource(node), result);
-				}
-				catch (TransformerException e) {
-					throw new IllegalStateException("failed to create DocumentBuilder", e);
-				}
-				return result.toString();
-			});
+			return new TransformFunctionIterator((Iterator<Node>) nodes,
+					node -> {
+						StringResult result = new StringResult();
+						try {
+							transformer.transform(new DOMSource(node), result);
+						}
+						catch (TransformerException e) {
+							throw new IllegalStateException("failed to create DocumentBuilder", e);
+						}
+						return result.toString();
+					});
 		}
 	}
 
@@ -243,7 +260,7 @@ public class XPathMessageSplitter extends AbstractMessageSplitter {
 
 	private List<Node> convertNodesToDocuments(List<Node> nodes) throws ParserConfigurationException {
 		DocumentBuilder documentBuilder = getNewDocumentBuilder();
-		List<Node> documents = new ArrayList<Node>(nodes.size());
+		List<Node> documents = new ArrayList<>(nodes.size());
 		for (Node node : nodes) {
 			Document document = convertNodeToDocument(documentBuilder, node);
 			documents.add(document);
@@ -306,4 +323,16 @@ public class XPathMessageSplitter extends AbstractMessageSplitter {
 
 	}
 
+	private static final class TransformFunctionIterator extends FunctionIterator<Node, String> {
+
+		private final Iterator<Node> delegate;
+
+		TransformFunctionIterator(Iterator<Node> delegate,
+				Function<? super Node, ? extends String> function) {
+			super(delegate, function);
+			this.delegate = delegate;
+		}
+
+	}
+
 }

spring-integration-xml/src/test/java/org/springframework/integration/xml/splitter/XPathMessageSplitterTests.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,9 +16,11 @@
 
 package org.springframework.integration.xml.splitter;
 
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertThat;
 
 import java.util.List;
 
@@ -27,6 +29,7 @@ import org.junit.Test;
 import org.w3c.dom.Document;
 import org.w3c.dom.Node;
 
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.handler.ReplyRequiredException;
 import org.springframework.integration.xml.util.XmlTestUtil;
@@ -36,6 +39,7 @@ import org.springframework.messaging.support.GenericMessage;
 
 /**
  * @author Jonas Partner
+ * @author Artem Bilan
  */
 public class XPathMessageSplitterTests {
 
@@ -47,58 +51,61 @@ public class XPathMessageSplitterTests {
 	@Before
 	public void setUp() {
 		String splittingXPath = "/orders/order";
-		splitter = new XPathMessageSplitter(splittingXPath);
-		splitter.setOutputChannel(replyChannel);
-		splitter.setRequiresReply(true);
+		this.splitter = new XPathMessageSplitter(splittingXPath);
+		this.splitter.setOutputChannel(replyChannel);
+		this.splitter.setRequiresReply(true);
 	}
 
 
 	@Test
 	public void splitDocument() throws Exception {
 		Document doc = XmlTestUtil.getDocumentForString("<orders><order>one</order><order>two</order><order>three</order></orders>");
-		splitter.handleMessage(new GenericMessage<Document>(doc));
+		this.splitter.handleMessage(new GenericMessage<>(doc));
 		List<Message<?>> docMessages = this.replyChannel.clear();
 		assertEquals("Wrong number of messages", 3, docMessages.size());
 		for (Message<?> message : docMessages) {
-			assertTrue("unexpected payload type" + message.getPayload().getClass().getName(), message.getPayload() instanceof Node);
-			assertFalse("unexpected payload type" + message.getPayload().getClass().getName(), message.getPayload() instanceof Document);
+			assertThat(message.getPayload(), instanceOf(Node.class));
+			assertThat(message.getPayload(), not(instanceOf(Document.class)));
+			assertThat(new IntegrationMessageHeaderAccessor(message).getSequenceSize(), greaterThan(0));
 		}
 	}
 
 	@Test(expected = ReplyRequiredException.class)
 	public void splitDocumentThatDoesNotMatch() throws Exception {
 		Document doc = XmlTestUtil.getDocumentForString("<wrongDocument/>");
-		splitter.handleMessage(new GenericMessage<Document>(doc));
+		this.splitter.handleMessage(new GenericMessage<>(doc));
 	}
 
 	@Test
 	public void splitDocumentWithCreateDocumentsTrue() throws Exception {
-		splitter.setCreateDocuments(true);
+		this.splitter.setCreateDocuments(true);
 		Document doc = XmlTestUtil.getDocumentForString("<orders><order>one</order><order>two</order><order>three</order></orders>");
-		splitter.handleMessage(new GenericMessage<Document>(doc));
+		this.splitter.handleMessage(new GenericMessage<>(doc));
 		List<Message<?>> docMessages = this.replyChannel.clear();
 		assertEquals("Wrong number of messages", 3, docMessages.size());
 		for (Message<?> message : docMessages) {
-			assertTrue("unexpected payload type" + message.getPayload().getClass().getName(), message.getPayload() instanceof Document);
+			assertThat(message.getPayload(), instanceOf(Document.class));
 			Document docPayload = (Document) message.getPayload();
 			assertEquals("Wrong root element name", "order", docPayload.getDocumentElement().getLocalName());
+			assertThat(new IntegrationMessageHeaderAccessor(message).getSequenceSize(), greaterThan(0));
 		}
 	}
 
 	@Test
 	public void splitStringXml() throws Exception {
 		String payload = "<orders><order>one</order><order>two</order><order>three</order></orders>";
-		splitter.handleMessage(new GenericMessage<String>(payload));
+		this.splitter.handleMessage(new GenericMessage<>(payload));
 		List<Message<?>> docMessages = this.replyChannel.clear();
 		assertEquals("Wrong number of messages", 3, docMessages.size());
 		for (Message<?> message : docMessages) {
-			assertTrue("unexpected payload type " + message.getPayload().getClass().getName(), message.getPayload() instanceof String);
+			assertThat(message.getPayload(), instanceOf(String.class));
+			assertThat(new IntegrationMessageHeaderAccessor(message).getSequenceSize(), greaterThan(0));
 		}
 	}
 
 	@Test(expected = MessageHandlingException.class)
 	public void invalidPayloadType() {
-		splitter.handleMessage(new GenericMessage<Integer>(123));
+		this.splitter.handleMessage(new GenericMessage<>(123));
 	}
 
 }

src/reference/asciidoc/splitter.adoc

@@ -43,17 +43,27 @@ The input argument might either be a `Message` or a simple POJO.
 In the latter case, the splitter will receive the payload of the incoming message.
 Since this decouples the code from the Spring Integration API and will typically be easier to test, it is the recommended approach.
 
-*Splitter and Iterators*
+*Iterators*
 
 Starting with _version 4.1_, the `AbstractMessageSplitter` supports the `Iterator` type for the `value` to split.
 Note, in the case of an `Iterator` (or `Iterable`), we don't have access to the number of underlying items and the `SEQUENCE_SIZE` header is set to `0`.
 This means that the default `SequenceSizeReleaseStrategy` of an `<aggregator>` won't work and the group for the `CORRELATION_ID` from the `splitter` won't be released; it will remain as `incomplete`.
 In this case you should use an appropriate custom `ReleaseStrategy` or rely on `send-partial-result-on-expiry` together with `group-timeout` or a `MessageGroupStoreReaper`.
 
+Starting with _version 5.0_, the `AbstractMessageSplitter` provides `protected obtainSizeIfPossible()` methods to allow the determination of the size of the `Iterable` and `Iterator` objects if that is possible.
+For example `XPathMessageSplitter` can determine the size of the underlying `NodeList` object.
+
 An `Iterator` object is useful to avoid the need for building an entire collection in the memory before splitting.
 For example, when underlying items are populated from some external system (e.g.
 DataBase or FTP `MGET`) using iterations or streams.
 
+*Stream and Flux*
+
+Starting with _version 5.0_, the `AbstractMessageSplitter` supports the Java `Stream` and Reactive Streams `Publisher` types for the `value` to split.
+In this case the target `Iterator` is built on their iteration functionality.
+
+In addition, if 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.
+
 [[splitter-config]]
 ==== Configuring Splitter
 

src/reference/asciidoc/whats-new.adoc

@@ -72,13 +72,6 @@ See <<transaction-synchronization>> for more information.
 The aggregator expression-based `ReleaseStrategy` now evaluates the expression against the `MesageGroup` instead of just the collection of `Message<?>`.
 See <<aggregator-spel>> for more information.
 
-==== JMS Changes
-
-Previously, Spring Integration JMS XML configuration used a default bean name `connectionFactory` for the JMS Connection Factory, allowing the property to be omitted from component definitions.
-It has now been renamed to `jmsConnectionFactory`, which is the bean name used by Spring Boot to auto-configure the JMS Connection Factory bean.
-
-If your application is relying on the previous behavior, rename your `connectionFactory` bean to `jmsConnectionFactory`, or specifically configure your components to use your bean using its current name.
-
 ==== Gateway Changes
 
 The gateway now correctly sets the `errorChannel` header when the gateway method has a `void` return type and an error channel is provided.
@@ -89,6 +82,28 @@ The `RequestReplyExchanger` interface now has a `throws MessagingException` clau
 
 See <<gateway-error-handling>> for more information.
 
+==== Aggregator Performance Changes
+
+Aggregators now use a `SimpleSequenceSizeReleaseStrategy` by default, which is more efficient, especially with large groups.
+Empty groups are now scheduled for removal after `empty-group-min-timeout`.
+See <<aggregator>> for more information.
+
+==== Splitter Changes
+
+The Splitter component now can handle and split Java `Stream` and Reactive Streams `Publisher` objects.
+If the output channel is a `ReactiveStreamsSubscribableChannel`, the `AbstractMessageSplitter` builds a `Flux` for subsequent iteration instead of a regular `Iterator` independent of object being split.
+In addition, `AbstractMessageSplitter` provides `protected obtainSizeIfPossible()` methods to allow the determination of the size of the `Iterable` and `Iterator` objects if that is possible.
+See <<splitter>> for more information.
+
+==== JMS Changes
+
+Previously, Spring Integration JMS XML configuration used a default bean name `connectionFactory` for the JMS Connection Factory, allowing the property to be omitted from component definitions.
+It has now been renamed to `jmsConnectionFactory`, which is the bean name used by Spring Boot to auto-configure the JMS Connection Factory bean.
+
+If your application is relying on the previous behavior, rename your `connectionFactory` bean to `jmsConnectionFactory`, or specifically configure your components to use your bean using its current name.
+
+See <<jms>> for more information.
+
 ==== Mail Changes
 
 Some inconsistencies with rendering IMAP mail content have been resolved.
@@ -180,12 +195,6 @@ See <<content-type-conversion-outbound>> for more information.
 The `DefaultHttpHeaderMapper.userDefinedHeaderPrefix` property is now an empty string by default instead of `X-`.
 See <<http-header-mapping>> for more information.
 
-==== Aggregator Performance Changes
-
-Aggregators now use a `SimpleSequenceSizeReleaseStrategy` by default, which is more efficient, especially with large groups.
-Empty groups are now scheduled for removal after `empty-group-min-timeout`.
-See <<aggregator>> for more information.
-
 ==== MQTT Changes
 
 Inbound messages are now mapped with headers `RECEIVED_TOPIC`, `RECEIVED_QOS` and `RECEIVED_RETAINED` to avoid inadvertent propagation to outbound messages when an application is relaying messages.