if (addMessage(aggregation, MessageUtil.copyIn(exchange), exchange)) {
if (addMessage(aggregation, in, exchange)) {
    public void process(MessageExchange exchange) throws Exception {        try {            // Skip DONE            if (exchange.getStatus() == ExchangeStatus.DONE) {                return;            // Skip ERROR            } else if (exchange.getStatus() == ExchangeStatus.ERROR) {                return;            // Handle an ACTIVE exchange as a PROVIDER            } else if (exchange.getRole() == MessageExchange.Role.PROVIDER) {                if (exchange instanceof InOnly == false &&                    exchange instanceof RobustInOnly == false) {                    fail(exchange, new UnsupportedOperationException("Use an InOnly or RobustInOnly MEP"));                } else {                    NormalizedMessage in = MessageUtil.copyIn(exchange);                    final String correlationId = getCorrelationID(exchange, in);                    if (correlationId == null || correlationId.length() == 0) {                        throw new IllegalArgumentException("Could not retrieve correlation id for incoming exchange");                    }                    // Load existing aggregation                    Lock lock = getLockManager().getLock(correlationId);                    lock.lock();                    try {                        Object aggregation = store.load(correlationId);                        Date timeout = null;                        // Create a new aggregate                        if (aggregation == null) {                            if (isAggregationClosed(correlationId)) {                                // TODO: should we return an error here ?                            } else {                                aggregation = createAggregation(correlationId);                                timeout = getTimeout(aggregation);                            }                        } else if (isRescheduleTimeouts()) {                            timeout = getTimeout(aggregation);                                                    }                        // If the aggregation is not closed                        if (aggregation != null) {                            if (addMessage(aggregation, MessageUtil.copyIn(exchange), exchange)) {                                sendAggregate(correlationId, aggregation, false);                            } else {                                store.store(correlationId, aggregation);                                if (timeout != null) {                                    if (log.isDebugEnabled()) {                                        log.debug("Scheduling timeout at " + timeout + " for aggregate " + correlationId);                                    }                                    getTimerManager().schedule(new TimerListener() {                                        public void timerExpired(Timer timer) {                                            AbstractAggregator.this.onTimeout(correlationId);                                        }                                    }, timeout);                                }                            }                        }                        done(exchange);                    } finally {                        lock.unlock();                    }                }            // Handle an ACTIVE exchange as a CONSUMER            } else if (exchange.getStatus() == ExchangeStatus.ACTIVE) {                done(exchange);            }        // If an error occurs, log it and report the error back to the sender        // if the exchange is still ACTIVE         } catch (Exception e) {            log.error("An exception occured while processing exchange", e);            if (exchange.getStatus() == ExchangeStatus.ACTIVE) {                fail(exchange, e);            }        }    }