From f112ecbb7e2233b37b3136a2054bf3a496087eb1 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 25 May 2017 14:21:40 -0400 Subject: [PATCH] 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 --- .../splitter/AbstractMessageSplitter.java | 124 ++++++++++++++--- .../integration/util/FunctionIterator.java | 4 +- .../splitter/DefaultSplitterTests.java | 127 +++++++++++++++++- .../file/splitter/FileSplitterTests.java | 47 ++++++- .../xml/splitter/XPathMessageSplitter.java | 53 ++++++-- .../splitter/XPathMessageSplitterTests.java | 39 +++--- src/reference/asciidoc/splitter.adoc | 12 +- src/reference/asciidoc/whats-new.adoc | 35 +++-- 8 files changed, 371 insertions(+), 70 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java index c27e0c0ac1..3a929c5b0b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java +++ b/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 iterator; + boolean reactive = getOutputChannel() instanceof ReactiveStreamsSubscribableChannel; + setAsync(reactive); + + Iterator iterator = null; + Flux flux = null; + final int sequenceSize; - if (result instanceof Collection) { - Collection items = (Collection) result; - sequenceSize = items.size(); - iterator = items.iterator(); + + if (result instanceof Iterable) { + Iterable iterable = (Iterable) 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) result).iterator(); + if (reactive) { + flux = Flux.fromArray(items); + } + else { + iterator = Arrays.asList(items).iterator(); + } } else if (result instanceof Iterator) { + Iterator iter = (Iterator) result; + sequenceSize = obtainSizeIfPossible(iter); + if (reactive) { + flux = Flux.fromIterable(() -> iter); + } + else { + iterator = iter; + } + } + else if (result instanceof Stream) { + Stream stream = ((Stream) result); sequenceSize = 0; - iterator = (Iterator) result; + if (reactive) { + flux = Flux.fromStream(stream); + } + else { + iterator = stream.iterator(); + } + } + else if (result instanceof Publisher) { + Publisher publisher = (Publisher) result; + sequenceSize = 0; + if (reactive) { + flux = Flux.from(publisher); + } + else { + iterator = Flux.from((Publisher) 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 headers = messageHeaders; final Object correlationId = message.getHeaders().getId(); final AtomicInteger sequenceNumber = new AtomicInteger(1); - return new FunctionIterator>(iterator, - object -> - createBuilder(object, headers, correlationId, sequenceNumber.getAndIncrement(), sequenceSize)); + Function> 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 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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java b/spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java index 1123f276f2..021398a5e3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/FunctionIterator.java +++ b/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 implements Iterator { +public class FunctionIterator implements Iterator { private final Iterator iterator; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java index 210500871a..411e27767f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java +++ b/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 payload = Arrays.asList(new String[] { "x", "y", "z" }); + List payload = Arrays.asList("x", "y", "z"); Message> 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 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 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 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(); + } + } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java index 4c06f61cc4..0a436d04da 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java +++ b/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") diff --git a/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java b/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java index aa1e79256f..4a6f06f6aa 100644 --- a/spring-integration-xml/src/main/java/org/springframework/integration/xml/splitter/XPathMessageSplitter.java +++ b/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) 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) 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 convertNodesToDocuments(List nodes) throws ParserConfigurationException { DocumentBuilder documentBuilder = getNewDocumentBuilder(); - List documents = new ArrayList(nodes.size()); + List 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 { + + private final Iterator delegate; + + TransformFunctionIterator(Iterator delegate, + Function function) { + super(delegate, function); + this.delegate = delegate; + } + + } + } diff --git a/spring-integration-xml/src/test/java/org/springframework/integration/xml/splitter/XPathMessageSplitterTests.java b/spring-integration-xml/src/test/java/org/springframework/integration/xml/splitter/XPathMessageSplitterTests.java index d0c235a7ea..023df34e03 100644 --- a/spring-integration-xml/src/test/java/org/springframework/integration/xml/splitter/XPathMessageSplitterTests.java +++ b/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("onetwothree"); - splitter.handleMessage(new GenericMessage(doc)); + this.splitter.handleMessage(new GenericMessage<>(doc)); List> 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(""); - splitter.handleMessage(new GenericMessage(doc)); + this.splitter.handleMessage(new GenericMessage<>(doc)); } @Test public void splitDocumentWithCreateDocumentsTrue() throws Exception { - splitter.setCreateDocuments(true); + this.splitter.setCreateDocuments(true); Document doc = XmlTestUtil.getDocumentForString("onetwothree"); - splitter.handleMessage(new GenericMessage(doc)); + this.splitter.handleMessage(new GenericMessage<>(doc)); List> 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 = "onetwothree"; - splitter.handleMessage(new GenericMessage(payload)); + this.splitter.handleMessage(new GenericMessage<>(payload)); List> 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(123)); + this.splitter.handleMessage(new GenericMessage<>(123)); } } diff --git a/src/reference/asciidoc/splitter.adoc b/src/reference/asciidoc/splitter.adoc index 68e2362782..24c010f3d6 100644 --- a/src/reference/asciidoc/splitter.adoc +++ b/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 `` 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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 917d6c0527..5af6167ece 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -72,13 +72,6 @@ See <> for more information. The aggregator expression-based `ReleaseStrategy` now evaluates the expression against the `MesageGroup` instead of just the collection of `Message`. See <> 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 <> 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 <> 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 <> 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 <> for more information. + ==== Mail Changes Some inconsistencies with rendering IMAP mail content have been resolved. @@ -180,12 +195,6 @@ See <> for more information. The `DefaultHttpHeaderMapper.userDefinedHeaderPrefix` property is now an empty string by default instead of `X-`. See <> 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 <> 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.