Added ReplyHolder and modified signature of AbstractReplyProducingMessageConsumer to return void and accept this additional parameter instead.

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java

@@ -31,6 +31,7 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
+import org.springframework.integration.endpoint.ReplyHolder;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.message.MessageHandlingException;
@@ -166,7 +167,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 	}
 
 	@Override
-	protected final Message<?> handle(Message<?> message) {
+	protected final void handle(Message<?> message, ReplyHolder replyHolder) {
 		if (!this.initialized) {
 			this.afterPropertiesSet();
 		}
@@ -181,7 +182,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 						+ correlationId + "' has already completed or timed out.");
 			}
 			this.sendToDiscardChannelIfAvailable(message);
-			return null;
+			return;
 		}
 		MessageBarrier barrier = barriers.putIfAbsent(correlationId, createMessageBarrier());
 		if (barrier == null) {
@@ -189,17 +190,17 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 		}
 		List<Message<?>> releasedMessages = barrier.addAndRelease(message);
 		if (CollectionUtils.isEmpty(releasedMessages)) {
-			return null;
+			return;
 		}
 		if (isBarrierRemovable(correlationId, releasedMessages)) {
 			this.removeBarrier(correlationId);
 		}
 		Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
 		if (ObjectUtils.isEmpty(processedMessages)) {
-			return null;
+			return;
 		}
 		this.afterRelease(correlationId, releasedMessages);
