public Aggregate(String corelation,
    long timeoutMillis,
    int min,
    int max,
    AggregateMediator mediator) 
{
        
        this.correlation = corelation;
        if (timeoutMillis  > 0) {
            expiryTimeMillis = System.currentTimeMillis() + timeoutMillis;
        }
        if (min  > 0) {
            minCount = min;
        }
        if (max  > 0) {
            maxCount = max;
        }
        this.aggregateMediator = mediator;
}

Save aggregation properties and timeout

Parameters:

corelation - representing the corelation name of the messages in the aggregate

timeoutMillis - the timeout duration in milliseconds

min - the minimum number of messages to be aggregated

max - the maximum number of messages to be aggregated

mediator -

 public synchronized boolean addMessage(MessageContext synCtx) 
{
        if (maxCount < = 0 || (maxCount  > 0 && messages.size() <  maxCount)) {
            messages.add(synCtx);
            return true;
        } else {
            return false;
        }
}

Add a message to the interlan message list

 public String getCorrelation() 
{
        return correlation;
}

 public long getExpiryTimeMillis() 
{
        return expiryTimeMillis;
}

 public synchronized boolean getLock() 
{
        return !locked;
}

 public int getMaxCount() 
{
        return maxCount;
}

 public List<MessageContext> getMessages() 
{
        return messages;
}

 public int getMinCount() 
{
        return minCount;
}

 public long getTimeoutMillis() 
{
        return timeoutMillis;
}

 public synchronized boolean isComplete(boolean traceOn,
    boolean traceOrDebugOn,
    Log trace,
    Log log) 
{
        if (!completed) {
            // if any messages have been collected, check if the completion criteria is met
            if (!messages.isEmpty()) {
                // get total messages for this group, from the first message we have collected
                MessageContext mc = messages.get(0);
                Object prop = mc.getProperty(EIPConstants.MESSAGE_SEQUENCE);
                if (prop != null && prop instanceof String) {
                    String[] msgSequence = prop.toString().split(
                            EIPConstants.MESSAGE_SEQUENCE_DELEMITER);
                    int total = Integer.parseInt(msgSequence[1]);
                    if (traceOrDebugOn) {
                        traceOrDebug(traceOn, trace, log, messages.size() +
                                " messages of " + total + " collected in current aggregation");
                    }
                    if (messages.size()  >= total) {
                        if (traceOrDebugOn) {
                            traceOrDebug(traceOn, trace, log, "Aggregation complete");
                        }
                        return true;
                    }
                }
            } else {
                if (traceOrDebugOn) {
                    traceOrDebug(traceOn, trace, log, "No messages collected in current aggregation");
                }
            }
            // if the minimum number of messages has been reached, its complete
            if (minCount  > 0 && messages.size()  >= minCount) {
                if (traceOrDebugOn) {
                    traceOrDebug(traceOn, trace, log,
                            "Aggregation complete - the minimum : " + minCount
                                    + " messages has been reached");
                }
                return true;
            }
            if (maxCount  > 0 && messages.size()  >= maxCount) {
                if (traceOrDebugOn) {
                    traceOrDebug(traceOn, trace, log,
                            "Aggregation complete - the maximum : " + maxCount
                                    + " messages has been reached");
                }
                return true;
            }
            // else, has this aggregation reached its timeout?
            if (expiryTimeMillis  > 0 && System.currentTimeMillis()  >= expiryTimeMillis) {
                if (traceOrDebugOn) {
                    traceOrDebug(traceOn, trace, log,
                            "Aggregation complete - the aggregation has timed out");
                }
                return true;
            }
        } else {
            if (traceOrDebugOn) {
                traceOrDebug(traceOn, trace, log,
                        "Aggregation already completed - this message will not be processed in aggregation");
            }
        }
        return false;
}

Has this aggregation group completed?

 public boolean isCompleted() 
{
        return completed;
}

 public  void releaseLock() 
{
        locked = false;
}

 public  void run() 
{
        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;
            }
        }
}

 public  void setCompleted(boolean completed) 
{
        this.completed = completed;
}

 public  void setCorrelation(String correlation) 
{
        this.correlation = correlation;
}

 public  void setExpiryTimeMillis(long expiryTimeMillis) 
{
        this.expiryTimeMillis = expiryTimeMillis;
}

 public  void setMaxCount(int maxCount) 
{
        this.maxCount = maxCount;
}

 public  void setMessages(List<MessageContext> messages) 
{
        this.messages = messages;
}

 public  void setMinCount(int minCount) 
{
        this.minCount = minCount;
}

 public  void setTimeoutMillis(long timeoutMillis) 
{
        this.timeoutMillis = timeoutMillis;
}