Author: davsclaus
Date: Thu Jun 2 12:36:50 2011
New Revision: 1130524
URL: http://svn.apache.org/viewvc?rev=1130524&view=rev
Log:
CAMEL-4037: Aggregate EIP with only completion timeout condition will restore
timeout map upon restart based on exchanges from aggregation repository.
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java?rev=1130524&r1=1130523&r2=1130524&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
Thu Jun 2 12:36:50 2011
@@ -198,8 +198,7 @@ public class AggregateProcessor extends
* @param key the correlation key
* @param exchange the exchange
* @return the aggregated exchange
- * @throws org.apache.camel.CamelExchangeException
- * is thrown if error aggregating
+ * @throws org.apache.camel.CamelExchangeException is thrown if error
aggregating
*/
private Exchange doAggregation(String key, Exchange exchange) throws
CamelExchangeException {
LOG.trace("onAggregation +++ start +++ with correlation key: {}", key);
@@ -408,15 +407,15 @@ public class AggregateProcessor extends
* Then the existing exchanges from the {@link AggregationRepository} must
have its timeout conditions restored.
*/
protected void restoreTimeoutMapFromAggregationRepository() throws
Exception {
- StopWatch watch = new StopWatch();
- LOG.trace("Starting restoring CompletionTimeout for existing exchanges
from the aggregation repository...");
-
// grab the timeout value for each partly aggregated exchange
Set<String> keys = aggregationRepository.getKeys();
if (keys == null || keys.isEmpty()) {
return;
}
+ StopWatch watch = new StopWatch();
+ LOG.trace("Starting restoring CompletionTimeout for {} existing
exchanges from the aggregation repository...", keys.size());
+
for (String key : keys) {
Exchange exchange = aggregationRepository.get(camelContext, key);
// grab the timeout value
