public AggregateMediator() 
{
        try {
            aggregationExpression = new SynapseXPath("s11:Body/child::*[position()=1] | " +
                "s12:Body/child::*[position()=1]");
            aggregationExpression.addNamespace("s11", SOAP11Constants.SOAP_ENVELOPE_NAMESPACE_URI);
            aggregationExpression.addNamespace("s12", SOAP12Constants.SOAP_ENVELOPE_NAMESPACE_URI);
        } catch (JaxenException e) {
            if (log.isDebugEnabled()) {
                handleException("Unable to set the default " +
                    "aggregationExpression for the aggregation", e, null);
            }
        }
}

 public synchronized  void completeAggregate(Aggregate aggregate) 
{
        if (log.isDebugEnabled()) {
            log.debug("Aggregation completed or timed out");
        }
        // cancel the timer
        aggregate.cancel();
        aggregate.setCompleted(true);
        MessageContext newSynCtx = getAggregatedMessage(aggregate);
        if (newSynCtx == null) {
            log.warn("An aggregation of messages timed out with no aggregated messages", null);
            return;
        }
        activeAggregates.remove(aggregate.getCorrelation());
        if ((correlateExpression != null &&
            !correlateExpression.toString().equals(aggregate.getCorrelation())) ||
            correlateExpression == null) {
            if (onCompleteSequence != null) {
                onCompleteSequence.mediate(newSynCtx);
            } else if (onCompleteSequenceRef != null
                && newSynCtx.getSequence(onCompleteSequenceRef) != null) {
                newSynCtx.getSequence(onCompleteSequenceRef).mediate(newSynCtx);
            } else {
                handleException("Unable to find the sequence for the mediation " +
                    "of the aggregated message", newSynCtx);
            }
        }
}

