You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@synapse.apache.org by ru...@apache.org on 2008/05/03 08:01:54 UTC
svn commit: r653007 - in
/synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator:
Aggregate.java AggregateMediator.java
Author: ruwan
Date: Fri May 2 23:01:54 2008
New Revision: 653007
URL: http://svn.apache.org/viewvc?rev=653007&view=rev
Log:
Fixing an issue with aggregate mediator
Modified:
synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator/Aggregate.java
synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator/AggregateMediator.java
Modified: synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator/Aggregate.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator/Aggregate.java?rev=653007&r1=653006&r2=653007&view=diff
==============================================================================
--- synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator/Aggregate.java (original)
+++ synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator/Aggregate.java Fri May 2 23:01:54 2008
@@ -51,6 +51,7 @@
private AggregateMediator aggregateMediator = null;
private List<MessageContext> messages = new ArrayList<MessageContext>();
private boolean locked = false;
+ private boolean completed = false;
/**
* Save aggregation properties and timeout
@@ -226,18 +227,34 @@
}
public void run() {
- if (log.isDebugEnabled()) {
- log.debug("Time : " + System.currentTimeMillis() + " and this aggregator expired at : " +
- expiryTimeMillis);
+ while (true) {
+ if (completed) {
+ break;
+ }
+ if (getLock()) {
+ if (log.isDebugEnabled()) {
+ log.debug("Time : " + System.currentTimeMillis() + " and this aggregator " +
+ "expired at : " + expiryTimeMillis);
+ }
+ aggregateMediator.completeAggregate(this);
+ break;
+ }
}
- aggregateMediator.completeAggregate(this);
}
- public boolean isLocked() {
- return locked;
+ public synchronized boolean getLock() {
+ return !locked;
+ }
+
+ public void releaseLock() {
+ locked = false;
+ }
+
+ public boolean isCompleted() {
+ return completed;
}
- public synchronized void setLocked(boolean locked) {
- this.locked = locked;
+ public void setCompleted(boolean completed) {
+ this.completed = completed;
}
}
Modified: synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator/AggregateMediator.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator/AggregateMediator.java?rev=653007&r1=653006&r2=653007&view=diff
==============================================================================
--- synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator/AggregateMediator.java (original)
+++ synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator/AggregateMediator.java Fri May 2 23:01:54 2008
@@ -125,38 +125,42 @@
if (correlateExpression != null
&& correlateExpression.evaluate(synCtx) != null) {
- if (activeAggregates.containsKey(correlateExpression.toString())) {
- while (aggregate == null) {
- aggregate = activeAggregates.get(correlateExpression.toString());
- if (aggregate != null) {
- if (aggregate.isLocked()) {
- aggregate = null;
- } else {
- aggregate.setLocked(true);
+ while (aggregate == null) {
+
+ synchronized (lock) {
+
+ if (activeAggregates.containsKey(correlateExpression.toString())) {
+
+ aggregate = activeAggregates.get(correlateExpression.toString());
+ if (aggregate != null) {
+ if (!aggregate.getLock()) {
+ aggregate = null;
+ }
}
+
} else {
- break;
- }
- }
- } else {
- if (traceOrDebugOn) {
- traceOrDebug(traceOn, "Creating new Aggregator - " +
- (completionTimeoutMillis > 0 ? "expires in : "
- + (completionTimeoutMillis / 1000) + "secs" :
- "without expiry time"));
- }
+ if (traceOrDebugOn) {
+ traceOrDebug(traceOn, "Creating new Aggregator - " +
+ (completionTimeoutMillis > 0 ? "expires in : "
+ + (completionTimeoutMillis / 1000) + "secs" :
+ "without expiry time"));
+ }
- aggregate = new Aggregate(
- correlateExpression.toString(),
- completionTimeoutMillis,
- minMessagesToComplete,
- maxMessagesToComplete, this);
- if (completionTimeoutMillis > 0) {
- synCtx.getConfiguration().getSynapseTimer().
- schedule(aggregate, completionTimeoutMillis);
+ aggregate = new Aggregate(
+ correlateExpression.toString(),
+ completionTimeoutMillis,
+ minMessagesToComplete,
+ maxMessagesToComplete, this);
+
+ if (completionTimeoutMillis > 0) {
+ synCtx.getConfiguration().getSynapseTimer().
+ schedule(aggregate, completionTimeoutMillis);
+ }
+ aggregate.getLock();
+ activeAggregates.put(correlateExpression.toString(), aggregate);
+ }
}
- activeAggregates.put(correlateExpression.toString(), aggregate);
}
} else if (synCtx.getProperty(EIPConstants.AGGREGATE_CORRELATION) != null) {
@@ -170,39 +174,39 @@
if (o != null && o instanceof String) {
correlation = (String) o;
- if (activeAggregates.containsKey(correlation)) {
- while (aggregate == null) {
- aggregate = activeAggregates.get(correlation);
- if (aggregate != null) {
- if (aggregate.isLocked()) {
- aggregate = null;
+ while (aggregate == null) {
+ synchronized (lock) {
+ if (activeAggregates.containsKey(correlation)) {
+ aggregate = activeAggregates.get(correlation);
+ if (aggregate != null) {
+ if (!aggregate.getLock()) {
+ aggregate = null;
+ }
} else {
- aggregate.setLocked(true);
+ break;
}
} else {
- break;
- }
- }
-
- } else {
- if (traceOrDebugOn) {
- traceOrDebug(traceOn, "Creating new Aggregator - " +
- (completionTimeoutMillis > 0 ? "expires in : "
- + (completionTimeoutMillis / 1000) + "secs" :
- "without expiry time"));
- }
+ if (traceOrDebugOn) {
+ traceOrDebug(traceOn, "Creating new Aggregator - " +
+ (completionTimeoutMillis > 0 ? "expires in : "
+ + (completionTimeoutMillis / 1000) + "secs" :
+ "without expiry time"));
+ }
- aggregate = new Aggregate(
- correlation,
- completionTimeoutMillis,
- minMessagesToComplete,
- maxMessagesToComplete, this);
-
- if (completionTimeoutMillis > 0) {
- synCtx.getConfiguration().getSynapseTimer().
- schedule(aggregate, completionTimeoutMillis);
+ aggregate = new Aggregate(
+ correlation,
+ completionTimeoutMillis,
+ minMessagesToComplete,
+ maxMessagesToComplete, this);
+
+ if (completionTimeoutMillis > 0) {
+ synCtx.getConfiguration().getSynapseTimer().
+ schedule(aggregate, completionTimeoutMillis);
+ }
+ aggregate.getLock();
+ activeAggregates.put(correlation, aggregate);
+ }
}
- activeAggregates.put(correlation, aggregate);
}
} else {
@@ -213,7 +217,8 @@
}
} else {
if (traceOrDebugOn) {
- traceOrDebug(traceOn, "Unable to find aggrgation correlation XPath or property");
+ traceOrDebug(traceOn,
+ "Unable to find aggrgation correlation XPath or property");
}
return true;
}
@@ -244,7 +249,7 @@
}
return true;
} else {
- aggregate.setLocked(false);
+ aggregate.releaseLock();
}
} else {
@@ -281,6 +286,7 @@
// cancel the timer
aggregate.cancel();
+ aggregate.setCompleted(true);
MessageContext newSynCtx = getAggregatedMessage(aggregate);
if (newSynCtx == null) {