Added Aggregator, AggregatingMessageHandler, RoutingBarrier, RoutingBarrierCompletionStrategy, and related classes.

spring-integration-core/src/main/java/org/springframework/integration/MessageDeliveryException.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration;
 
+import org.springframework.integration.message.Message;
+
 /**
  * Exception that indicates an error during message delivery.
  * 
@@ -24,16 +26,36 @@ package org.springframework.integration;
 @SuppressWarnings("serial")
 public class MessageDeliveryException extends MessagingException {
 
+	private Message<?> undeliveredMessage;
+
+
 	public MessageDeliveryException() {
 		super();
 	}
 
-	public MessageDeliveryException(String message) {
-		super(message);
+	public MessageDeliveryException(Message<?> undeliveredMessage) {
+		this.undeliveredMessage = undeliveredMessage;
 	}
 
-	public MessageDeliveryException(String message, Throwable cause) {
-		super(message, cause);
+	public MessageDeliveryException(String description) {
+		super(description);
+	}
+
+	public MessageDeliveryException(Message<?> undeliveredMessage, String description) {
+		super(description);
+		this.undeliveredMessage = undeliveredMessage;
+	}
+
+	public MessageDeliveryException(String description, Throwable cause) {
+		super(description, cause);
+	}
+
+
+	/**
+	 * Return the undelivered {@link Message} if available, may be null.
+	 */
+	public Message<?> getUndeliveredMessage() {
+		return this.undeliveredMessage;
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2002-2007 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.router;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.integration.MessageHandlingException;
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.message.Message;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link MessageHandler} implementation that waits for a <em>complete</em>
+ * group of {@link Message Messages} to arrive and then delegates to an
+ * {@link Aggregator} to combine them into a single {@link Message}.
+ * <p>
+ * Each {@link Message} that is received by this handler will be associated with
+ * a group based upon the '<code>correlationId</code>' property of its
+ * header. If no such property is available, a {@link MessageHandlingException}
+ * will be thrown.
+ * <p>
+ * The default strategy for determining whether a group is complete is based on
+ * the '<code>sequenceSize</code>' property of the header. Alternatively, a
+ * custom implementation of the {@link RoutingBarrierCompletionStrategy} may be
+ * provided.
+ * <p>
+ * The '<code>timeout</code>' value determines how long to wait for the
+ * complete group after the arrival of the first {@link Message} of the group.
+ * The default value is 1 minute. If the timeout elapses prior to completion,
+ * the handler will throw a {@link MessageHandlingException} by default. To
+ * prevent the exception and aggregate the group even when incomplete, set the
+ * '<code>shouldFailOnTimeout</code>' property to '<code>false</code>'.
+ * 
+ * @author Mark Fisher
+ */
+public class AggregatingMessageHandler implements MessageHandler {
+
+	private long timeout = 60000;
+
+	private boolean shouldFailOnTimeout = true;
+
+	private Aggregator aggregator;
+
+	private RoutingBarrierCompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
+
+	private ConcurrentHashMap<Object, RoutingBarrier> barriers = new ConcurrentHashMap<Object, RoutingBarrier>();
+
+
+	/**
+	 * Create a handler that delegates to the provided aggregator to combine a
+	 * group of messages into a single message.
+	 */
+	public AggregatingMessageHandler(Aggregator aggregator) {
+		Assert.notNull(aggregator, "'aggregator' must not be null");
+		this.aggregator = aggregator;
+	}
+
+
+	/**
+	 * Strategy to determine whether the group of messages is complete.
+	 */
+	public void setCompletionStrategy(RoutingBarrierCompletionStrategy completionStrategy) {
+		Assert.notNull(completionStrategy, "'completionStrategy' must not be null");
+		this.completionStrategy = completionStrategy;
+	}
+
+	/**
+	 * Maximum time to wait (in milliseconds) for the completion strategy to
+	 * become true.
+	 */
+	public void setTimeout(long timeout) {
+		Assert.isTrue(timeout >= 0, "'timeout' must not be negative");
+		this.timeout = timeout;
+	}
+
+	/**
+	 * Specify whether this handler should throw a {@link MessageHandlingException}
+	 * when a message group does not reach completion within the allotted time. The
+	 * default is '<code>true</code>'. Setting this to '<code>false</code>' will cause
+	 * the {@link Aggregator} to be invoked even when the group is incomplete.
+	 */
+	public void setShouldFailOnTimeout(boolean setShouldFailOnTimeout) {
+		this.shouldFailOnTimeout = setShouldFailOnTimeout;
+	}
+
+	public Message<?> handle(Message<?> message) {
+		Object correlationId = message.getHeader().getCorrelationId();
+		if (correlationId == null) {
+			throw new MessageHandlingException(this.getClass().getSimpleName() +
+					" requires the 'correlationId' property");
+		}
+		RoutingBarrier barrier = barriers.putIfAbsent(correlationId, new RoutingBarrier(this.completionStrategy));
+		if (barrier == null) {
+			try {
+				barrier = barriers.get(correlationId);
+				barrier.addMessage(message);
+				if (!barrier.waitForCompletion(this.timeout) && this.shouldFailOnTimeout) {
+					throw new MessageHandlingException("aggregation did not complete "
+							+ "within the allotted time limit of " + this.timeout + " milliseconds");
+				}
+				Message<?> result = aggregator.aggregate(barrier.getMessages());
+				return result;
+			}
+			finally {
+				this.barriers.remove(correlationId);
+			}
+		}
+		else {
+			barriers.get(correlationId).addMessage(message);
+			return null;
+		}
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2002-2007 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.router;
+
+import java.util.List;
+
+import org.springframework.integration.message.Message;
+
+/**
+ * Strategy interface for aggregating a list of {@link Message Messages} into a
+ * single {@link Message}.
+ * 
+ * @author Mark Fisher
+ */
+public interface Aggregator {
+
+	Message<?> aggregate(List<Message<?>> messages);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/router/MessageSequenceComparator.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2002-2007 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.router;
+
+import java.util.Comparator;
+
+import org.springframework.integration.message.Message;
+
+/**
+ * A {@link Comparator} implementation based on the '<code>sequenceNumber</code>'
+ * property of a {@link Message Message's} header.
+ * 
+ * @author Mark Fisher
+ */
+public class MessageSequenceComparator implements Comparator<Message<?>> {
+
+	public int compare(Message<?> message1, Message<?> message2) {
+		int s1 = message1.getHeader().getSequenceNumber();
+		int s2 = message2.getHeader().getSequenceNumber();
+		return (s1 < s2) ? -1 : (s1 == s2) ? 0 : 1;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2002-2007 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.router;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.springframework.integration.message.Message;
+import org.springframework.util.Assert;
+
+/**
+ * A rendezvous point for {@link Message Messages} that delegates to a
+ * {@link RoutingBarrierCompletionStrategy} to determine when a
+ * <em>complete</em> message group is available.
+ * 
+ * @author Mark Fisher
+ */
+public class RoutingBarrier {
+
+	private List<Message<?>> messages = new CopyOnWriteArrayList<Message<?>>();
+
+	private final RoutingBarrierCompletionStrategy completionStrategy;
+
+	private volatile boolean complete = false;
+
+	private ReentrantLock lock = new ReentrantLock();
+
+	private Condition condition = lock.newCondition();
+
+
+	public RoutingBarrier(RoutingBarrierCompletionStrategy completionStrategy) {
+		Assert.notNull(completionStrategy, "'completionStrategy' must not be null");
+		this.completionStrategy = completionStrategy;
+	}
+
+
+	public void addMessage(Message<?> message) {
+		this.messages.add(message);
+		if (this.completionStrategy.isComplete(this.messages)) {
+			try {
+				this.lock.lock();
+				if (!this.complete) {
+					this.complete = true;
+					this.condition.signalAll();
+				}
+			}
+			finally {
+				this.lock.unlock();
+			}
+		}
+	}
+
+	public boolean waitForCompletion(long timeout) {
+		if (this.complete) {
+			return true;
+		}
+		lock.lock();
+		try {
+			if (this.complete) {
+				return true;
+			}
+			if (timeout >= 0) {
+				return this.condition.await(timeout, TimeUnit.MILLISECONDS);
+			}
+			else {
+				this.condition.await();
+				return true;
+			}
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			return false;
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	public boolean isComplete() {
+		return this.complete;
+	}
+
+	public List<Message<?>> getMessages() {
+		return this.messages;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrierCompletionStrategy.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2002-2007 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.router;
+
+import java.util.List;
+
+import org.springframework.integration.message.Message;
+
+/**
+ * Strategy for determining when a group of messages reaches a state of
+ * completion (i.e. can trip a barrier).
+ * 
+ * @author Mark Fisher
+ * @see RoutingBarrier
+ */
+public interface RoutingBarrierCompletionStrategy {
+
+	boolean isComplete(List<Message<?>> messages);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/router/SequenceSizeCompletionStrategy.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2002-2007 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.router;
+
+import java.util.List;
+
+import org.springframework.integration.message.Message;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * An implementation of {@link RoutingBarrierCompletionStrategy} that simply
+ * compares the current size of the message list to the expected 'sequenceSize'
+ * according to the first {@link Message} in the list.
+ * 
+ * @author Mark Fisher
+ */
+public class SequenceSizeCompletionStrategy implements RoutingBarrierCompletionStrategy {
+
+	public boolean isComplete(List<Message<?>> messages) {
+		if (CollectionUtils.isEmpty(messages)) {
+			return false;
+		}
+		return (messages.size() >= messages.get(0).getHeader().getSequenceSize());
+	}
+
+}