Invoked by the Aggregate objects that are timed out, to signal timeout/completion of itself

 public Map getActiveAggregates() 
{
        return activeAggregates;
}

 public SynapseXPath getAggregationExpression() 
{
        return aggregationExpression;
}

 public long getCompletionTimeoutMillis() 
{
        return completionTimeoutMillis;
}

 public SynapseXPath getCorrelateExpression() 
{
        return correlateExpression;
}

 public int getMaxMessagesToComplete() 
{
        return maxMessagesToComplete;
}

 public int getMinMessagesToComplete() 
{
        return minMessagesToComplete;
}

 public SequenceMediator getOnCompleteSequence() 
{
        return onCompleteSequence;
}

 public String getOnCompleteSequenceRef() 
{
        return onCompleteSequenceRef;
}

 public boolean mediate(MessageContext synCtx) 
{
        boolean traceOn = isTraceOn(synCtx);
        boolean traceOrDebugOn = isTraceOrDebugOn(traceOn);
        if (traceOrDebugOn) {
            traceOrDebug(traceOn, "Start : Aggregate mediator");
            if (traceOn && trace.isTraceEnabled()) {
                trace.trace("Message : " + synCtx.getEnvelope());
            }
        }
        try {
            Aggregate aggregate = null;
            // if a correlateExpression is provided and there is a coresponding
            // element in the current message prepare to correlate the messages on that
            if (correlateExpression != null
                    && correlateExpression.evaluate(synCtx) != null) {
                while (aggregate == null) {
                    synchronized (lock) {
                        if (activeAggregates.containsKey(correlateExpression.toString())) {
                            aggregate = activeAggregates.get(correlateExpression.toString());
                            if (aggregate != null) {
                                if (!aggregate.getLock()) {
                                    aggregate = null;
                                }
                            }
                        } else {
                            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.getLock();
                            activeAggregates.put(correlateExpression.toString(), aggregate);
                        }
                    }
                }
            } else if (synCtx.getProperty(EIPConstants.AGGREGATE_CORRELATION) != null) {
                // if the correlattion cannot be found using the correlateExpression then
                // try the default which is through the AGGREGATE_CORRELATION message property
                // which is the unique original message id of a split or iterate operation and
                // which thus can be used to uniquely group messages into aggregates
                Object o = synCtx.getProperty(EIPConstants.AGGREGATE_CORRELATION);
                String correlation;
                if (o != null && o instanceof String) {
                    correlation = (String) o;
                    while (aggregate == null) {
                        synchronized (lock) {
                            if (activeAggregates.containsKey(correlation)) {
                                aggregate = activeAggregates.get(correlation);
                                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"));
                                }
                                aggregate = new Aggregate(
                                        correlation,
                                        completionTimeoutMillis,
                                        minMessagesToComplete,
                                        maxMessagesToComplete, this);
                                if (completionTimeoutMillis  > 0) {
                                    synCtx.getConfiguration().getSynapseTimer().
                                            schedule(aggregate, completionTimeoutMillis);
                                }
                                aggregate.getLock();
                                activeAggregates.put(correlation, aggregate);
                            }
                        }
                    }
                } else {
                    if (traceOrDebugOn) {
                        traceOrDebug(traceOn, "Unable to find aggrgation correlation property");
                    }
                    return true;
                }
            } else {
                if (traceOrDebugOn) {
                    traceOrDebug(traceOn,
                            "Unable to find aggrgation correlation XPath or property");
                }
                return true;
            }
            // if there is an aggregate continue on aggregation
            if (aggregate != null) {
                boolean collected = aggregate.addMessage(synCtx);
                if (traceOrDebugOn) {
                    if (collected) {
                        traceOrDebug(traceOn, "Collected a message during aggregation");
                        if (traceOn && trace.isTraceEnabled()) {
                            trace.trace("Collected message : " + synCtx);
                        }
                    }
                }
                // check the completeness of the aggregate and if completed aggregate the messages
                // if not completed return false and block the message sequence till it completes
                if (aggregate.isComplete(traceOn, traceOrDebugOn, trace, log)) {
                    if (traceOrDebugOn) {
                        traceOrDebug(traceOn, "Aggregation completed - invoking onComplete");
                    }
                    completeAggregate(aggregate);
                    if (traceOrDebugOn) {
                        traceOrDebug(traceOn, "End : Aggregate mediator");
                    }
                    return true;
                } else {
                    aggregate.releaseLock();
                }
            } else {
                // if the aggregation correlation cannot be found then continue the message on the
                // normal path by returning true
                if (traceOrDebugOn) {
                    traceOrDebug(traceOn, "Unable to find an aggregate for this message - skip");
                }
                return true;
            }
        } catch (JaxenException e) {
            handleException("Unable to execute the XPATH over the message", e, synCtx);
        }
        if (traceOrDebugOn) {
            traceOrDebug(traceOn, "End : Aggregate mediator");
        }
        return false;
}

Aggregate messages flowing through this mediator according to the correlation criteria and the aggregation algorithm specified to it

 public  void setAggregationExpression(SynapseXPath aggregationExpression) 
{
        this.aggregationExpression = aggregationExpression;
}

 public  void setCompletionTimeoutMillis(long completionTimeoutMillis) 
{
        this.completionTimeoutMillis = completionTimeoutMillis;
}

 public  void setCorrelateExpression(SynapseXPath correlateExpression) 
{
        this.correlateExpression = correlateExpression;
}

 public  void setMaxMessagesToComplete(int maxMessagesToComplete) 
{
        this.maxMessagesToComplete = maxMessagesToComplete;
}

 public  void setMinMessagesToComplete(int minMessagesToComplete) 
{
        this.minMessagesToComplete = minMessagesToComplete;
}

 public  void setOnCompleteSequence(SequenceMediator onCompleteSequence) 
{
        this.onCompleteSequence = onCompleteSequence;
}

 public  void setOnCompleteSequenceRef(String onCompleteSequenceRef) 
{
        this.onCompleteSequenceRef = onCompleteSequenceRef;
}