if (future == null && timedOut) {
// we are timed out and no more tasks complete so break out
break;
} else if (future == null) {
// timeout occurred
AggregationStrategy strategy = getAggregationStrategy(null);
if (strategy instanceof TimeoutAwareAggregationStrategy) {
// notify the strategy we timed out
Exchange oldExchange = result.get();
if (oldExchange == null) {
// if they all timed out the result may not have been set yet, so use the original exchange
oldExchange = original;
}
((TimeoutAwareAggregationStrategy) strategy).timeout(oldExchange, aggregated, total.intValue(), timeout);
} else {
// log a WARN we timed out since it will not be aggregated and the Exchange will be lost
LOG.warn("Parallel processing timed out after " + timeout + " millis for number " + aggregated + ". This task will be cancelled and will not be aggregated.");
}
if (LOG.isDebugEnabled()) {
LOG.debug("Timeout occurred after " + timeout + " millis for number " + aggregated + " task.");
}
timedOut = true;
// mark that index as timed out, which allows us to try to retrieve
// any already completed tasks in the next loop
ExecutorServiceHelper.timeoutTask(completion);
} else {
// there is a result to aggregate
Exchange subExchange = future.get();
// Decide whether to continue with the multicast or not; similar logic to the Pipeline
Integer number = getExchangeIndex(subExchange);
boolean continueProcessing = PipelineHelper.continueProcessing(subExchange, "Parallel processing failed for number " + number, LOG);
if (stopOnException && !continueProcessing) {
// we want to stop on exception and an exception or failure occurred
// this is similar to what the pipeline does, so we should do the same to not surprise end users
// so we should set the failed exchange as the result and break out
result.set(subExchange);
stoppedOnException = true;
break;
}
// we got a result so aggregate it
AggregationStrategy strategy = getAggregationStrategy(subExchange);
doAggregate(strategy, result, subExchange);
}
aggregated++;
}