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)

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

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

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<String> 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<Message<?>> 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<String> message = new GenericMessage<String>("foo.bar");
@@ -486,6 +511,17 @@ public class MethodInvokingSplitterTests {
 			return messages;
 		}
 
+		public List<AbstractIntegrationMessageBuilder> messageToMessageBuilderList(Message<?> input) {
+			String[] strings = input.getPayload().toString().split("\\.");
+			List<AbstractIntegrationMessageBuilder> messageBuilders = new ArrayList<>();
+			for (String s : strings) {
+				MessageBuilder<String> builder = MessageBuilder.withPayload(s)
+						.setHeader("foo", "bar");
+				messageBuilders.add(builder);
+			}
+			return messageBuilders;
+		}
+
 		public Message<String>[] stringToMessageArray(String input) {
 			String[] strings = input.split("\\.");
 			Message<String>[] messages = new TestStringMessage[strings.length];