Skip to content

Commit 716494c

Browse files
artembilangaryrussell
authored andcommitted
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**
1 parent c4d6cf2 commit 716494c

File tree

4 files changed

+49
-10
lines changed

4 files changed

+49
-10
lines changed

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

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -117,6 +117,17 @@ public static ObjectToJsonTransformer toJson(JsonObjectMapper<?, ?> jsonObjectMa
117117
return toJson(jsonObjectMapper, null, contentType);
118118
}
119119

120+
/**
121+
* Factory for the {@link ObjectToJsonTransformer} based on the provided {@link ObjectToJsonTransformer.ResultType}.
122+
* @param resultType the {@link ObjectToJsonTransformer.ResultType} to use.
123+
* Defaults to {@link ObjectToJsonTransformer.ResultType#STRING}.
124+
* @return the ObjectToJsonTransformer
125+
* @since 5.0.9
126+
*/
127+
public static ObjectToJsonTransformer toJson(ObjectToJsonTransformer.ResultType resultType) {
128+
return toJson(null, resultType, null);
129+
}
130+
120131
public static ObjectToJsonTransformer toJson(ObjectToJsonTransformer.ResultType resultType, String contentType) {
121132
return toJson(null, resultType, contentType);
122133
}

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

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,9 +31,11 @@
3131
import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
3232
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
3333
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
34+
import org.springframework.integration.support.json.JacksonPresent;
3435
import org.springframework.integration.util.FunctionIterator;
3536
import org.springframework.messaging.Message;
3637

38+
import com.fasterxml.jackson.core.TreeNode;
3739
import reactor.core.publisher.Flux;
3840

3941
/**
@@ -163,12 +165,21 @@ else if (result instanceof Publisher<?>) {
163165
/**
164166
* Obtain a size of the provided {@link Iterable}. Default implementation returns
165167
* {@link Collection#size()} if the iterable is a collection, or {@code 0} otherwise.
168+
* If iterable is a Jackson {@code TreeNode}, then its size is used.
166169
* @param iterable the {@link Iterable} to obtain the size
167170
* @return the size of the {@link Iterable}
168171
* @since 5.0
169172
*/
170173
protected int obtainSizeIfPossible(Iterable<?> iterable) {
171-
return iterable instanceof Collection ? ((Collection<?>) iterable).size() : 0;
174+
if (iterable instanceof Collection) {
175+
return ((Collection<?>) iterable).size();
176+
}
177+
else if (JacksonPresent.isJackson2Present() && JacksonNodeHelper.isNode(iterable)) {
178+
return JacksonNodeHelper.nodeSize(iterable);
179+
}
180+
else {
181+
return 0;
182+
}
172183
}
173184

174185
/**
@@ -266,4 +277,18 @@ public String getComponentType() {
266277
*/
267278
protected abstract Object splitMessage(Message<?> message);
268279

280+
281+
private static class JacksonNodeHelper {
282+
283+
private static boolean isNode(Object object) {
284+
return object instanceof TreeNode;
285+
}
286+
287+
@SuppressWarnings("unchecked")
288+
private static int nodeSize(Object node) {
289+
return ((TreeNode) node).size();
290+
}
291+
292+
}
293+
269294
}

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

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,6 @@
3636
import org.springframework.beans.factory.annotation.Qualifier;
3737
import org.springframework.context.annotation.Bean;
3838
import org.springframework.context.annotation.Configuration;
39-
import org.springframework.context.annotation.DependsOn;
4039
import org.springframework.integration.IntegrationMessageHeaderAccessor;
4140
import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
4241
import org.springframework.integration.channel.QueueChannel;
@@ -45,7 +44,9 @@
4544
import org.springframework.integration.dsl.IntegrationFlows;
4645
import org.springframework.integration.dsl.MessageChannelSpec;
4746
import org.springframework.integration.dsl.MessageChannels;
47+
import org.springframework.integration.dsl.Transformers;
4848
import org.springframework.integration.handler.MessageTriggerAction;
49+
import org.springframework.integration.json.ObjectToJsonTransformer;
4950
import org.springframework.integration.support.MessageBuilder;
5051
import org.springframework.messaging.Message;
5152
import org.springframework.messaging.MessageChannel;
@@ -55,6 +56,8 @@
5556
import org.springframework.test.annotation.DirtiesContext;
5657
import org.springframework.test.context.junit4.SpringRunner;
5758

