From 40a9d28667befdddb613e989ca0f0f2c4efc9e6b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 6 Apr 2016 19:44:45 -0400 Subject: [PATCH] INT-3979: Properly Handle `MBuilder` in Splitter JIRA: https://jira.spring.io/browse/INT-3979 Previously we had a to instantiate all messages from the `MessageBuilder` if we would like to return messages as splitted items. For example to populate some item-specific headers. From other hand the `MessageBuilder` as an item incorrectly remained as a `payload`. * Add condition logic to the `AbstractMessageSplitter` to properly handle `AbstractIntegrationMessageBuilder` and don't create an extra `Message` * Apply the same logic to the `AbstractAggregatingMessageGroupProcessor` (We may consider to backport it later) --- ...tractAggregatingMessageGroupProcessor.java | 12 ++++-- .../splitter/AbstractMessageSplitter.java | 11 ++++-- .../splitter/MethodInvokingSplitterTests.java | 38 ++++++++++++++++++- 3 files changed, 53 insertions(+), 8 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index 01deeb34e2..539cccfe60 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -47,6 +47,7 @@ import org.springframework.util.Assert; * @author Dave Syer * @author Gary Russell * @author Artem Bilan + * * @since 2.0 */ public abstract class AbstractAggregatingMessageGroupProcessor implements MessageGroupProcessor, @@ -83,13 +84,18 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag Object payload = this.aggregatePayloads(group, headers); AbstractIntegrationMessageBuilder builder; if (payload instanceof Message) { - builder = getMessageBuilderFactory().fromMessage((Message) payload).copyHeadersIfAbsent(headers); + builder = getMessageBuilderFactory().fromMessage((Message) payload); + } + else if (payload instanceof AbstractIntegrationMessageBuilder) { + builder = (AbstractIntegrationMessageBuilder) payload; } else { - builder = getMessageBuilderFactory().withPayload(payload).copyHeadersIfAbsent(headers); + builder = getMessageBuilderFactory().withPayload(payload); } - return builder.popSequenceDetails().build(); + return builder.copyHeadersIfAbsent(headers) + .popSequenceDetails() + .build(); } /** 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 2d8dc1f302..33bc48de5c 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-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -112,12 +112,15 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess Object correlationId, int sequenceNumber, int sequenceSize) { AbstractIntegrationMessageBuilder builder; if (item instanceof Message) { - builder = this.getMessageBuilderFactory().fromMessage((Message) item); + builder = getMessageBuilderFactory().fromMessage((Message) item); + } + else if (item instanceof AbstractIntegrationMessageBuilder) { + builder = (AbstractIntegrationMessageBuilder) item; } else { - builder = this.getMessageBuilderFactory().withPayload(item); - builder.copyHeaders(headers); + builder = getMessageBuilderFactory().withPayload(item); } + builder.copyHeadersIfAbsent(headers); if (this.applySequence) { builder.pushSequenceDetails(correlationId, sequenceNumber, sequenceSize); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/splitter/MethodInvokingSplitterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/splitter/MethodInvokingSplitterTests.java index 4333b18142..75321ac6d5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/splitter/MethodInvokingSplitterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/splitter/MethodInvokingSplitterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 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. @@ -30,6 +30,7 @@ import org.junit.Test; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.annotation.Splitter; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.handler.annotation.Header; @@ -37,6 +38,7 @@ import org.springframework.messaging.support.GenericMessage; /** * @author Mark Fisher + * @author Artem Bilan */ public class MethodInvokingSplitterTests { @@ -138,6 +140,29 @@ public class MethodInvokingSplitterTests { assertEquals("bar", reply2.getPayload()); } + @Test + public void splitMessageToMessageBuilderList() throws Exception { + Message message = MessageBuilder.withPayload("foo.bar") + .setHeader("myHeader", "myValue") + .build(); + MethodInvokingSplitter splitter = new MethodInvokingSplitter(testBean, "messageToMessageBuilderList"); + QueueChannel replyChannel = new QueueChannel(); + splitter.setOutputChannel(replyChannel); + splitter.handleMessage(message); + List> replies = replyChannel.clear(); + Message reply1 = replies.get(0); + assertNotNull(reply1); + assertEquals("foo", reply1.getPayload()); + assertEquals("myValue", reply1.getHeaders().get("myHeader")); + assertEquals("bar", reply1.getHeaders().get("foo")); + + Message reply2 = replies.get(1); + assertNotNull(reply2); + assertEquals("bar", reply2.getPayload()); + assertEquals("myValue", reply2.getHeaders().get("myHeader")); + assertEquals("bar", reply2.getHeaders().get("foo")); + } + @Test public void splitStringToMessageArray() throws Exception { GenericMessage message = new GenericMessage("foo.bar"); @@ -486,6 +511,17 @@ public class MethodInvokingSplitterTests { return messages; } + public List messageToMessageBuilderList(Message input) { + String[] strings = input.getPayload().toString().split("\\."); + List messageBuilders = new ArrayList<>(); + for (String s : strings) { + MessageBuilder builder = MessageBuilder.withPayload(s) + .setHeader("foo", "bar"); + messageBuilders.add(builder); + } + return messageBuilders; + } + public Message[] stringToMessageArray(String input) { String[] strings = input.split("\\."); Message[] messages = new TestStringMessage[strings.length];