You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2010/04/16 13:57:09 UTC
svn commit: r934812 - in /camel/trunk:
camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.java
Author: davsclaus
Date: Fri Apr 16 11:57:09 2010
New Revision: 934812
URL: http://svn.apache.org/viewvc?rev=934812&view=rev
Log:
camel-hawtdb is sync by default. Also better guard timeout completed exchanges from being completed in the mean time.
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
camel/trunk/components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.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=934812&r1=934811&r2=934812&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 Fri Apr 16 11:57:09 2010
@@ -542,14 +542,24 @@ public class AggregateProcessor extends
log.debug("Completion timeout triggered for correlation key: " + key);
}
- // get the aggregated exchange
- Exchange answer = aggregationRepository.get(camelContext, key);
-
- // indicate it was completed by timeout
- answer.setProperty(Exchange.AGGREGATED_COMPLETED_BY, "timeout");
-
try {
lock.lock();
+
+ // double check that its not already in progress
+ boolean inProgress = inProgressCompleteExchanges.contains(exchangeId);
+ if (inProgress) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Aggregated exchange with id: " + exchangeId + " is already in progress.");
+ }
+ return;
+ }
+
+ // get the aggregated exchange
+ Exchange answer = aggregationRepository.get(camelContext, key);
+
+ // indicate it was completed by timeout
+ answer.setProperty(Exchange.AGGREGATED_COMPLETED_BY, "timeout");
+
onCompletion(key, answer, true);
} finally {
lock.unlock();
Modified: camel/trunk/components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.java?rev=934812&r1=934811&r2=934812&view=diff
==============================================================================
--- camel/trunk/components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.java (original)
+++ camel/trunk/components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.java Fri Apr 16 11:57:09 2010
@@ -47,7 +47,7 @@ public class HawtDBAggregationRepository
private String persistentFileName;
private String repositoryName;
private Integer bufferSize;
- private boolean sync;
+ private boolean sync = true;
private boolean returnOldExchange;
private HawtDBCamelMarshaller<K> marshaller = new HawtDBCamelMarshaller<K>();
private long recoveryInterval = 5000;