59+
import com.fasterxml.jackson.databind.node.TextNode;
60+
5861
/**
5962
* @author Artem Bilan
6063
* @author Gary Russell
@@ -73,7 +76,6 @@ public class CorrelationHandlerTests {
7376

7477

7578
@Autowired
76-
@Qualifier("splitAggregateInput")
7779
private MessageChannel splitAggregateInput;
7880

7981
@Autowired
@@ -114,7 +116,7 @@ public void testSplitterResequencer() {
114116

115117
@Test
116118
public void testSplitterAggregator() {
117-
List<Character> payload = Arrays.asList('a', 'b', 'c', 'd', 'e');
119+
List<String> payload = Arrays.asList("a", "b", "c", "d", "e");
118120

119121
QueueChannel replyChannel = new QueueChannel();
120122
this.splitAggregateInput.send(MessageBuilder.withPayload(payload)
@@ -127,7 +129,8 @@ public void testSplitterAggregator() {
127129
@SuppressWarnings("unchecked")
128130
List<Object> result = (List<Object>) receive.getPayload();
129131
for (int i = 0; i < payload.size(); i++) {
130-
assertEquals(payload.get(i), result.get(i));
132+
assertThat(result.get(i), instanceOf(TextNode.class));
133+
assertEquals(TextNode.valueOf(payload.get(i)), result.get(i));
131134
}
132135
}
133136

@@ -204,6 +207,7 @@ public IntegrationFlow splitResequenceFlow(MessageChannel executorChannel) {
204207
@Bean
205208
public IntegrationFlow splitAggregateFlow() {
206209
return IntegrationFlows.from("splitAggregateInput", true)
210+
.transform(Transformers.toJson(ObjectToJsonTransformer.ResultType.NODE))
207211
.split()
208212
.channel(MessageChannels.executor(taskExecutor()))
209213
.resequence()
@@ -260,7 +264,6 @@ public IntegrationFlow barrierFlow() {
260264
}
261265

262266
@Bean
263-
@DependsOn("barrierFlow")
264267
public IntegrationFlow releaseBarrierFlow(MessageTriggerAction barrierTriggerAction) {
265268
return IntegrationFlows.from(releaseChannel())
266269
.trigger(barrierTriggerAction,

src/reference/asciidoc/splitter.adoc

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -45,17 +45,17 @@ We recommend this approach, because it decouples the code from the Spring Integr
4545

4646
===== Iterators
4747

48-
Starting with _version 4.1_, the `AbstractMessageSplitter` supports the `Iterator` type for the `value` to split.
48+
Starting with version 4.1, the `AbstractMessageSplitter` supports the `Iterator` type for the `value` to split.
4949
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`.
5050
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`.
5151
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`.
5252

53-
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.
53+
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.
5454
For example `XPathMessageSplitter` can determine the size of the underlying `NodeList` object.
55+
And starting with version 5.0.9, this method also properly returns a size of the `com.fasterxml.jackson.core.TreeNode`.
5556

5657
An `Iterator` object is useful to avoid the need for building an entire collection in the memory before splitting.
57-
For example, when underlying items are populated from some external system (e.g.
58-
DataBase or FTP `MGET`) using iterations or streams.
58+
For example, when underlying items are populated from some external system (e.g. DataBase or FTP `MGET`) using iterations or streams.
5959

6060
===== Stream and Flux
6161

0 commit comments

Comments
 (0)