From 2510a7ad757635a07ea77061ed8fb6b54eb009f0 Mon Sep 17 00:00:00 2001
From: Mark Fisher <markfisher@vmware.com>
Date: Tue, 20 Apr 2010 23:13:10 +0000
Subject: [PATCH] INT-1059 added AbstractAggregatingMessageGroupProcessor

---
 ...tractAggregatingMessageGroupProcessor.java | 51 +++++++++++++++++++
 ...faultAggregatingMessageGroupProcessor.java | 34 ++++++-------
 2 files changed, 68 insertions(+), 17 deletions(-)
 create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java

diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
new file mode 100644
index 0000000000..45862fb1e4
--- /dev/null
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2002-2010 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.aggregator;
+
+import java.util.Map;
+
+import org.springframework.integration.channel.MessageChannelTemplate;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessageChannel;
+import org.springframework.integration.message.MessageBuilder;
+import org.springframework.util.Assert;
+
+/**
+ * Base class for MessageGroupProcessor implementations that aggregate the
+ * group of Messages into a single Message.
+ * 
+ * @author Iwein Fuld
+ * @author Alexander Peters
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public abstract class AbstractAggregatingMessageGroupProcessor implements MessageGroupProcessor {
+
+	public final void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) {
+		Assert.notNull(group, "MessageGroup must not be null");
+		Assert.notNull(outputChannel, "'outputChannel' must not be null");
+		Object payload = this.aggregatePayloads(group);
+		Map<String, Object> headers = this.aggregateHeaders(group);
+		Message<?> message = MessageBuilder.withPayload(payload).copyHeadersIfAbsent(headers).build();
+		channelTemplate.send(message, outputChannel);
+	}
+
+	protected abstract Map<String, Object> aggregateHeaders(MessageGroup group);
+
+	protected abstract Object aggregatePayloads(MessageGroup group);
+
+}
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java
index 27c401de49..0da367891d 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java
@@ -16,15 +16,13 @@
 
 package org.springframework.integration.aggregator;
 
-import org.springframework.integration.channel.MessageChannelTemplate;
-import org.springframework.integration.core.Message;
-import org.springframework.integration.core.MessageChannel;
-import org.springframework.integration.message.MessageBuilder;
-import org.springframework.util.Assert;
-
 import java.util.ArrayList;
-import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+
+import org.springframework.integration.core.Message;
+import org.springframework.util.Assert;
 
 /**
  * This implementation of MessageGroupProcessor will take the messages from the
@@ -32,24 +30,26 @@ import java.util.List;
  * 
  * @author Iwein Fuld
  * @author Alexander Peters
- * @since 2.0.0
+ * @author Mark Fisher
+ * @since 2.0
  */
-public class DefaultAggregatingMessageGroupProcessor implements MessageGroupProcessor {
+public class DefaultAggregatingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor {
 
-	public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) {
-		Assert.notNull(group, "MessageGroup must not be null");
-		Assert.notNull(outputChannel, "'outputChannel' must not be null");
-		List<Message<?>> messages = group.getMessages();
-		Assert.notEmpty(messages, this.getClass().getSimpleName() + " cannot process empty message groups");
-		channelTemplate.send(aggregateMessages(messages), outputChannel);
+	@Override
+	protected Map<String, Object> aggregateHeaders(MessageGroup group) {
+		// TODO: return all non-conflicting headers
+		return new HashMap<String, Object>();
 	}
 
-	private Message<? extends Collection<?>> aggregateMessages(List<Message<?>> messages) {
+	@Override
+	protected Object aggregatePayloads(MessageGroup group) {
+		List<Message<?>> messages = group.getMessages();
+		Assert.notEmpty(messages, this.getClass().getSimpleName() + " cannot process empty message groups");
 		List<Object> payloads = new ArrayList<Object>(messages.size());
 		for (Message<?> message : messages) {
 			payloads.add(message.getPayload());
 		}
-		return MessageBuilder.withPayload(payloads).build();
+		return payloads;
 	}
 
 }