From ac770ce178acd8987901cdb4392db487a79c8b14 Mon Sep 17 00:00:00 2001
From: Mark Fisher <markfisher@vmware.com>
Date: Tue, 12 Aug 2008 11:52:25 +0000
Subject: [PATCH] MessageTarget is now sufficient for the replyTarget (no
 longer requires MessageChannel), so for example it can now support the
 temporary returnAddress set by the MessageExchangeTemplate. Also, the
 'outputChannel' is given precedence while 'returnAddress' is used as a
 fallback. This is consistent with the other Message-handling endpoint types
 (INT-332, INT-333).

---
 .../AbstractMessageBarrierHandler.java        | 31 ++++++++++++-------
 1 file changed, 19 insertions(+), 12 deletions(-)

diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java
index a2e24e89be..c6aebff690 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java
@@ -32,8 +32,10 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.message.BlockingTarget;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.message.MessageTarget;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 
@@ -201,18 +203,23 @@ public abstract class AbstractMessageBarrierHandler implements MessageHandler, I
 	private void afterRelease(Object correlationId, List<Message<?>> releasedMessages) {
 		Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
 		for (Message<?> result : processedMessages) {
-			MessageChannel replyChannel = this.resolveReplyChannelFromMessage(result);
-			if (replyChannel == null) {
-				replyChannel = this.resolveReplyChannelFromMessage(releasedMessages.get(0));
-				if (replyChannel == null) {
-					replyChannel = this.outputChannel;
+			MessageTarget replyTarget = this.outputChannel;
+			if (replyTarget == null) {
+				replyTarget = this.resolveReplyTargetFromMessage(result);
+				if (replyTarget == null) {
+					replyTarget = this.resolveReplyTargetFromMessage(releasedMessages.get(0));
 				}
 			}
-			if (replyChannel != null) {
-				replyChannel.send(result, this.sendTimeout);
+			if (replyTarget != null) {
+				if (replyTarget instanceof BlockingTarget && this.sendTimeout >= 0) {
+					((BlockingTarget) replyTarget).send(result, this.sendTimeout);
+				}
+				else {
+					replyTarget.send(result);
+				}
 			}
 			else if (logger.isWarnEnabled()) {
-				logger.warn("unable to determine reply channel for aggregation result: " + result);
+				logger.warn("unable to determine reply target for aggregation result: " + result);
 			}
 		}
 	}
@@ -227,14 +234,14 @@ public abstract class AbstractMessageBarrierHandler implements MessageHandler, I
 		}
 	}
 
-	protected MessageChannel resolveReplyChannelFromMessage(Message<?> message) {
+	protected MessageTarget resolveReplyTargetFromMessage(Message<?> message) {
 		Object returnAddress = message.getHeaders().getReturnAddress();
 		if (returnAddress != null) {
-			if (returnAddress instanceof MessageChannel) {
-				return (MessageChannel) returnAddress;
+			if (returnAddress instanceof MessageTarget) {
+				return (MessageTarget) returnAddress;
 			}
 			if (logger.isWarnEnabled()) {
-				logger.warn("Aggregator can only reply to a 'returnAddress' of type MessageChannel.");
+				logger.warn("Aggregator can only reply to a 'returnAddress' of type MessageTarget.");
 			}
 		}
 		return null;