-		return null;
+		return;
 	}
 
 	private void afterRelease(Object correlationId, List<Message<?>> releasedMessages) {

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java

@@ -16,16 +16,12 @@
 
 package org.springframework.integration.endpoint;
 
-import java.util.ArrayList;
-import java.util.List;
-
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.integration.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.channel.ChannelResolver;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.MessageChannelTemplate;
-import org.springframework.integration.message.CompositeMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageBuilder;
 import org.springframework.integration.message.MessageHandlingException;
@@ -101,36 +97,31 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess
 		if (!this.supports(message)) {
 			throw new MessageRejectedException(message, "unsupported message");
 		}
-		Object result = this.handle(message);
-		if (result == null) {
+		ReplyHolder replyHolder = new ReplyHolder();
+		this.handle(message, replyHolder);
+		if (replyHolder.isEmpty()) {
 			if (this.requiresReply) {
 				throw new MessageHandlingException(message, "consumer '" + this
 						+ "' requires a reply, but no reply was received");
 			}
 			return;
 		}
-		Message<?> reply = null;
-		if (result instanceof Message && result.equals(message)) {
-			// we simply pass along an unaltered request Message
-			reply = (Message<?>) result;
+		Object targetChannelValue = replyHolder.getTargetChannel();
+		MessageChannel replyChannel = null;
+		if (targetChannelValue == null) {
+			replyChannel = this.resolveReplyChannel(message);
 		}
-		else {
-			reply = buildReplyMessage(result, message.getHeaders());
+		else if (targetChannelValue instanceof String) {
+			replyChannel = this.channelResolver.resolveChannelName((String) targetChannelValue);
 		}
-		MessageChannel replyChannel = this.resolveReplyChannel(message);
-		if (reply instanceof CompositeMessage && this.shouldSplitComposite()) {
-			boolean sentAtLeastOne = false;
-			for (Message<?> nextReply : (CompositeMessage) reply) {
-				boolean sent = this.sendReplyMessage(nextReply, replyChannel);
-				sentAtLeastOne = (sentAtLeastOne || sent);
-			}
-		}
-		else {
-			this.sendReplyMessage(reply, replyChannel);
+		MessageHeaders requestHeaders = message.getHeaders();
+		for (MessageBuilder<?> builder : replyHolder.builders()) {
+			builder.copyHeadersIfAbsent(requestHeaders);
+			this.sendReplyMessage(builder.build(), replyChannel);
 		}
 	}
 
-	protected abstract Object handle(Message<?> message);
+	protected abstract void handle(Message<?> message, ReplyHolder replyHolder);
 
 	protected boolean supports(Message<?> message) {
 		if (this.selector != null && !this.selector.accept(message)) {
@@ -142,38 +133,10 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess
 		return true;
 	}
 
-	protected boolean shouldSplitComposite() {
-		return false;
-	}
-
 	protected boolean sendReplyMessage(Message<?> replyMessage, MessageChannel replyChannel) {
 		return this.channelTemplate.send(replyMessage, replyChannel);
 	}
 
-	private Message<?> buildReplyMessage(Object result, MessageHeaders requestHeaders) {
-		MessageBuilder<?> builder = null;
-		if (result instanceof MessageBuilder) {
-			builder = (MessageBuilder<?>) result;
-		}
-		else if (result instanceof CompositeMessage) {
-			List<Message<?>> messages = ((CompositeMessage) result).getPayload();
-			List<Message<?>> replies = new ArrayList<Message<?>>();
-			for (Message<?> message : messages) {
-				replies.add(this.buildReplyMessage(message, requestHeaders));
-			}
-			return new CompositeMessage(replies);
-		}
-		else if (result instanceof Message<?>) {
-			builder = MessageBuilder.fromMessage((Message<?>) result);
-		}
-		else {
-			builder = MessageBuilder.withPayload(result);
-		}
-		return builder.copyHeadersIfAbsent(requestHeaders)
-			.setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, requestHeaders.getId())
-			.build();
-	}
-
 	private MessageChannel resolveReplyChannel(Message<?> requestMessage) {
 		MessageChannel replyChannel = this.getOutputChannel();
 		if (replyChannel == null) {

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ReplyHolder.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2002-2008 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.endpoint;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageBuilder;
+
+/**
+ * @author Mark Fisher
+ */
+public class ReplyHolder {
+
+	private final List<MessageBuilder<?>> builders = new ArrayList<MessageBuilder<?>>();
+
+	private volatile Object targetChannel;
+
+
+	public MessageBuilder<?> set(Object replyObject) {
+		return this.createAndAddBuilder(replyObject, true);
+	}
+
+	public MessageBuilder<?> add(Object replyObject) {
+		return this.createAndAddBuilder(replyObject, false);
+	}
+
+	public void setTargetChannel(MessageChannel targetChannel) {
+		this.targetChannel = targetChannel;
+	}
+
+	public void setTargetChannelName(String targetChannelName) {
+		this.targetChannel = targetChannelName;
+	}
+
+	protected Object getTargetChannel() {
+		return this.targetChannel;
+	}
+
+	public boolean isEmpty() {
+		return this.builders.isEmpty();
+	}
+
+	public List<MessageBuilder<?>> builders() {
+		return Collections.unmodifiableList(this.builders);
+	}
+
+	private MessageBuilder<?> createAndAddBuilder(Object replyObject, boolean clearExistingValues) {
+		MessageBuilder<?> builder = null;
+		if (replyObject instanceof MessageBuilder) {
+			builder = (MessageBuilder<?>) replyObject;
+		}
+		else if (replyObject instanceof Message) {
+			builder = MessageBuilder.fromMessage((Message<?>) replyObject);
+		}
+		else {
+			builder = MessageBuilder.withPayload(replyObject);
+		}
+		synchronized (this.builders) {
+			if (clearExistingValues) {
+				this.builders.clear();
+			}
+			this.builders.add(builder);
+		}
+		return builder;
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceActivatorEndpoint.java

@@ -24,8 +24,8 @@ import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageHandlingException;
 import org.springframework.integration.message.MessageMappingMethodInvoker;
 import org.springframework.integration.util.DefaultMethodResolver;
-import org.springframework.integration.util.MethodResolver;
 import org.springframework.integration.util.MethodInvoker;
+import org.springframework.integration.util.MethodResolver;
 import org.springframework.util.Assert;
 
 /**
@@ -63,9 +63,12 @@ public class ServiceActivatorEndpoint extends AbstractReplyProducingMessageConsu
 	}
 
 	@Override
-	protected Object handle(Message<?> message) {
+	protected void handle(Message<?> message, ReplyHolder replyHolder) {
 		try {
-			return this.invoker.invokeMethod(message);
+			Object result = this.invoker.invokeMethod(message);
+			if (result != null) {
+				replyHolder.set(result);
+			}
 		}
 		catch (Exception e) {
 			if (e instanceof RuntimeException) {

org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.filter;
 
 import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
+import org.springframework.integration.endpoint.ReplyHolder;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.selector.MessageSelector;
 import org.springframework.util.Assert;
@@ -40,11 +41,10 @@ public class MessageFilter extends AbstractReplyProducingMessageConsumer {
 
 
 	@Override
-	protected Message<?> handle(Message<?> message) {
+	protected void handle(Message<?> message, ReplyHolder replyHolder) {
 		if (this.selector.accept(message)) {
-			return message;
+			replyHolder.set(message);
 		}
-		return null;
 	}
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java

@@ -26,6 +26,7 @@ import org.springframework.integration.endpoint.AbstractReplyProducingMessageCon
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.endpoint.MessagingGateway;
 import org.springframework.integration.endpoint.PollingConsumerEndpoint;
+import org.springframework.integration.endpoint.ReplyHolder;
 import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageConsumer;
@@ -154,8 +155,8 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
 			MessageEndpoint correlator = null;
 			MessageConsumer consumer = new AbstractReplyProducingMessageConsumer() {
 				@Override
-				protected Object handle(Message<?> message) {
-					return message;
+				protected void handle(Message<?> message, ReplyHolder replyHolder) {
+					replyHolder.set(message);
 				}
 			};
 			if (this.replyChannel instanceof SubscribableChannel) {

org.springframework.integration/src/main/java/org/springframework/integration/message/CompositeMessage.java

@@ -1,42 +0,0 @@
-/*
- * Copyright 2002-2008 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.integration.message;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-
-/**
- * @author Mark Fisher
- */
-public class CompositeMessage extends GenericMessage<List<Message<?>>> implements Iterable<Message<?>> {
-
-	public CompositeMessage(Message<?>[] messages) {
-		this(Arrays.asList(messages));
-	}
-
-	public CompositeMessage(List<Message<?>> messages) {
-		super(Collections.unmodifiableList(messages));
-	}
-
-
-	public Iterator<Message<?>> iterator() {
-		return this.getPayload().iterator();
-	}
-
-}

org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java

@@ -35,13 +35,21 @@ public final class MessageBuilder<T> {
 
 	private final Map<String, Object> headers = new HashMap<String, Object>();
 
+	private final Message<T> originalMessage;
+
+	private volatile boolean modified;
+
 
 	/**
 	 * Private constructor to be invoked from the static factory methods only.
 	 */
-	private MessageBuilder(T payload) {
+	private MessageBuilder(T payload, Message<T> originalMessage) {
 		Assert.notNull(payload, "payload must not be null");
 		this.payload = payload;
+		this.originalMessage = originalMessage;
+		if (originalMessage != null) {
+			this.headers.putAll(originalMessage.getHeaders());
+		}
 	}
 
 
@@ -54,8 +62,8 @@ public final class MessageBuilder<T> {
 	 * will be copied
 	 */
 	public static <T> MessageBuilder<T> fromMessage(Message<T> message) {
-		MessageBuilder<T> builder = new MessageBuilder<T>(message.getPayload());
-		builder.headers.putAll(message.getHeaders());
+		Assert.notNull(message, "message must not be null");
+		MessageBuilder<T> builder = new MessageBuilder<T>(message.getPayload(), message);
 		return builder;
 	}
 
@@ -65,7 +73,7 @@ public final class MessageBuilder<T> {
 	 * @param payload the payload for the new message
 	 */
 	public static <T> MessageBuilder<T> withPayload(T payload) {
-		MessageBuilder<T> builder = new MessageBuilder<T>(payload);
+		MessageBuilder<T> builder = new MessageBuilder<T>(payload, null);
 		return builder;
 	}
 
@@ -76,6 +84,7 @@ public final class MessageBuilder<T> {
 	 */
 	public MessageBuilder<T> setHeader(String headerName, Object headerValue) {
 		if (StringUtils.hasLength(headerName) && !(this.isReadOnly(headerName))) {
+			this.modified = true;
 			if (headerValue == null) {
 				this.headers.remove(headerName);
 			}
@@ -102,6 +111,7 @@ public final class MessageBuilder<T> {
 	 */
 	public MessageBuilder<T> removeHeader(String headerName) {
 		if (StringUtils.hasLength(headerName)) {
+			this.modified = true;
 			this.headers.remove(headerName);
 		}
 		return this;
@@ -174,6 +184,9 @@ public final class MessageBuilder<T> {
 	}
 
 	public Message<T> build() {
+		if (!this.modified && this.originalMessage != null) {
+			return this.originalMessage;
+		}
 		return new GenericMessage<T>(this.payload, this.headers);
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java

@@ -16,15 +16,11 @@
 
 package org.springframework.integration.splitter;
 
-import java.util.ArrayList;
 import java.util.Collection;
-import java.util.List;
 
 import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
-import org.springframework.integration.message.CompositeMessage;
+import org.springframework.integration.endpoint.ReplyHolder;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageBuilder;
-import org.springframework.integration.message.MessageHeaders;
 
 /**
  * Base class for Message-splitting consumers.
@@ -34,24 +30,18 @@ import org.springframework.integration.message.MessageHeaders;
 public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageConsumer {
 
 	@Override
-	protected final boolean shouldSplitComposite() {
-		return true;
-	}
-
-	@Override
-	protected final Message<?> handle(Message<?> message) {
+	protected final void handle(Message<?> message, ReplyHolder replyHolder) {
 		Object result = this.splitMessage(message);
 		if (result == null) {
-			return null;
+			return;
 		}
-		MessageHeaders requestHeaders = message.getHeaders();
-		List<Message<?>> results = new ArrayList<Message<?>>();
+		Object correlationId = message.getHeaders().getId();
 		if (result instanceof Collection) {
 			Collection<?> items = (Collection<?>) result;
 			int sequenceNumber = 0;
 			int sequenceSize = items.size();
 			for (Object item : items) {
-				results.add(this.createSplitMessage(item, requestHeaders, ++sequenceNumber, sequenceSize));
+				this.addReply(replyHolder, item, correlationId, ++sequenceNumber, sequenceSize);
 			}
 		}
 		else if (result.getClass().isArray()) {
@@ -59,16 +49,18 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 			int sequenceNumber = 0;
 			int sequenceSize = items.length;
 			for (Object item : items) {
-				results.add(this.createSplitMessage(item, requestHeaders, ++sequenceNumber, sequenceSize));
+				this.addReply(replyHolder, item, correlationId, ++sequenceNumber, sequenceSize);
 			}
 		}
 		else {
-			results.add(this.createSplitMessage(result, requestHeaders, 1, 1));
+			this.addReply(replyHolder, result, correlationId, 1, 1);
 		}
-		if (results.isEmpty()) {
-			return null;
-		}
-		return new CompositeMessage(results);
+	}
+
+	private void addReply(ReplyHolder replyHolder, Object item, Object correlationId, int sequenceNumber, int sequenceSize) {
+		replyHolder.add(item).setCorrelationId(correlationId)
+				.setSequenceNumber(sequenceNumber)
+				.setSequenceSize(sequenceSize);
 	}
 
 	/**
@@ -81,20 +73,4 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 	 */
 	protected abstract Object splitMessage(Message<?> message);
 
-
-	private Message<?> createSplitMessage(Object item, MessageHeaders requestHeaders, int sequenceNumber, int sequenceSize) {
-		if (item instanceof Message<?>) {
-			return setSplitMessageHeaders(MessageBuilder.fromMessage((Message<?>) item),
-					requestHeaders.getId(), sequenceNumber, sequenceSize);
-		}
-		return setSplitMessageHeaders(MessageBuilder.withPayload(item),
-				requestHeaders.getId(), sequenceNumber, sequenceSize);
-	}
-
-	private Message<?> setSplitMessageHeaders(MessageBuilder<?> builder, Object requestMessageId, int sequenceNumber, int sequenceSize) {
-		return builder.setCorrelationId(requestMessageId)
-				.setSequenceNumber(sequenceNumber)
-				.setSequenceSize(sequenceSize).build();
-	}
-
 }

org.springframework.integration/src/main/java/org/springframework/integration/transformer/MessageTransformingConsumer.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.transformer;
 
 import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
+import org.springframework.integration.endpoint.ReplyHolder;
 import org.springframework.integration.message.Message;
 import org.springframework.util.Assert;
 
@@ -43,8 +44,11 @@ public class MessageTransformingConsumer extends AbstractReplyProducingMessageCo
 
 
 	@Override
-	protected Message<?> handle(Message<?> message) {
-		return transformer.transform(message);
+	protected void handle(Message<?> message, ReplyHolder replyHolder) {
+		Message<?> result = transformer.transform(message);
+		if (result != null) {
+			replyHolder.set(result);
+		}
 	}
 
 }