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/03/26 06:08:29 UTC
svn commit: r641151 - in
/synapse/trunk/java/modules/core/src/main/java/org/apache/synapse/mediators/eip/aggregator:
Aggregate.java AggregateMediator.java
Author: ruwan
Date: Tue Mar 25 22:08:28 2008
New Revision: 641151
URL: http://svn.apache.org/viewvc?rev=641151&view=rev
Log:
Fixing the issue SYNAPSE-257
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=641151&r1=641150&r2=641151&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 Tue Mar 25 22:08:28 2008
@@ -60,7 +60,9 @@
* @param max the maximum number of messages to be aggregated
* @param mediator
*/
- public Aggregate(String corelation, long timeoutMillis, int min, int max, AggregateMediator mediator) {
+ public Aggregate(String corelation, long timeoutMillis, int min,
+ int max, AggregateMediator mediator) {
+
this.correlation = corelation;
if (timeoutMillis > 0) {
expiryTimeMillis = System.currentTimeMillis() + timeoutMillis;
@@ -80,7 +82,7 @@
* @param synCtx message to be added into this aggregation group
* @return true if the message was added or false if not
*/
- public boolean addMessage(MessageContext synCtx) {
+ public synchronized boolean addMessage(MessageContext synCtx) {
if (maxCount <= 0 || (maxCount > 0 && messages.size() < maxCount)) {
messages.add(synCtx);
return true;
@@ -92,9 +94,15 @@
/**
* Has this aggregation group completed?
*
+ * @param traceOn is tracing on
+ * @param traceOrDebugOn is trace or debug on
+ * @param trace trace log to be used
+ * @param log log to be used
+ *
* @return boolean true if aggregation is complete
*/
- public boolean isComplete(boolean traceOn, boolean traceOrDebugOn, Log trace, Log log) {
+ public synchronized boolean isComplete(boolean traceOn, boolean traceOrDebugOn,
+ Log trace, Log log) {
// if any messages have been collected, check if the completion criteria is met
if (!messages.isEmpty()) {
@@ -104,7 +112,8 @@
Object prop = mc.getProperty(EIPConstants.MESSAGE_SEQUENCE);
if (prop != null && prop instanceof String) {
- String[] msgSequence = prop.toString().split(EIPConstants.MESSAGE_SEQUENCE_DELEMITER);
+ String[] msgSequence = prop.toString().split(
+ EIPConstants.MESSAGE_SEQUENCE_DELEMITER);
int total = Integer.parseInt(msgSequence[1]);
if (traceOrDebugOn) {
@@ -129,7 +138,8 @@
if (minCount > 0 && messages.size() >= minCount) {
if (traceOrDebugOn) {
traceOrDebug(traceOn, trace, log,
- "Aggregation complete - the minimum : " + minCount + " messages has been reached");
+ "Aggregation complete - the minimum : " + minCount
+ + " messages has been reached");
}
return true;
}
@@ -137,7 +147,8 @@
if (maxCount > 0 && messages.size() >= maxCount) {
if (traceOrDebugOn) {
traceOrDebug(traceOn, trace, log,
- "Aggregation complete - the maximum : " + maxCount + " messages has been reached");
+ "Aggregation complete - the maximum : " + maxCount
+ + " messages has been reached");
}
return true;
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=641151&r1=641150&r2=641151&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 Tue Mar 25 22:08:28 2008
@@ -34,7 +34,6 @@
import java.util.Collections;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.Map;
/**
@@ -116,7 +115,7 @@
}
try {
- Aggregate aggregate = null;
+ Aggregate aggregate;
// if a correlateExpression is provided and there is a coresponding
// element in the current message prepare to correlate the messages on that
@@ -127,9 +126,12 @@
aggregate = activeAggregates.get(correlateExpression.toString());
} else {
+
if (traceOrDebugOn) {
- traceOrDebug(traceOn, "Creating new Aggregator - expires in : " +
- (completionTimeoutMillis / 1000) + "secs");
+ traceOrDebug(traceOn, "Creating new Aggregator - " +
+ (completionTimeoutMillis > 0 ? "expires in : "
+ + (completionTimeoutMillis / 1000) + "secs" :
+ "without expiry time"));
}
aggregate = new Aggregate(
@@ -137,8 +139,10 @@
completionTimeoutMillis,
minMessagesToComplete,
maxMessagesToComplete, this);
- synCtx.getConfiguration().getSynapseTimer().
- schedule(aggregate, completionTimeoutMillis);
+ if (completionTimeoutMillis > 0) {
+ synCtx.getConfiguration().getSynapseTimer().
+ schedule(aggregate, completionTimeoutMillis);
+ }
activeAggregates.put(correlateExpression.toString(), aggregate);
}
@@ -149,7 +153,7 @@
// which thus can be used to uniquely group messages into aggregates
Object o = synCtx.getProperty(EIPConstants.AGGREGATE_CORRELATION);
- String correlation = null;
+ String correlation;
if (o != null && o instanceof String) {
correlation = (String) o;
@@ -159,17 +163,22 @@
} else {
if (traceOrDebugOn) {
- traceOrDebug(traceOn, "Creating new Aggregator - expires in : " +
- (completionTimeoutMillis / 1000) + "secs");
+ traceOrDebug(traceOn, "Creating new Aggregator - " +
+ (completionTimeoutMillis > 0 ? "expires in : "
+ + (completionTimeoutMillis / 1000) + "secs" :
+ "without expiry time"));
}
aggregate = new Aggregate(
- correlation,
- completionTimeoutMillis,
- minMessagesToComplete,
+ correlation,
+ completionTimeoutMillis,
+ minMessagesToComplete,
maxMessagesToComplete, this);
- synCtx.getConfiguration().getSynapseTimer().
- schedule(aggregate, completionTimeoutMillis);
+
+ if (completionTimeoutMillis > 0) {
+ synCtx.getConfiguration().getSynapseTimer().
+ schedule(aggregate, completionTimeoutMillis);
+ }
activeAggregates.put(correlation, aggregate);
}
@@ -239,7 +248,7 @@
* itself
* @param aggregate the timed out Aggregate that holds collected messages and properties
*/
- public void completeAggregate(Aggregate aggregate) {
+ public synchronized void completeAggregate(Aggregate aggregate) {
if (log.isDebugEnabled()) {
log.debug("Aggregation completed or timed out");
@@ -254,7 +263,7 @@
return;
}
- activeAggregates.remove(aggregate);
+ activeAggregates.remove(aggregate.getCorrelation());
if ((correlateExpression != null &&
!correlateExpression.toString().equals(aggregate.getCorrelation())) ||
@@ -284,10 +293,9 @@
private MessageContext getAggregatedMessage(Aggregate aggregate) {
MessageContext newCtx = null;
- Iterator<MessageContext> itr = aggregate.getMessages().iterator();
- while (itr.hasNext()) {
- MessageContext synCtx = itr.next();
+ for (MessageContext synCtx : aggregate.getMessages()) {
+
if (newCtx == null) {
newCtx = synCtx;
@@ -299,19 +307,19 @@
try {
if (log.isDebugEnabled()) {
log.debug("Merging message : " + synCtx.getEnvelope() + " using XPath : " +
- aggregationExpression);
+ aggregationExpression);
}
EIPUtils.enrichEnvelope(
- newCtx.getEnvelope(), synCtx.getEnvelope(), aggregationExpression);
+ newCtx.getEnvelope(), synCtx.getEnvelope(), aggregationExpression);
if (log.isDebugEnabled()) {
- log.debug("Merged result : " + newCtx.getEnvelope());
+ log.debug("Merged result : " + newCtx.getEnvelope());
}
} catch (JaxenException e) {
handleException("Error merging aggregation results using XPath : " +
- aggregationExpression.toString(), e, synCtx);
+ aggregationExpression.toString(), e, synCtx);
}
}
}