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**

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);
 	}

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();
+		}
+
+	}
+
 }

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<Character> payload = Arrays.asList('a', 'b', 'c', 'd', 'e');
+		List<String> 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<Object> result = (List<Object>) 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,