From 7faeee849de80cb7f38e6a054f01ebd6eea0f417 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 5 Oct 2018 15:25:27 -0400 Subject: [PATCH] INT-4538: Fix Splitter for ArrayNode.size() JIRA: https://jira.spring.io/browse/INT-4538 * Add support for Jackson `TreeNode.size()` in the `AbstractMessageSplitter` * Add overloaded `Transformers.toJson(ObjectToJsonTransformer.ResultType)` * Modify test to verify Jackson `ArrayNode` use-case with the splitter and aggregator **Cherry-pick to 5.0.x** --- .../integration/dsl/Transformers.java | 11 ++++++++ .../splitter/AbstractMessageSplitter.java | 27 ++++++++++++++++++- .../correlation/CorrelationHandlerTests.java | 13 +++++---- src/reference/asciidoc/splitter.adoc | 8 +++--- 4 files changed, 49 insertions(+), 10 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Transformers.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Transformers.java index f5a538260c..a4037d802e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Transformers.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Transformers.java @@ -110,6 +110,17 @@ public abstract class Transformers { return toJson(jsonObjectMapper, null, contentType); } + /** + * Factory for the {@link ObjectToJsonTransformer} based on the provided {@link ObjectToJsonTransformer.ResultType}. + * @param resultType the {@link ObjectToJsonTransformer.ResultType} to use. + * Defaults to {@link ObjectToJsonTransformer.ResultType#STRING}. + * @return the ObjectToJsonTransformer + * @since 5.0.9 + */ + public static ObjectToJsonTransformer toJson(ObjectToJsonTransformer.ResultType resultType) { + return toJson(null, resultType, null); + } + public static ObjectToJsonTransformer toJson(ObjectToJsonTransformer.ResultType resultType, String contentType) { return toJson(null, resultType, contentType); } 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 3878d0c424..3544316d06 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 @@ -31,9 +31,11 @@ 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.support.json.JacksonPresent; import org.springframework.integration.util.FunctionIterator; import org.springframework.messaging.Message; +import com.fasterxml.jackson.core.TreeNode; import reactor.core.publisher.Flux; /** @@ -163,12 +165,21 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess /** * Obtain a size of the provided {@link Iterable}. Default implementation returns * {@link Collection#size()} if the iterable is a collection, or {@code 0} otherwise. + * If iterable is a Jackson {@code TreeNode}, then its size is used. * @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; + if (iterable instanceof Collection) { + return ((Collection) iterable).size(); + } + else if (JacksonPresent.isJackson2Present() && JacksonNodeHelper.isNode(iterable)) { + return JacksonNodeHelper.nodeSize(iterable); + } + else { + return 0; + } } /** @@ -266,4 +277,18 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess */ protected abstract Object splitMessage(Message message); + + private static class JacksonNodeHelper { + + private static boolean isNode(Object object) { + return object instanceof TreeNode; + } + + @SuppressWarnings("unchecked") + private static int nodeSize(Object node) { + return ((TreeNode) node).size(); + } + + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java index 596b2e90a2..6af6592b5e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java @@ -36,7 +36,6 @@ 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.context.annotation.DependsOn; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy; import org.springframework.integration.channel.QueueChannel; @@ -45,7 +44,9 @@ import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.Transformers; import org.springframework.integration.handler.MessageTriggerAction; +import org.springframework.integration.json.ObjectToJsonTransformer; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -55,6 +56,8 @@ 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.TextNode; + /** * @author Artem Bilan * @author Gary Russell @@ -73,7 +76,6 @@ public class CorrelationHandlerTests { @Autowired - @Qualifier("splitAggregateInput") private MessageChannel splitAggregateInput; @Autowired @@ -114,7 +116,7 @@ public class CorrelationHandlerTests { @Test public void testSplitterAggregator() { - List payload = Arrays.asList('a', 'b', 'c', 'd', 'e'); + List payload = Arrays.asList("a", "b", "c", "d", "e"); QueueChannel replyChannel = new QueueChannel(); this.splitAggregateInput.send(MessageBuilder.withPayload(payload) @@ -127,7 +129,8 @@ public class CorrelationHandlerTests { @SuppressWarnings("unchecked") List result = (List) receive.getPayload(); for (int i = 0; i < payload.size(); i++) { - assertEquals(payload.get(i), result.get(i)); + assertThat(result.get(i), instanceOf(TextNode.class)); + assertEquals(TextNode.valueOf(payload.get(i)), result.get(i)); } } @@ -203,6 +206,7 @@ public class CorrelationHandlerTests { @Bean public IntegrationFlow splitAggregateFlow() { return IntegrationFlows.from("splitAggregateInput", true) + .transform(Transformers.toJson(ObjectToJsonTransformer.ResultType.NODE)) .split() .channel(MessageChannels.executor(taskExecutor())) .resequence() @@ -259,7 +263,6 @@ public class CorrelationHandlerTests { } @Bean - @DependsOn("barrierFlow") public IntegrationFlow releaseBarrierFlow(MessageTriggerAction barrierTriggerAction) { return IntegrationFlows.from(releaseChannel()) .trigger(barrierTriggerAction, diff --git a/src/reference/asciidoc/splitter.adoc b/src/reference/asciidoc/splitter.adoc index a09c0aeddd..a00df2d887 100644 --- a/src/reference/asciidoc/splitter.adoc +++ b/src/reference/asciidoc/splitter.adoc @@ -45,17 +45,17 @@ Since this decouples the code from the Spring Integration API and will typically *Iterators* -Starting with _version 4.1_, the `AbstractMessageSplitter` supports the `Iterator` type for the `value` to split. +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. +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. +And starting with version 5.0.9, this method also properly returns a size of the `com.fasterxml.jackson.core.TreeNode`. 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. +For example, when underlying items are populated from some external system (e.g. DataBase or FTP `MGET`) using iterations or streams. *Stream and Flux*